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_order_priority_column_and_labels(self):
        """Check if all orders are labeled correctly as 'High' or 'Standard' in the first query."""
        with open("query.sql", "r") as file:
            queries = file.read().split(';')
        # First query for orders
        query = queries[0].strip()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        condition = 'order_priority' in df.columns
        _dynamic_test(
            self,
            condition,
            "Column order_priority exists in orders query result.",
            "Column order_priority is missing from orders query result."
        )
        # Check labels
        correct = True
        for _, row in df.iterrows():
            if row['total_amount'] >= 500:
                if row['order_priority'] != 'High':
                    correct = False
                    break
            else:
                if row['order_priority'] != 'Standard':
                    correct = False
                    break
        _dynamic_test(
            self,
            correct,
            "All orders are labeled correctly as 'High' or 'Standard'.",
            "Some orders are not labeled correctly as 'High' or 'Standard'."
        )

    def test_product_type_column_and_labels(self):
        """Check if all products are categorized as 'Tech' or 'Other' in the second query."""
        with open("query.sql", "r") as file:
            queries = file.read().split(';')
        # Second query for products
        query = queries[1].strip()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        condition = 'product_type' in df.columns
        _dynamic_test(
            self,
            condition,
            "Column product_type exists in products query result.",
            "Column product_type is missing from products query result."
        )
        # Get product_id for Electronics category
        self.cursor.execute("""
            SELECT p.product_id
            FROM products p
            JOIN product_categories pc ON p.category_id = pc.category_id
            WHERE pc.category_name = 'Electronics'
        """)
        tech_ids = set([row[0] for row in self.cursor.fetchall()])
        correct = True
        for _, row in df.iterrows():
            if row['product_id'] in tech_ids:
                if row['product_type'] != 'Tech':
                    correct = False
                    break
            else:
                if row['product_type'] != 'Other':
                    correct = False
                    break
        _dynamic_test(
            self,
            correct,
            "All products are categorized correctly as 'Tech' or 'Other'.",
            "Some products are not categorized correctly as 'Tech' or 'Other'."
        )

    def test_orders_query_returns_all_columns(self):
        """Check if orders query returns all required columns."""
        with open("query.sql", "r") as file:
            queries = file.read().split(';')
        query = queries[0].strip()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required = set(['order_id', 'order_date', 'total_amount', 'order_priority'])
        condition = required.issubset(set(columns))
        _dynamic_test(
            self,
            condition,
            "Orders query returns all required columns.",
            "Orders query is missing required columns."
        )

    def test_products_query_returns_all_columns(self):
        """Check if products query returns all required columns."""
        with open("query.sql", "r") as file:
            queries = file.read().split(';')
        query = queries[1].strip()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required = set(['product_id', 'product_name', 'price', 'product_type'])
        condition = required.issubset(set(columns))
        _dynamic_test(
            self,
            condition,
            "Products query returns all required columns.",
            "Products query is missing required columns."
        )

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: Apply Business Logic with CASE WHEN

解答