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_columns_present(self):
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected = {'order_id', 'region_name', 'order_label'}
        _dynamic_test(
            self,
            expected.issubset(set(columns)),
            "All required columns are present.",
            f"Missing columns: {expected - set(columns)}"
        )

    def test_priority_label(self):
        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]
        df = pd.DataFrame(rows, columns=columns)
        # Order 2: Europe, total_amount = 69.98 (should NOT be Priority)
        # Order 7: Europe, total_amount = 89.99 (should NOT be Priority)
        # No Europe order > 100 except order 2 and 7 (both below 100)
        # So, test that no order is wrongly labeled as Priority unless Europe and >100
        priorities = df[(df['region_name'] == 'Europe') & (df['order_label'] == 'Priority')]
        correct = all(df[(df['region_name'] == 'Europe') & (df['total_amount'] > 100)]['order_label'] == 'Priority')
        _dynamic_test(
            self,
            priorities.empty,
            "No 'Priority' label assigned incorrectly.",
            "'Priority' label assigned to Europe orders with total_amount <= 100."
        )

    def test_high_value_label(self):
        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]
        df = pd.DataFrame(rows, columns=columns)
        # Orders with total_amount > 200, but not Priority (not Europe > 100)
        # Order 1: North America, 749.98 -> High Value
        # Order 3: Asia, 1199.99 -> High Value
        high_value_orders = df[(df['order_label'] == 'High Value')]
        condition = True
        for _, row in high_value_orders.iterrows():
            if row['region_name'] == 'Europe' and row['total_amount'] > 100:
                condition = False
                break
            if row['total_amount'] <= 200:
                condition = False
                break
        _dynamic_test(
            self,
            condition,
            "'High Value' label assigned only to correct orders.",
            "'High Value' label incorrectly assigned."
        )

    def test_regular_label(self):
        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]
        df = pd.DataFrame(rows, columns=columns)
        # Orders that are not Priority or High Value should be Regular
        regular_orders = df[df['order_label'] == 'Regular']
        condition = True
        for _, row in regular_orders.iterrows():
            if (row['region_name'] == 'Europe' and row['total_amount'] > 100):
                condition = False
                break
            if (row['total_amount'] > 200):
                condition = False
                break
        _dynamic_test(
            self,
            condition,
            "'Regular' label assigned only to correct orders.",
            "'Regular' label incorrectly assigned."
        )

    def test_full_label_set(self):
        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]
        df = pd.DataFrame(rows, columns=columns)
        # Only allowed labels
        allowed = {'Priority', 'High Value', 'Regular'}
        actual = set(df['order_label'])
        _dynamic_test(
            self,
            actual <= allowed,
            "Only valid labels are present.",
            f"Invalid labels found: {actual - allowed}"
        )

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

Master SQL from the ground up with a focus on business intelligence. Learn to write powerful queries for KPIs, implement business logic, analyze trends with window functions, design data models for BI tools, and apply reporting patterns. Each section blends engaging theory with hands-on, real-world challenges to prepare you for BI analyst roles.

Learn to use SQL aggregations to calculate and interpret key business metrics, such as revenue, customer counts, and averages, essential for BI analysis.

Implement business rules and conditional logic in SQL queries using CASE WHEN, enabling dynamic categorization and decision-making.

Analyze business trends and rankings over time using advanced SQL window functions, essential for BI trend analysis and leaderboards.

Challenge: Advanced Business Logic

解答