import unittest
import psycopg2
import pandas as pd

class TestPostgresQuery(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.connection = psycopg2.connect(
            host="localhost",
            user="agent",
            password="qwerty",
            dbname="testdb",
            sslmode="disable"
        )
        cls.cursor = cls.connection.cursor()

    def test_returns_all_direct_and_indirect_dependencies(self):
        """Check that all direct and indirect dependencies are returned for task_id = 6 (Deployment)."""
        with open("query.sql", "r") as file:
            query = file.read()

        # Expected: Deployment (6) depends on Testing (5), which depends on Frontend (3) and Backend (4),
        # Frontend (3) depends on Design Mockups (2), which depends on Requirements Gathering (1)
        # Backend (4) depends on Design Mockups (2) as well
        expected = set([
            (1, 'Requirements Gathering'),
            (2, 'Design Mockups'),
            (3, 'Frontend Development'),
            (4, 'Backend Development'),
            (5, 'Testing')
        ])
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result = set((row[0], row[1]) for row in rows)
        missing = expected - result
        extra = result - expected
        condition = len(missing) == 0 and len(extra) == 0
        failure_msg = "Missing: {}. Extra: {}".format(missing, extra) if not condition else ""
        _dynamic_test(self, condition, "All direct and indirect dependencies are returned for task_id = 6.", failure_msg)

    def test_does_not_include_specified_task(self):
        """Check that the specified task_id itself is not included in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        ids = [row[0] for row in rows]
        condition = 6 not in ids
        _dynamic_test(self, condition, "Specified task_id is not included in the results.", "Specified task_id=6 was included in the results.")

    def test_returns_task_id_and_task_name(self):
        """Check that both task_id and task_name are returned in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        condition = ('prereq_task_id' in columns or 'task_id' in columns) and ('prereq_task_name' in columns or 'task_name' in columns)
        _dynamic_test(self, condition, "Result includes both task_id and task_name columns.", "Result must include both task_id and task_name columns.")

    def test_multiple_dependencies(self):
        """Check that a task with multiple dependencies at any level returns all prerequisites (task_id = 5)."""
        # For task_id = 5 (Testing): depends on 3 (Frontend Development) and 4 (Backend Development)
        # 3 depends on 2, 2 depends on 1; 4 depends on 2
        query = """
        WITH RECURSIVE task_prereqs AS (
            SELECT td.depends_on_task_id AS prereq_task_id, pt.task_name AS prereq_task_name
            FROM task_dependencies td JOIN project_tasks pt ON td.depends_on_task_id = pt.task_id
            WHERE td.task_id = 5
            UNION
            SELECT td.depends_on_task_id AS prereq_task_id, pt.task_name AS prereq_task_name
            FROM task_dependencies td JOIN project_tasks pt ON td.depends_on_task_id = pt.task_id
            JOIN task_prereqs tp ON td.task_id = tp.prereq_task_id
        )
        SELECT prereq_task_id, prereq_task_name FROM task_prereqs ORDER BY prereq_task_id;
        """
        expected = set([
            (1, 'Requirements Gathering'),
            (2, 'Design Mockups'),
            (3, 'Frontend Development'),
            (4, 'Backend Development')
        ])
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result = set((row[0], row[1]) for row in rows)
        missing = expected - result
        extra = result - expected
        condition = len(missing) == 0 and len(extra) == 0
        failure_msg = "Missing: {}. Extra: {}".format(missing, extra) if not condition else ""
        _dynamic_test(self, condition, "All prerequisites for a task with multiple dependencies are returned.", failure_msg)

    def test_no_dependencies_returns_empty(self):
        """Check that a task with no dependencies returns zero rows (task_id = 1)."""
        query = """
        WITH RECURSIVE task_prereqs AS (
            SELECT td.depends_on_task_id AS prereq_task_id, pt.task_name AS prereq_task_name
            FROM task_dependencies td JOIN project_tasks pt ON td.depends_on_task_id = pt.task_id
            WHERE td.task_id = 1
            UNION
            SELECT td.depends_on_task_id AS prereq_task_id, pt.task_name AS prereq_task_name
            FROM task_dependencies td JOIN project_tasks pt ON td.depends_on_task_id = pt.task_id
            JOIN task_prereqs tp ON td.task_id = tp.prereq_task_id
        )
        SELECT prereq_task_id, prereq_task_name FROM task_prereqs;
        """
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        condition = len(rows) == 0
        _dynamic_test(self, condition, "Task with no dependencies returns zero rows.", "Task with no dependencies did not return zero rows.")

    def test_indirect_dependencies_for_other_project(self):
        """Check indirect dependencies for a task in the second project (task_id = 12)."""
        # 12 depends on 11, which depends on 9, which depends on 8, which depends on 7
        query = """
        WITH RECURSIVE task_prereqs AS (
            SELECT td.depends_on_task_id AS prereq_task_id, pt.task_name AS prereq_task_name
            FROM task_dependencies td JOIN project_tasks pt ON td.depends_on_task_id = pt.task_id
            WHERE td.task_id = 12
            UNION
            SELECT td.depends_on_task_id AS prereq_task_id, pt.task_name AS prereq_task_name
            FROM task_dependencies td JOIN project_tasks pt ON td.depends_on_task_id = pt.task_id
            JOIN task_prereqs tp ON td.task_id = tp.prereq_task_id
        )
        SELECT prereq_task_id, prereq_task_name FROM task_prereqs ORDER BY prereq_task_id;
        """
        expected = set([
            (7, 'Market Research'),
            (8, 'Product Design'),
            (9, 'Prototype Development'),
            (11, 'Marketing Campaign')
        ])
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result = set((row[0], row[1]) for row in rows)
        missing = expected - result
        extra = result - expected
        condition = len(missing) == 0 and len(extra) == 0
        failure_msg = "Missing: {}. Extra: {}".format(missing, extra) if not condition else ""
        _dynamic_test(self, condition, "All indirect dependencies for task_id = 12 are returned.", failure_msg)


def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

if __name__ == "__main__":
    unittest.main()


test_solution.py

A comprehensive course designed to help you master Common Table Expressions (CTEs) in SQL, from foundational concepts to advanced real-world applications. Explore recursive queries, modular query design, and performance considerations through engaging theory and practical, scenario-based tasks.

Explore the basics of Common Table Expressions, their syntax, and how they can simplify SQL queries. Practice using CTEs in real-world scenarios such as reporting and data transformation.

Delve into recursive CTEs for handling hierarchical data, such as organizational charts and bill of materials. Practice building recursive queries to solve real-world problems.

Master advanced CTE techniques, including combining CTEs with aggregates, optimizing performance, and using CTEs for complex reporting.

Challenge: Project Task Dependency Resolution

Рішення