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()

    @classmethod
    def tearDownClass(cls):
        cls.cursor.close()
        cls.connection.close()

    def test_query_returns_top_3_products(self):
        """Check that only top 3 products by total sales are returned."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == 3,
            "Returned exactly 3 products.",
            f"Returned {len(rows)} products instead of 3."
        )

    def test_result_has_product_name_and_total_sales(self):
        """Check that result includes product_name and total_sales 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_columns = {"product_name", "total_sales"}
        _dynamic_test(
            self,
            required_columns.issubset(set(columns)),
            "Result includes product_name and total_sales columns.",
            f"Missing one or both required columns: {required_columns - set(columns)}."
        )

    def test_total_sales_calculation(self):
        """Check that total_sales values are correct for the top 3 products."""
        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)

        # Calculate correct top 3 products by total sales
        self.cursor.execute("""
            SELECT 
                p.product_name, 
                SUM(s.total_amount) AS total_sales
            FROM sales s
            JOIN products p ON s.product_id = p.product_id
            GROUP BY p.product_name
            ORDER BY total_sales DESC
            LIMIT 3
        """)
        sol_rows = self.cursor.fetchall()
        sol_columns = [desc[0] for desc in self.cursor.description]
        sol_df = pd.DataFrame(sol_rows, columns=sol_columns)

        # Compare sets of (product_name, total_sales)
        user_set = set(tuple(row) for row in result_df.itertuples(index=False, name=None))
        sol_set = set(tuple(row) for row in sol_df.itertuples(index=False, name=None))

        _dynamic_test(
            self,
            user_set == sol_set,
            "Top 3 products and their total sales match expected values.",
            f"Expected: {sol_set}\\nGot: {user_set}"
        )

    def test_uses_subquery_in_from(self):
        """Check that a subquery is used in the FROM section."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Normalize and check for (SELECT ... ) in FROM
        query_norm = normalize_text(query)
        has_subquery_from = False
        lines = [line.strip() for line in query_norm.splitlines()]
        for i, line in enumerate(lines):
            if "from" in line:
                # Look for (select ... ) after FROM
                rest = " ".join(lines[i:])
                if "(select" in rest.replace(' ',''):
                    has_subquery_from = True
                    break
        _dynamic_test(
            self,
            has_subquery_from,
            "Query uses a subquery in the FROM section.",
            "Query does not use a subquery in the FROM section."
        )

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

Master the art of writing and optimizing nested subqueries in SQL. This course explores advanced querying techniques using subqueries in WHERE, FROM, and SELECT clauses, and teaches best practices for performance optimization. Each section blends engaging theory with real-world challenges to solidify your understanding.

Explore how to use nested subqueries within the WHERE clause to filter data based on complex conditions and relationships.

Learn how to use nested subqueries in the FROM clause to create derived tables for complex data analysis.

Discover how to use nested subqueries in the SELECT clause to compute values for each row in your result set.

Master best practices and techniques for optimizing nested subqueries for performance and maintainability.

Challenge: Top-Selling Products

解答