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_returns_all_rows(self):
        """Check that all rows from sales table are returned."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT COUNT(*) FROM sales;")
        expected_count = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        condition = len(rows) == expected_count
        _dynamic_test(
            self,
            condition,
            "Query returns all rows from sales table.",
            f"Query does not return all rows. Expected {expected_count}, got {len(rows)}."
        )

    def test_discount_eligible_column_present(self):
        """Check that discount_eligible column is present in result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        condition = "discount_eligible" in [normalize_text(col) for col in columns]
        _dynamic_test(
            self,
            condition,
            "discount_eligible column is present in result.",
            "discount_eligible column is missing in result."
        )

    def test_discount_eligible_yes(self):
        """Check that rows with region East or West have discount_eligible = 'Yes'."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        rows = self.cursor.fetchall()
        df = pd.DataFrame(rows, columns=columns)
        if 'discount_eligible' not in df.columns or 'region' not in df.columns:
            _dynamic_test(self, False, "", "discount_eligible or region column missing.")
            return
        subset = df[df['region'].isin(['East', 'West'])]
        condition = all(subset['discount_eligible'] == 'Yes')
        _dynamic_test(
            self,
            condition,
            "Rows with region East or West have discount_eligible = 'Yes'.",
            "Some rows with region East or West do not have discount_eligible = 'Yes'."
        )

    def test_discount_eligible_no(self):
        """Check that rows with region not East or West have discount_eligible = 'No'."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        rows = self.cursor.fetchall()
        df = pd.DataFrame(rows, columns=columns)
        if 'discount_eligible' not in df.columns or 'region' not in df.columns:
            _dynamic_test(self, False, "", "discount_eligible or region column missing.")
            return
        subset = df[~df['region'].isin(['East', 'West'])]
        condition = all(subset['discount_eligible'] == 'No')
        _dynamic_test(
            self,
            condition,
            "Rows with region not East or West have discount_eligible = 'No'.",
            "Some rows with region not East or West do not have discount_eligible = 'No'."
        )


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: Regional Discount Flag

解答