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_columns_and_order(self):
        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]
        expected_columns = ["month", "total_revenue", "revenue_change"]
        _dynamic_test(
            self,
            columns == expected_columns,
            "Query returns correct column names in correct order.",
            f"Expected columns {expected_columns}, got {columns}."
        )
        # Check that months are ordered ascending
        months = [row[0] for row in rows]
        _dynamic_test(
            self,
            months == sorted(months),
            "Months are ordered in ascending order.",
            "Months are not ordered in ascending order."
        )

    def test_monthly_revenue_and_change(self):
        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)
        # Compute expected values
        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
        """)
        monthly = self.cursor.fetchall()
        expected = []
        for idx, (month, total) in enumerate(monthly):
            if idx == 0:
                change = None
            else:
                change = float(total) - float(monthly[idx-1][1])
            expected.append({
                "month": month,
                "total_revenue": float(total),
                "revenue_change": change
            })
        for idx, exp in enumerate(expected):
            row = result_df.iloc[idx]
            _dynamic_test(
                self,
                row["month"] == exp["month"],
                f"Row {idx+1}: correct month.",
                f"Row {idx+1}: expected month {exp['month']}, got {row['month']}"
            )
            _dynamic_test(
                self,
                abs(float(row["total_revenue"]) - exp["total_revenue"]) < 0.01,
                f"Row {idx+1}: correct total_revenue.",
                f"Row {idx+1}: expected total_revenue {exp['total_revenue']}, got {row['total_revenue']}"
            )
            if exp["revenue_change"] is None:
                _dynamic_test(
                    self,
                    pd.isnull(row["revenue_change"]),
                    f"Row {idx+1}: revenue_change is NULL for first month.",
                    f"Row {idx+1}: expected revenue_change NULL, got {row['revenue_change']}"
                )
            else:
                _dynamic_test(
                    self,
                    abs(float(row["revenue_change"]) - exp["revenue_change"]) < 0.01,
                    f"Row {idx+1}: correct revenue_change.",
                    f"Row {idx+1}: expected revenue_change {exp['revenue_change']}, got {row['revenue_change']}"
                )

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: Month-over-Month Revenue Growth

解答