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

SQLにおけるCASE WHEN文を基礎的な条件分岐からデータ分析・レポーティングでの高度な活用例まで習得できる包括的なモジュールです。理論と実践的な課題を通じて、複雑なビジネス要件に対応する表現力豊かなクエリの作成方法を学びます。

CASE WHEN文の基本、構文、およびSQLクエリでの単純な条件付きロジックへの活用方法を学習します。

チャレンジ：Customer Spending Analysis

解答