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_grouping_by_month_and_customer(self):
        """Check if results are grouped by month and customer_id."""
        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)
        unique_keys = df[['month', 'customer_id']].drop_duplicates()
        condition = len(unique_keys) == len(df)
        _dynamic_test(
            self,
            condition,
            "Results are grouped by month and customer_id.",
            "Each row should be unique per month and customer_id."
        )

    def test_total_amount_and_transaction_count(self):
        """Check if total_amount and transaction_count are calculated per group."""
        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)
        for _, row in df.iterrows():
            self.cursor.execute(
                """
                SELECT SUM(amount), COUNT(*) FROM transactions
                WHERE customer_id = %s AND TO_CHAR(transaction_date, 'YYYY-MM') = %s
                """,
                (row['customer_id'], row['month'])
            )
            expected_sum, expected_count = self.cursor.fetchone()
            condition = (float(row['total_amount']) == float(expected_sum)) and (int(row['transaction_count']) == int(expected_count))
            _dynamic_test(
                self,
                condition,
                f"Correct total_amount and transaction_count for customer {row['customer_id']} in {row['month']}.",
                f"Incorrect totals for customer {row['customer_id']} in {row['month']}."
            )

    def test_vip_labeling(self):
        """Check if vip_status is 'VIP' only when criteria are met."""
        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)
        for _, row in df.iterrows():
            amount = float(row['total_amount'])
            count = int(row['transaction_count'])
            vip_status = row['vip_status']
            condition = (amount >= 500 and count >= 2 and vip_status == 'VIP')
            _dynamic_test(
                self,
                condition,
                f"VIP label correctly assigned for customer {row['customer_id']} in {row['month']}.",
                f"VIP label should only be present if total_amount >= 500 and transaction_count >= 2."
            )

    def test_only_vip_customers_included(self):
        """Check that only VIP customers are returned in the result."""
        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)
        condition = all((float(r['total_amount']) >= 500 and int(r['transaction_count']) >= 2) for _, r in df.iterrows())
        _dynamic_test(
            self,
            condition,
            "Only VIP customers are included in the result.",
            "Result contains customers who do not meet VIP criteria."
        )

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 the CASE WHEN statement in SQL, from basic conditional logic to advanced use cases in data analysis and reporting. Through engaging theory and real-world tasks, you'll learn to write expressive, powerful queries that adapt to complex business requirements.

Learn the basics of the CASE WHEN statement, its syntax, and how to use it for simple conditional logic in SQL queries.

Delve deeper into complex CASE WHEN expressions, including nested conditions and multiple criteria.

Apply CASE WHEN in WHERE and ORDER BY clauses to filter and sort data conditionally.

Combine CASE WHEN with aggregate functions for advanced reporting and analytics.

Challenge: Monthly VIP Customer Report

解答