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_all_customers_present(self):
        """Check that every customer_id in transactions appears in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT DISTINCT customer_id FROM transactions")
        expected_customers = set(row[0] for row in self.cursor.fetchall())
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_customers = set(row[0] for row in rows)
        condition = expected_customers == result_customers
        _dynamic_test(
            self,
            condition,
            "All customer_id values are present in the result.",
            f"Missing or extra customer_id values. Expected: {expected_customers}, Got: {result_customers}"
        )

    def test_high_value_transaction_count_correct(self):
        """Check that high_value_transaction_count is correct for each customer."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Get expected result using user solution
        self.cursor.execute(query)
        user_rows = self.cursor.fetchall()
        user_result = {row[0]: row[1] for row in user_rows}
        # Build expected result
        self.cursor.execute("SELECT customer_id, SUM(CASE WHEN amount >= 1000 THEN 1 ELSE 0 END) AS high_value_transaction_count FROM transactions GROUP BY customer_id;")
        expected_rows = self.cursor.fetchall()
        expected_result = {row[0]: row[1] for row in expected_rows}
        condition = user_result == expected_result
        _dynamic_test(
            self,
            condition,
            "high_value_transaction_count is correct for all customers.",
            f"Expected: {expected_result}, Got: {user_result}"
        )

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: High-Value Transaction Count

解答