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_total_revenue(self):
        """Check that total revenue is correct."""
        with open("query.sql", "r") as file:
            queries = file.read()
        # Only extract the first query
        user_query = queries.split(';')[0].strip()
        self.cursor.execute(user_query)
        row = self.cursor.fetchone()
        total_revenue = row[0]
        # Calculate expected total revenue
        self.cursor.execute("SELECT SUM(total_amount) FROM sales;")
        expected = self.cursor.fetchone()[0]
        _dynamic_test(
            self,
            abs(total_revenue - expected) < 0.01,
            "Total revenue is correct.",
            f"Total revenue is incorrect. Expected {expected}, got {total_revenue}."
        )

    def test_average_order_value(self):
        """Check that average order value is correct."""
        with open("query.sql", "r") as file:
            queries = file.read()
        user_queries = [q.strip() for q in queries.split(';') if q.strip()]
        if len(user_queries) < 2:
            _dynamic_test(self, False, "", "Second query (average order value) is missing.")
            return
        user_query = user_queries[1]
        self.cursor.execute(user_query)
        row = self.cursor.fetchone()
        avg_order_value = row[0]
        self.cursor.execute("SELECT AVG(total_amount) FROM sales;")
        expected = self.cursor.fetchone()[0]
        _dynamic_test(
            self,
            abs(avg_order_value - expected) < 0.01,
            "Average order value is correct.",
            f"Average order value is incorrect. Expected {expected}, got {avg_order_value}."
        )

    def test_unique_customers(self):
        """Check that unique customer count is correct."""
        with open("query.sql", "r") as file:
            queries = file.read()
        user_queries = [q.strip() for q in queries.split(';') if q.strip()]
        if len(user_queries) < 3:
            _dynamic_test(self, False, "", "Third query (unique customers) is missing.")
            return
        user_query = user_queries[2]
        self.cursor.execute(user_query)
        row = self.cursor.fetchone()
        unique_customers = row[0]
        self.cursor.execute("SELECT COUNT(DISTINCT customer_id) FROM sales;")
        expected = self.cursor.fetchone()[0]
        _dynamic_test(
            self,
            unique_customers == expected,
            "Unique customer count is correct.",
            f"Unique customer count is incorrect. Expected {expected}, got {unique_customers}."
        )

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: Calculate Key Business Metrics

解答