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_result_categories_and_totals(self):
        """Check that only categories with total sales > 150 are returned, and correct totals are present."""
        with open("query.sql", "r") as file:
            query = file.read()

        expected = pd.DataFrame([
            {"category": "Electronics", "total_sales": 257.94},
            {"category": "Home Appliances", "total_sales": 179.98},
        ])

        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)

        # Normalize column names
        result_df.columns = [normalize_text(c) for c in result_df.columns]
        expected.columns = [normalize_text(c) for c in expected.columns]

        # Convert total_sales column to float for arithmetic comparison
        if 'total_sales' in result_df.columns:
            result_df['total_sales'] = result_df['total_sales'].astype(float)

        # Check that all expected rows are present (category, total_sales)
        missing_rows = []
        for _, expected_row in expected.iterrows():
            expected_dict = expected_row.to_dict()
            # Allow small rounding errors for total_sales
            mask = (
                (result_df['category'] == expected_dict['category']) &
                (abs(result_df['total_sales'] - expected_dict['total_sales']) < 0.01)
            )
            if not mask.any():
                missing_rows.append(expected_dict)

        # Check that no extra categories are present
        extra_rows = []
        for _, result_row in result_df.iterrows():
            cat = result_row['category']
            val = result_row['total_sales']
            if not (
                (expected['category'] == cat) &
                (abs(expected['total_sales'] - val) < 0.01)
            ).any():
                extra_rows.append({'category': cat, 'total_sales': val})

        condition = len(missing_rows) == 0 and len(extra_rows) == 0
        msg = ""
        if missing_rows:
            msg += f"Missing expected rows: {missing_rows}\\n"
        if extra_rows:
            msg += f"Unexpected extra rows: {extra_rows}\\n"
        _dynamic_test(
            self,
            condition,
            "Query returns only categories with total sales > 150 and correct totals.",
            msg
        )

    def test_column_names(self):
        """Check that the result contains columns for category and total_sales."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [normalize_text(desc[0]) for desc in self.cursor.description]
        _dynamic_test(
            self,
            set(['category', 'total_sales']).issubset(set(columns)),
            "Result includes 'category' and 'total_sales' columns.",
            f"Result is missing required columns. Columns found: {columns}"
        )

if __name__ == "__main__":
    unittest.main()

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: Filter Top Sales

解答