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_all_electronics_categories_included(self):
        """Check all categories under 'Electronics' (including itself) are present."""
        with open("query.sql", "r") as file:
            query = file.read()

        # Expected category_ids under Electronics tree: 1,2,3,4,5,6,7,8
        expected_ids = set([1,2,3,4,5,6,7,8])
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_ids = set(row[0] for row in rows)
        missing = expected_ids - result_ids
        extra = result_ids - expected_ids
        condition = not missing and not extra
        failure_msg = ""
        if missing:
            failure_msg += f"Missing expected category_id(s): {sorted(list(missing))}. "
        if extra:
            failure_msg += f"Unexpected category_id(s): {sorted(list(extra))}."
        _dynamic_test(
            self,
            condition,
            "All expected categories under 'Electronics' are included, and no extra categories are present.",
            failure_msg
        )

    def test_columns_present(self):
        """Check that all required columns are present in the output."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required = {'category_id', 'category_name', 'parent_category_id', 'depth'}
        missing = required - set(columns)
        _dynamic_test(
            self,
            not missing,
            "All required columns are present.",
            f"Missing columns: {', '.join(missing)}"
        )

    def test_electronics_root_depth(self):
        """Check that 'Electronics' row has depth 1."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        row = df[df['category_name'] == 'Electronics']
        condition = not row.empty and row.iloc[0]['depth'] == 1
        _dynamic_test(
            self,
            condition,
            "'Electronics' row has depth 1.",
            "'Electronics' row missing or does not have depth 1."
        )

    def test_direct_children_depth(self):
        """Check that direct children of 'Electronics' have depth 2."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        # Direct children: Computers (2), Smartphones (5), Accessories (6)
        for name in ['Computers', 'Smartphones', 'Accessories']:
            row = df[df['category_name'] == name]
            condition = not row.empty and row.iloc[0]['depth'] == 2
            _dynamic_test(
                self,
                condition,
                f"Direct child '{name}' has depth 2.",
                f"Direct child '{name}' missing or does not have depth 2."
            )

    def test_correct_depth_for_all(self):
        """Check that all included categories have correct depth."""
        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]
        df = pd.DataFrame(rows, columns=columns)
        # Expected: (category_name, depth)
        expected_depths = {
            'Electronics': 1,
            'Computers': 2,
            'Laptops': 3,
            'Desktops': 3,
            'Smartphones': 2,
            'Accessories': 2,
            'Chargers': 3,
            'Cables': 3
        }
        failures = []
        for name, depth in expected_depths.items():
            row = df[df['category_name'] == name]
            if row.empty or int(row.iloc[0]['depth']) != depth:
                failures.append(f"{name} should have depth {depth}")
        _dynamic_test(
            self,
            not failures,
            "All included categories have the correct depth.",
            "; ".join(failures)
        )

    def test_no_non_electronics_categories(self):
        """Check that no categories outside the 'Electronics' tree 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]
        df = pd.DataFrame(rows, columns=columns)
        not_electronics = set(df['category_name']) - set([
            'Electronics','Computers','Laptops','Desktops','Smartphones','Accessories','Chargers','Cables'])
        _dynamic_test(
            self,
            not not_electronics,
            "No categories outside 'Electronics' subtree are included.",
            f"Non-Electronics categories present: {', '.join(not_electronics)}"
        )

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: Product Category Tree Expansion

解答