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_monthly_grouping_and_sum(self):
        """Check that each row represents a unique month and total_revenue is sum of that month's 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)

        # Calculate expected monthly sums
        self.cursor.execute('''
            SELECT DATE_TRUNC('month', revenue_date) AS month, SUM(amount) AS total_revenue
            FROM revenue
            GROUP BY DATE_TRUNC('month', revenue_date)
            ORDER BY month
        ''')
        expected = self.cursor.fetchall()
        expected_df = pd.DataFrame(expected, columns=["month", "total_revenue"])

        # Merge and check sums
        merged = pd.merge(result_df, expected_df, on="month", suffixes=("_user", "_expected"))
        sums_match = all(abs(merged["total_revenue_user"] - merged["total_revenue_expected"]) < 0.01)

        _dynamic_test(
            self,
            sums_match,
            "Each row represents a unique month and the total revenue matches the expected sum.",
            "Monthly grouping or sum is incorrect."
        )

    def test_lag_previous_month(self):
        """Check that previous_month_revenue column correctly uses the previous month's total 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)

        # Check previous_month_revenue for all but the first month
        prevs = result_df["total_revenue"].shift(1)
        diffs = abs(result_df["previous_month_revenue"] - prevs)
        correct = diffs[1:].lt(0.01).all()
        _dynamic_test(
            self,
            correct,
            "previous_month_revenue matches the previous month's total revenue.",
            "previous_month_revenue does not match the previous month's total revenue."
        )

    def test_percentage_growth_calculation(self):
        """Check that percent_growth is correctly calculated."""
        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)

        # Calculate expected percent growth
        result_df["expected_growth"] = None
        for i in range(1, len(result_df)):
            prev = result_df.loc[i-1, "total_revenue"]
            curr = result_df.loc[i, "total_revenue"]
            if prev == 0 or pd.isnull(prev):
                result_df.loc[i, "expected_growth"] = None
            else:
                result_df.loc[i, "expected_growth"] = round(((curr - prev) / prev) * 100, 2)

        # Compare user's percent_growth to expected
        user_growth = result_df["percent_growth"].tolist()
        expected_growth = result_df["expected_growth"].tolist()
        match = True
        for u, e in zip(user_growth, expected_growth):
            if pd.isnull(e):
                if not pd.isnull(u):
                    match = False
                    break
            else:
                if u is None or abs(u - e) > 0.01:
                    match = False
                    break
        _dynamic_test(
            self,
            match,
            "percent_growth is correctly calculated for each month.",
            "percent_growth calculation is incorrect."
        )

    def test_null_for_first_month_and_zero_prev(self):
        """Check that percent_growth is NULL for first month and when previous month's revenue is zero."""
        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)

        # First row's percent_growth must be None/NaN
        null_first = pd.isnull(result_df.loc[0, "percent_growth"])
        # Simulate a zero previous month
        # Add a fake zero-revenue month and check
        self.cursor.execute("""
            INSERT INTO revenue (revenue_date, amount) VALUES ('2025-01-31', 0.00)
        """)
        self.connection.commit()
        # Re-run user's query
        self.cursor.execute(query)
        rows2 = self.cursor.fetchall()
        columns2 = [desc[0] for desc in self.cursor.description]
        result_df2 = pd.DataFrame(rows2, columns=columns2)
        # Find row where previous_month_revenue is 0
        mask = result_df2["previous_month_revenue"] == 0
        all_null = result_df2[mask]["percent_growth"].isnull().all()
        _dynamic_test(
            self,
            null_first and all_null,
            "percent_growth is NULL for the first month and when previous month's revenue is zero.",
            "percent_growth is not NULL for the first month or when previous month's revenue is zero."
        )

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: Calculate Percentage Growth

解答