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_covers_required_columns(self):
        """Check if output includes product_id, product_name, total_sales, average_inventory columns."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required = ["product_id", "product_name", "total_sales", "average_inventory"]
        missing = [col for col in required if col not in columns]
        _dynamic_test(
            self,
            len(missing) == 0,
            "All required columns are present in the output.",
            f"Missing columns: {missing}"
        )

    def test_uses_cte(self):
        """Check if query uses a CTE (WITH clause)."""
        with open("query.sql", "r") as file:
            query = file.read().lower()
        _dynamic_test(
            self,
            "with" in query,
            "Query uses a CTE.",
            "Query does not use a CTE (WITH clause)."
        )

    def test_above_average_sales_only(self):
        """Check that only products with above-average sales are returned."""
        # Compute total sales per product
        solution_query = '''
            WITH product_sales_inventory AS (
                SELECT
                    p.product_id,
                    p.product_name,
                    SUM(st.quantity) AS total_sales,
                    AVG(p.stock_quantity) AS average_inventory
                FROM
                    products p
                LEFT JOIN
                    sales_transactions st ON p.product_id = st.product_id
                GROUP BY
                    p.product_id, p.product_name
            ),
            average_sales AS (
                SELECT AVG(total_sales) AS avg_sales
                FROM product_sales_inventory
            )
            SELECT
                psi.product_id,
                psi.product_name,
                psi.total_sales,
                psi.average_inventory
            FROM
                product_sales_inventory psi,
                average_sales a
            WHERE
                psi.total_sales > a.avg_sales;
        '''
        self.cursor.execute(solution_query)
        expected_rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        expected_df = pd.DataFrame(expected_rows, columns=columns)

        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)

        # Check that all result rows are in expected and no extra rows
        missing = []
        for _, row in result_df.iterrows():
            if not ((expected_df["product_id"] == row["product_id"]) & (expected_df["product_name"] == row["product_name"]) & (expected_df["total_sales"] == row["total_sales"]) & (expected_df["average_inventory"] == row["average_inventory"]) ).any():
                missing.append(row.to_dict())
        extra = []
        for _, row in expected_df.iterrows():
            if not ((result_df["product_id"] == row["product_id"]) & (result_df["product_name"] == row["product_name"]) & (result_df["total_sales"] == row["total_sales"]) & (result_df["average_inventory"] == row["average_inventory"]) ).any():
                extra.append(row.to_dict())
        condition = (len(missing) == 0) and (len(extra) == 0)
        fail_msg = ""
        if len(missing) > 0:
            fail_msg += f"Rows in your result not expected: {missing}\\n"
        if len(extra) > 0:
            fail_msg += f"Expected rows missing from your result: {extra}\\n"
        _dynamic_test(
            self,
            condition,
            "Only products with above-average total sales are returned and all expected rows are present.",
            fail_msg
        )

    def test_average_sales_calculation(self):
        """Check if the query calculates average total sales and uses it as a threshold."""
        # Calculate average total sales from the data
        self.cursor.execute('''
            SELECT p.product_id, SUM(st.quantity) AS total_sales
            FROM products p
            LEFT JOIN sales_transactions st ON p.product_id = st.product_id
            GROUP BY p.product_id
        ''')
        sales = self.cursor.fetchall()
        total_sales_list = [r[1] if r[1] is not None else 0 for r in sales]
        avg_sales = sum(total_sales_list) / len(total_sales_list)

        # Now run user query and check all total_sales > avg_sales
        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 len(result_df) > 0:
            all_above = all([row["total_sales"] > avg_sales for _, row in result_df.iterrows()])
        else:
            all_above = True
        _dynamic_test(
            self,
            all_above,
            "All returned products have total sales above the average.",
            "Some products in the result do not have total sales above the average."
        )

    def test_grouping_and_aggregation(self):
        """Check that the query aggregates total sales and average inventory per product."""
        with open("query.sql", "r") as file:
            query = file.read().lower()
        has_sum = "sum(" in query
        has_avg = "avg(" in query
        _dynamic_test(
            self,
            has_sum and has_avg,
            "Query uses SUM and AVG aggregate functions.",
            "Query does not use both SUM and AVG aggregate functions."
        )

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 Inventory Analysis

Soluzione