import unittest
import psycopg2
import pandas as pd
import re

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_uses_cte(self):
        """Check if the query uses at least one CTE (WITH clause)."""
        with open("query.sql", "r") as file:
            query = file.read()
        condition = 'with' in normalize_text(query)
        _dynamic_test(
            self,
            condition,
            "Query uses at least one CTE.",
            "Query does not use any CTE (WITH clause)."
        )

    def test_joins_all_tables(self):
        """Check if the query joins sales_transactions, customers, and products."""
        with open("query.sql", "r") as file:
            query = file.read()
        norm = normalize_text(query)
        condition = all(x in norm for x in ['sales_transactions', 'customers', 'products'])
        _dynamic_test(
            self,
            condition,
            "Query joins all required tables.",
            "Query does not join all required tables (sales_transactions, customers, products)."
        )

    def test_output_columns(self):
        """Check if output columns are correct."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected_cols = [
            'transaction_id',
            'transaction_date',
            'first_name',
            'last_name',
            'city',
            'state',
            'product_name',
            'category',
            'price',
            'quantity',
            'total_amount'
        ]
        condition = set(expected_cols).issubset(set(columns))
        missing = set(expected_cols) - set(columns)
        _dynamic_test(
            self,
            condition,
            "All required columns are present in output.",
            f"The following columns are missing in output: {', '.join(missing)}"
        )

    def test_column_selection_optimization(self):
        """Check if at least one CTE selects only needed columns (optimization)."""
        with open("query.sql", "r") as file:
            query = file.read()
        norm = normalize_text(query)
        # Look for column lists in CTEs (avoid SELECT *)
        has_col_selection = (
            'select *' not in norm
            and (
                'with' in norm and (
                    'select st.transaction_id' in norm
                    or 'select c.customer_id' in norm
                    or 'select p.product_id' in norm
                )
            )
        )
        _dynamic_test(
            self,
            has_col_selection,
            "CTE uses column selection as optimization technique.",
            "No column selection optimization detected in CTEs."
        )

    def test_ordering(self):
        """Check if results are ordered by transaction_date, last_name, first_name."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        if not rows:
            _dynamic_test(
                self,
                False,
                "Results are present.",
                "No rows returned to check ordering."
            )
            return
        df = pd.DataFrame(rows, columns=[desc[0] for desc in self.cursor.description])
        sorted_df = df.sort_values(by=["transaction_date", "last_name", "first_name"])
        condition = df.reset_index(drop=True).equals(sorted_df.reset_index(drop=True))
        _dynamic_test(
            self,
            condition,
            "Results are ordered correctly.",
            "Results are not ordered by transaction_date, last_name, first_name."
        )

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: Optimizing Sales Report Queries

Lösung