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_cte_is_used(self):
        """Check if a CTE is used in the query."""
        with open("query.sql", "r") as file:
            query = file.read()
        condition = normalize_text(query).startswith("with ")
        _dynamic_test(
            self,
            condition,
            "A CTE is used at the beginning of the query.",
            "The query must start with a CTE using 'WITH'."
        )

    def test_customers_sales_transactions_join(self):
        """Check if customers and sales_transactions are joined on customer_id in the CTE."""
        with open("query.sql", "r") as file:
            query = normalize_text(file.read())
        join_present = (
            "from customers" in query and
            "join sales_transactions" in query and
            "on c.customer_id = st.customer_id" in query
        )
        _dynamic_test(
            self,
            join_present,
            "customers and sales_transactions are joined on customer_id.",
            "The CTE must join customers and sales_transactions on customer_id."
        )

    def test_group_by_customer(self):
        """Check if the CTE groups by customer fields."""
        with open("query.sql", "r") as file:
            query = normalize_text(file.read())
        group_by_present = "group by c.customer_id" in query
        _dynamic_test(
            self,
            group_by_present,
            "Grouping by customer_id is present in the CTE.",
            "The CTE must group by customer_id."
        )

    def test_sum_total_amount(self):
        """Check if SUM is used to aggregate total_amount in the CTE."""
        with open("query.sql", "r") as file:
            query = normalize_text(file.read())
        sum_present = "sum(st.total_amount)" in query
        _dynamic_test(
            self,
            sum_present,
            "SUM(st.total_amount) is used in the CTE.",
            "The CTE must use SUM(st.total_amount) to aggregate purchases."
        )

    def test_where_total_spent_gt_5000(self):
        """Check if the main query filters for total_spent > 5000."""
        with open("query.sql", "r") as file:
            query = normalize_text(file.read())
        where_present = (
            ("where total_spent > 5000" in query) or
            ("> 5000" in query and "where" in query)
        )
        _dynamic_test(
            self,
            where_present,
            "The query filters for total_spent > 5000.",
            "The main query must filter for total_spent > 5000."
        )

    def test_output_columns(self):
        """Check if output columns are as required."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required = {"customer_id", "first_name", "last_name", "total_spent"}
        condition = required.issubset(set([c.lower() for c in columns]))
        _dynamic_test(
            self,
            condition,
            "Output includes customer_id, first_name, last_name, total_spent.",
            "Output must include customer_id, first_name, last_name, total_spent."
        )

    def test_no_results_for_current_data(self):
        """Check that no customer is returned based on current data (since max total is less than 5000)."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        condition = len(rows) == 0
        _dynamic_test(
            self,
            condition,
            "No customers returned as expected for current data.",
            "Query should return no rows for current data set."
        )

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: Customer Purchase Aggregation

解答