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_one_row_per_quarter(self):
        """Check that the query returns one row per quarter with at least one expense."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        quarters = [row[0] for row in rows]
        expected_quarters = set()
        self.cursor.execute("""
            SELECT DISTINCT CONCAT(EXTRACT(YEAR FROM transaction_date), '-Q', EXTRACT(QUARTER FROM transaction_date))
            FROM transactions WHERE category = 'Expense'
        """)
        for row in self.cursor.fetchall():
            expected_quarters.add(row[0])
        condition = set(quarters) == expected_quarters
        _dynamic_test(
            self,
            condition,
            "One row per quarter with at least one expense.",
            f"Expected quarters: {expected_quarters}, but got: {set(quarters)}."
        )

    def test_only_expense_category(self):
        """Check that only 'Expense' category transactions are included in the calculation."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        # For each row, check the sum matches only 'Expense' transactions
        for row in rows:
            quarter = row[0]
            user_sum = row[1]
            # Calculate expected sum for this quarter
            self.cursor.execute("""
                SELECT SUM(amount)
                FROM transactions
                WHERE category = 'Expense'
                  AND CONCAT(EXTRACT(YEAR FROM transaction_date), '-Q', EXTRACT(QUARTER FROM transaction_date)) = %s
            """, (quarter,))
            expected_sum = self.cursor.fetchone()[0]
            condition = user_sum == expected_sum
            _dynamic_test(
                self,
                condition,
                f"Total for {quarter} matches sum of 'Expense' transactions.",
                f"Total for {quarter} does not match sum of 'Expense' transactions. Expected {expected_sum}, got {user_sum}."
            )

    def test_sum_by_quarter(self):
        """Check that the sum is per quarter, not for other time periods."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        # Check that no two rows have the same quarter value
        quarters = [row[0] for row in rows]
        condition = len(quarters) == len(set(quarters))
        _dynamic_test(
            self,
            condition,
            "Each quarter appears only once.",
            "Duplicate quarters found in the result."
        )

    def test_quarter_format(self):
        """Check that the quarter is displayed in the format YYYY-Qn."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        import re
        pattern = re.compile(r"^\\d{4}-Q[1-4]$")
        all_match = all(pattern.match(str(row[0])) for row in rows)
        _dynamic_test(
            self,
            all_match,
            "Quarter format is correct for all rows.",
            "Quarter format is incorrect for one or more rows."
        )

    def test_ordered_by_quarter(self):
        """Check that the results are ordered by quarter ascending."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        quarters = [row[0] for row in rows]
        # Convert to (year, quarter_num) for sorting
        def parse_q(q):
            year, qn = q.split('-Q')
            return (int(year), int(qn))
        parsed = [parse_q(q) for q in quarters]
        condition = parsed == sorted(parsed)
        _dynamic_test(
            self,
            condition,
            "Results are ordered by quarter ascending.",
            "Results are not ordered by quarter ascending."
        )

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: Quarterly Expense Report

解答