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_performance_column_exists(self):
        """Check if 'performance' column is present in query output."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        _dynamic_test(
            self,
            "performance" in [c.lower() for c in columns],
            "'performance' column exists in output.",
            "'performance' column is missing from output."
        )

    def test_all_rows_included(self):
        """Check if all rows from sales table are present in output."""
        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()
        _dynamic_test(
            self,
            len(rows) == expected_count,
            f"All {expected_count} rows from sales are included.",
            f"Expected {expected_count} rows but found {len(rows)}."
        )

    def test_top_label(self):
        """Rows with quantity >= 50 AND price >= 100 must be labeled 'Top'."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Insert a matching row
        self.cursor.execute("INSERT INTO sales (id, product, quantity, price, region) VALUES (100, 'TestProd', 55, 150, 'TestReg') ON CONFLICT DO NOTHING;")
        self.cursor.connection.commit()
        self.cursor.execute(query)
        df = pd.DataFrame(self.cursor.fetchall(), columns=[desc[0] for desc in self.cursor.description])
        row = df[(df['id'] == 100)]
        label = row['performance'].iloc[0] if not row.empty else None
        _dynamic_test(
            self,
            label == 'Top',
            "Rows with quantity >= 50 and price >= 100 are labeled 'Top'.",
            "Row with quantity >= 50 and price >= 100 is not labeled 'Top'."
        )

    def test_average_label(self):
        """Rows with only one of quantity >= 50 or price >= 100 must be labeled 'Average'."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Insert two rows: one with quantity >= 50, one with price >= 100
        self.cursor.execute("INSERT INTO sales (id, product, quantity, price, region) VALUES (101, 'TestProd', 51, 50, 'TestReg') ON CONFLICT DO NOTHING;")
        self.cursor.execute("INSERT INTO sales (id, product, quantity, price, region) VALUES (102, 'TestProd', 10, 120, 'TestReg') ON CONFLICT DO NOTHING;")
        self.cursor.connection.commit()
        self.cursor.execute(query)
        df = pd.DataFrame(self.cursor.fetchall(), columns=[desc[0] for desc in self.cursor.description])
        row1 = df[(df['id'] == 101)]
        row2 = df[(df['id'] == 102)]
        label1 = row1['performance'].iloc[0] if not row1.empty else None
        label2 = row2['performance'].iloc[0] if not row2.empty else None
        cond = (label1 == 'Average') and (label2 == 'Average')
        _dynamic_test(
            self,
            cond,
            "Rows with only one condition met are labeled 'Average'.",
            "Rows with only one condition met are not labeled 'Average'."
        )

    def test_low_label(self):
        """Rows with neither quantity >= 50 nor price >= 100 must be labeled 'Low'."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Insert a row with neither condition met
        self.cursor.execute("INSERT INTO sales (id, product, quantity, price, region) VALUES (103, 'TestProd', 10, 50, 'TestReg') ON CONFLICT DO NOTHING;")
        self.cursor.connection.commit()
        self.cursor.execute(query)
        df = pd.DataFrame(self.cursor.fetchall(), columns=[desc[0] for desc in self.cursor.description])
        row = df[(df['id'] == 103)]
        label = row['performance'].iloc[0] if not row.empty else None
        _dynamic_test(
            self,
            label == 'Low',
            "Rows with neither condition met are labeled 'Low'.",
            "Row with neither condition met is not labeled 'Low'."
        )

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: Sales Performance Label

解答