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_yoy_growth_percent_correctness(self):
        """Check if the query returns correct YoY growth percentages for each month."""
        with open("query.sql", "r") as file:
            query = file.read()

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

        # Prepare expected results
        expected = []
        # Data from init script, grouped by year and month
        month_revenue = {
            (2022, 1): 12000.00 + 13500.50,
            (2022, 2): 14000.00 + 14250.75,
            (2022, 3): 15000.00 + 15500.25,
            (2022, 4): 16000.00 + 16250.00,
            (2022, 5): 17000.00 + 17500.00,
            (2023, 1): 18000.00 + 18500.00,
            (2023, 2): 19000.00 + 19500.00,
            (2023, 3): 20000.00 + 20500.00,
            (2023, 4): 21000.00 + 21500.00,
            (2023, 5): 22000.00 + 22500.00,
            (2024, 1): 23000.00 + 23500.00,
            (2024, 2): 24000.00 + 24500.00,
            (2024, 3): 25000.00 + 25500.00,
            (2024, 4): 26000.00 + 26500.00,
            (2024, 5): 27000.00 + 27500.00,
        }
        # For each month (1-5), for years 2023 and 2024, calculate expected growth
        for month in range(1, 6):
            for year in [2023, 2024]:
                current = month_revenue[(year, month)]
                last = month_revenue[(year-1, month)]
                growth = ((current - last) / last) * 100
                expected.append({
                    'year': float(year),
                    'month': float(month),
                    'total_revenue': float(current),
                    'last_year_revenue': float(last),
                    'yoy_growth_percent': growth
                })
        # Convert to DataFrame for easy comparison
        expected_df = pd.DataFrame(expected)
        # Round numeric columns for comparison
        result_df['total_revenue'] = result_df['total_revenue'].astype(float).round(2)
        result_df['last_year_revenue'] = result_df['last_year_revenue'].astype(float).round(2)
        result_df['yoy_growth_percent'] = result_df['yoy_growth_percent'].astype(float).round(6)
        expected_df['total_revenue'] = expected_df['total_revenue'].round(2)
        expected_df['last_year_revenue'] = expected_df['last_year_revenue'].round(2)
        expected_df['yoy_growth_percent'] = expected_df['yoy_growth_percent'].round(6)
        # Compare row by row
        missing = []
        for _, ex_row in expected_df.iterrows():
            mask = (
                (result_df['year'] == ex_row['year']) &
                (result_df['month'] == ex_row['month']) &
                (result_df['total_revenue'] == ex_row['total_revenue']) &
                (result_df['last_year_revenue'] == ex_row['last_year_revenue']) &
                (abs(result_df['yoy_growth_percent'] - ex_row['yoy_growth_percent']) < 0.0001)
            )
            if not mask.any():
                missing.append(ex_row.to_dict())
        condition = len(missing) == 0
        if not condition:
            failure_msg = f"The following expected rows are missing or incorrect (showing up to 2):\\n" + "\\n".join(str(m) for m in missing[:2])
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "All expected YoY growth rows and values are present.",
            failure_msg
        )

    def test_no_null_last_year_revenue(self):
        """Check that there are no rows with null last_year_revenue."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        null_count = result_df['last_year_revenue'].isnull().sum()
        _dynamic_test(
            self,
            null_count == 0,
            "No rows with null last_year_revenue.",
            f"Found {null_count} rows with null last_year_revenue."
        )

    def test_column_names(self):
        """Check that the query returns the required columns."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required = ['year', 'month', 'total_revenue', 'last_year_revenue', 'yoy_growth_percent']
        condition = all(col in columns for col in required)
        missing = [col for col in required if col not in columns]
        _dynamic_test(
            self,
            condition,
            "All required columns are present.",
            f"Missing columns: {', '.join(missing)}"
        )

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 essentials for finance and accounting by learning to analyze time-series data, perform period-over-period comparisons, execute ledger-style joins, ensure data accuracy, and understand transaction logic. This course is designed for absolute beginners and uses real-world finance scenarios to make learning engaging and practical.

Learn to aggregate and analyze financial data over time, such as daily sales, monthly expenses, and yearly profits, using SQL's aggregation functions and date handling.

Learn to compare financial performance across different periods, such as month-over-month or year-over-year growth, using SQL.

Master ledger-style joins to reconcile accounts, match transactions, and ensure financial data integrity.

Ensure data accuracy, perform reconciliation, and understand transaction logic and ACID properties in SQL for reliable financial operations.

Challenge: Year-over-Year Revenue Comparison

解答