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_one_row_per_month(self):
        """Check that the query returns one row per month present in the table."""
        with open("query.sql", "r") as file:
            query = file.read()

        self.cursor.execute("SELECT DISTINCT TO_CHAR(transaction_date, 'YYYY-MM') FROM transactions")
        expected_months = set(row[0] for row in self.cursor.fetchall())

        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        months = set(row[0] for row in rows)

        condition = months == expected_months and len(rows) == len(expected_months)
        _dynamic_test(
            self,
            condition,
            "Returned one row per month present in the table.",
            f"Returned months: {months}, but expected: {expected_months}."
        )

    def test_purchase_count_column(self):
        """Check that purchase_count column shows correct count of 'purchase' transactions per month."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)

        self.cursor.execute("""
            SELECT TO_CHAR(transaction_date, 'YYYY-MM') AS month, COUNT(*) AS purchase_count
            FROM transactions
            WHERE transaction_type = 'purchase'
            GROUP BY TO_CHAR(transaction_date, 'YYYY-MM')
        """)
        expected = dict(self.cursor.fetchall())

        for idx, row in result_df.iterrows():
            m = row['month']
            actual = row['purchase_count']
            expected_count = expected.get(m, 0)
            condition = actual == expected_count
            _dynamic_test(
                self,
                condition,
                f"Correct purchase_count for month {m}.",
                f"Incorrect purchase_count for month {m}: got {actual}, expected {expected_count}."
            )

    def test_refund_count_column(self):
        """Check that refund_count column shows correct count of 'refund' transactions per month."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)

        self.cursor.execute("""
            SELECT TO_CHAR(transaction_date, 'YYYY-MM') AS month, COUNT(*) AS refund_count
            FROM transactions
            WHERE transaction_type = 'refund'
            GROUP BY TO_CHAR(transaction_date, 'YYYY-MM')
        """)
        expected = dict(self.cursor.fetchall())

        for idx, row in result_df.iterrows():
            m = row['month']
            actual = row['refund_count']
            expected_count = expected.get(m, 0)
            condition = actual == expected_count
            _dynamic_test(
                self,
                condition,
                f"Correct refund_count for month {m}.",
                f"Incorrect refund_count for month {m}: got {actual}, expected {expected_count}."
            )

    def test_fee_count_column(self):
        """Check that fee_count column shows correct count of 'fee' transactions per month."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)

        self.cursor.execute("""
            SELECT TO_CHAR(transaction_date, 'YYYY-MM') AS month, COUNT(*) AS fee_count
            FROM transactions
            WHERE transaction_type = 'fee'
            GROUP BY TO_CHAR(transaction_date, 'YYYY-MM')
        """)
        expected = dict(self.cursor.fetchall())

        for idx, row in result_df.iterrows():
            m = row['month']
            actual = row['fee_count']
            expected_count = expected.get(m, 0)
            condition = actual == expected_count
            _dynamic_test(
                self,
                condition,
                f"Correct fee_count for month {m}.",
                f"Incorrect fee_count for month {m}: got {actual}, expected {expected_count}."
            )

    def test_month_format(self):
        """Check that the month column is in 'YYYY-MM' format and represents a valid year and month."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        months = [str(row[0]) for row in rows]
        pattern = r"^\d{4}-\d{2}$"
        all_valid = all(re.match(pattern, m) for m in months)
        _dynamic_test(
            self,
            all_valid,
            "Month column is in 'YYYY-MM' format.",
            f"Month column not in correct format: {months}."
        )

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: Transaction Type Breakdown

Lösung