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 setUp(self):
        self.cursor.execute("ROLLBACK")
        self.cursor.execute("BEGIN")

    def test_all_direct_and_indirect_subordinates(self):
        """Returns all direct and indirect subordinates of the specified manager (employee_id = 2)."""
        with open("query.sql", "r") as file:
            query = file.read()

        # Expected subordinates for Bob Smith (employee_id = 2):
        # David Brown (4), Eve Black (5), Frank Green (6), Grace Lee (7)
        # Levels: David=1, Eve=2, Frank/Grace=3
        expected = [
            (4, "David Brown", "Engineering Manager", 1),
            (5, "Eve Black", "Lead Developer", 2),
            (6, "Frank Green", "Developer", 3),
            (7, "Grace Lee", "Developer", 3),
        ]
        expected_df = pd.DataFrame(expected, columns=["employee_id", "name", "position", "reporting_level"])

        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)

        # Check each expected row is present
        missing_rows = []
        for _, expected_row in expected_df.iterrows():
            mask = (
                (result_df["employee_id"] == expected_row["employee_id"]) &
                (result_df["name"] == expected_row["name"]) &
                (result_df["position"] == expected_row["position"]) &
                (result_df["reporting_level"] == expected_row["reporting_level"])
            )
            if not mask.any():
                missing_rows.append(dict(expected_row))

        condition = len(missing_rows) == 0
        failure_msg = "Missing expected subordinates: " + ", ".join(str(r) for r in missing_rows) if missing_rows else ""
        _dynamic_test(
            self,
            condition,
            "All direct and indirect subordinates are present.",
            failure_msg
        )

    def test_reporting_levels_correct(self):
        """Each subordinate has the correct reporting level."""
        with open("query.sql", "r") as file:
            query = file.read()

        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)

        # Map employee_id to correct reporting_level
        correct_levels = {
            4: 1,  # David Brown
            5: 2,  # Eve Black
            6: 3,  # Frank Green
            7: 3,  # Grace Lee
        }
        incorrect = []
        for eid, level in correct_levels.items():
            found = result_df[result_df["employee_id"] == eid]
            if found.empty or found.iloc[0]["reporting_level"] != level:
                incorrect.append((eid, level))
        condition = len(incorrect) == 0
        failure_msg = "Incorrect reporting levels for: " + ", ".join(str(r) for r in incorrect) if incorrect else ""
        _dynamic_test(
            self,
            condition,
            "All reporting levels are correct.",
            failure_msg
        )

    def test_only_subordinates_included(self):
        """Only employees who report (directly or indirectly) to the specified manager are included."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        # Bob Smith's subordinates are 4,5,6,7
        allowed_ids = {4,5,6,7}
        extra = result_df[~result_df["employee_id"].isin(allowed_ids)]
        condition = extra.empty
        failure_msg = f"Found employees not reporting to the specified manager: {extra['employee_id'].tolist()}" if not extra.empty else ""
        _dynamic_test(
            self,
            condition,
            "No unrelated employees included.",
            failure_msg
        )

    def test_output_columns(self):
        """Output columns include employee_id, name, position, and reporting_level."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required = {"employee_id", "name", "position", "reporting_level"}
        condition = required.issubset(set(columns))
        failure_msg = f"Missing columns: {required - set(columns)}" if not condition else ""
        _dynamic_test(
            self,
            condition,
            "All required output columns are present.",
            failure_msg
        )

    def test_ordering(self):
        """Results are ordered by reporting_level and employee_id."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        if not result_df.empty:
            sorted_df = result_df.sort_values(["reporting_level", "employee_id"]).reset_index(drop=True)
            condition = result_df.reset_index(drop=True).equals(sorted_df)
        else:
            condition = True
        failure_msg = "Results are not ordered by reporting_level and employee_id."
        _dynamic_test(
            self,
            condition,
            "Results are ordered correctly.",
            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: Organizational Hierarchy Traversal

Рішення