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_running_total_sales_amount_by_date(self):
        """Check running total of sales amount by sale date is correct."""
        with open("query.sql", "r") as file:
            query = file.read().split(';')[0]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        # Compute expected running total
        self.cursor.execute("""
        SELECT d.date, SUM(s.amount) OVER (ORDER BY d.date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) as running_total_sales
        FROM sales s JOIN dates d ON s.sale_date_id = d.date_id
        ORDER BY d.date;
        """)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        expected_df = pd.DataFrame(rows, columns=columns)
        result = df.round(2).values.tolist()
        expected = expected_df.round(2).values.tolist()
        condition = result == expected
        _dynamic_test(
            self,
            condition,
            "Running total of sales amount by date is correct.",
            "Running total of sales amount by date is incorrect."
        )

    def test_moving_average_sales_amount_3day(self):
        """Check 3-day moving average of sales amount by sale date is correct."""
        with open("query.sql", "r") as file:
            query = file.read().split(';')[1]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        # Compute expected 3-day moving average
        self.cursor.execute("""
        SELECT d.date, AVG(s.amount) OVER (ORDER BY d.date ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) as moving_avg_sales_3day
        FROM sales s JOIN dates d ON s.sale_date_id = d.date_id
        ORDER BY d.date;
        """)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        expected_df = pd.DataFrame(rows, columns=columns)
        result = df.round(2).values.tolist()
        expected = expected_df.round(2).values.tolist()
        condition = result == expected
        _dynamic_test(
            self,
            condition,
            "3-day moving average of sales amount by date is correct.",
            "3-day moving average of sales amount by date is incorrect."
        )

    def test_employee_running_total_sales(self):
        """Check running total of sales amount for each employee is correct."""
        with open("query.sql", "r") as file:
            query = file.read().split(';')[2]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        # Compute expected running total per employee
        self.cursor.execute("""
        SELECT e.employee_id, e.first_name, d.date, SUM(s.amount) OVER (PARTITION BY e.employee_id ORDER BY d.date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) as employee_running_total_sales
        FROM sales s JOIN dates d ON s.sale_date_id = d.date_id JOIN employees e ON s.employee_id = e.employee_id
        ORDER BY e.employee_id, d.date;
        """)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        expected_df = pd.DataFrame(rows, columns=columns)
        result = df.round(2).values.tolist()
        expected = expected_df.round(2).values.tolist()
        condition = result == expected
        _dynamic_test(
            self,
            condition,
            "Running total of sales amount per employee is correct.",
            "Running total of sales amount per employee is incorrect."
        )

    def test_employee_moving_avg_sales_achieved_2review(self):
        """Check 2-period moving average of sales achieved in reviews for each employee is correct."""
        with open("query.sql", "r") as file:
            query = file.read().split(';')[3]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        # Compute expected moving average
        self.cursor.execute("""
        SELECT e.employee_id, e.first_name, d.date, AVG(ep.sales_achieved) OVER (PARTITION BY e.employee_id ORDER BY d.date ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) as moving_avg_sales_achieved_2review
        FROM employee_performance ep JOIN dates d ON ep.review_date_id = d.date_id JOIN employees e ON ep.employee_id = e.employee_id
        ORDER BY e.employee_id, d.date;
        """)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        expected_df = pd.DataFrame(rows, columns=columns)
        result = df.round(2).values.tolist()
        expected = expected_df.round(2).values.tolist()
        condition = result == expected
        _dynamic_test(
            self,
            condition,
            "2-period moving average of sales achieved in reviews per employee is correct.",
            "2-period moving average of sales achieved in reviews per employee is incorrect."
        )

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: Analyze Trends with Window Functions

解答