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_unique_customer_ids(self):
        """Check that each customer_id appears only once in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        customer_ids = [row[0] for row in rows]
        condition = len(customer_ids) == len(set(customer_ids))
        _dynamic_test(
            self,
            condition,
            "Each customer_id appears only once.",
            "Some customer_id values appear more than once in the results."
        )

    def test_all_customers_present(self):
        """Check that all unique customer_ids from transactions are present in the result."""
        self.cursor.execute("SELECT DISTINCT customer_id FROM transactions;")
        all_customers = set(row[0] for row in self.cursor.fetchall())
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_customers = set(row[0] for row in rows)
        condition = all_customers == result_customers
        _dynamic_test(
            self,
            condition,
            "All unique customer_ids from transactions are present.",
            f"Missing customer_ids: {all_customers - result_customers}"
        )

    def test_purchase_sum(self):
        """Check that total_purchase is correct for each customer."""
        self.cursor.execute("""
            SELECT customer_id, COALESCE(SUM(CASE WHEN transaction_type = 'purchase' THEN amount ELSE 0 END),0) AS total_purchase
            FROM transactions
            GROUP BY customer_id
        """)
        expected = dict(self.cursor.fetchall())
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        for row in rows:
            customer_id, total_purchase, _ = row
            expected_purchase = float(expected.get(customer_id, 0))
            condition = abs(float(total_purchase) - expected_purchase) < 0.01
            _dynamic_test(
                self,
                condition,
                f"Correct total_purchase for customer {customer_id}.",
                f"Incorrect total_purchase for customer {customer_id}: expected {expected_purchase}, got {total_purchase}."
            )

    def test_refund_sum(self):
        """Check that total_refund is correct for each customer (should be 0 for all in this dataset)."""
        self.cursor.execute("""
            SELECT customer_id, COALESCE(SUM(CASE WHEN transaction_type = 'refund' THEN amount ELSE 0 END),0) AS total_refund
            FROM transactions
            GROUP BY customer_id
        """)
        expected = dict(self.cursor.fetchall())
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        for row in rows:
            customer_id, _, total_refund = row
            expected_refund = float(expected.get(customer_id, 0))
            condition = abs(float(total_refund) - expected_refund) < 0.01
            _dynamic_test(
                self,
                condition,
                f"Correct total_refund for customer {customer_id}.",
                f"Incorrect total_refund for customer {customer_id}: expected {expected_refund}, got {total_refund}."
            )

    def test_zero_for_missing_types(self):
        """Check that customers with no purchase or no refund transactions have 0 in the respective column."""
        self.cursor.execute("SELECT DISTINCT customer_id FROM transactions;")
        all_customers = set(row[0] for row in self.cursor.fetchall())
        self.cursor.execute("""
            SELECT customer_id, SUM(CASE WHEN transaction_type = 'purchase' THEN amount ELSE 0 END) AS total_purchase,
                   SUM(CASE WHEN transaction_type = 'refund' THEN amount ELSE 0 END) AS total_refund
            FROM transactions
            GROUP BY customer_id
        """)
        expected = {row[0]: (row[1], row[2]) for row in self.cursor.fetchall()}
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        for row in rows:
            customer_id, total_purchase, total_refund = row
            exp_purchase, exp_refund = expected.get(customer_id, (0, 0))
            condition = (float(total_purchase) == float(exp_purchase)) and (float(total_refund) == float(exp_refund))
            _dynamic_test(
                self,
                condition,
                f"Correct 0 values for missing purchase/refund types for customer {customer_id}.",
                f"Incorrect 0 values for missing purchase/refund types for customer {customer_id}: expected ({exp_purchase}, {exp_refund}), got ({total_purchase}, {total_refund})."
            )


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: Customer Spending Analysis

Рішення

Challenge: Customer Spending Analysis

Рішення