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_top_2_employees_by_sales(self):
        """Check if query returns top 2 employees by total sales with correct ranking."""
        with open("query.sql", "r") as file:
            user_query = file.read()

        # Run user query
        self.cursor.execute(user_query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        user_df = pd.DataFrame(rows, columns=columns)

        # Compute expected result
        self.cursor.execute("""
            SELECT employee_id,
                   SUM(amount) AS total_sales
            FROM sales
            GROUP BY employee_id
        """)
        sales_rows = self.cursor.fetchall()
        sales_df = pd.DataFrame(sales_rows, columns=["employee_id", "total_sales"])
        sales_df["total_sales"] = sales_df["total_sales"].astype(float)
        sales_df = sales_df.sort_values("total_sales", ascending=False)
        sales_df["sales_rank"] = sales_df["total_sales"].rank(method="min", ascending=False).astype(int)
        expected_df = sales_df[sales_df["sales_rank"] <= 2].copy()
        expected_df = expected_df[["employee_id", "total_sales", "sales_rank"]].sort_values(["sales_rank", "employee_id"]).reset_index(drop=True)

        # Normalize user df columns
        user_df.columns = [c.lower() for c in user_df.columns]
        # Convert types for comparison
        user_df["total_sales"] = user_df["total_sales"].astype(float)
        user_df["sales_rank"] = user_df["sales_rank"].astype(int)
        user_df = user_df[["employee_id", "total_sales", "sales_rank"]].sort_values(["sales_rank", "employee_id"]).reset_index(drop=True)

        # Check if user result matches expected
        condition = user_df.equals(expected_df)
        failure_msg = f"Expected:\\n{expected_df}\\nGot:\\n{user_df}" if not condition else ""
        _dynamic_test(
            self,
            condition,
            "Query returns correct top 2 employees by total sales with correct ranking.",
            failure_msg
        )

    def test_only_top_2_ranks_present(self):
        """Check that only employees with rank 1 or 2 are included in the result."""
        with open("query.sql", "r") as file:
            user_query = file.read()
        self.cursor.execute(user_query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        user_df = pd.DataFrame(rows, columns=columns)
        user_df.columns = [c.lower() for c in user_df.columns]
        user_df["sales_rank"] = user_df["sales_rank"].astype(int)
        only_top_2 = user_df["sales_rank"].isin([1,2]).all() and user_df["sales_rank"].max() <= 2
        failure_msg = f"Result contains ranks outside of 1 or 2: {user_df['sales_rank'].unique()}"
        _dynamic_test(
            self,
            only_top_2,
            "Only employees with rank 1 or 2 are present.",
            failure_msg
        )

    def test_column_names(self):
        """Check that result has correct column names: employee_id, total_sales, sales_rank."""
        with open("query.sql", "r") as file:
            user_query = file.read()
        self.cursor.execute(user_query)
        result_columns = [desc[0].lower() for desc in self.cursor.description]
        expected_columns = ["employee_id", "total_sales", "sales_rank"]
        condition = result_columns == expected_columns
        failure_msg = f"Expected columns: {expected_columns}, got: {result_columns}"
        _dynamic_test(
            self,
            condition,
            "Result has correct column names.",
            failure_msg
        )

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: Build Leaderboards and Sequences

解答