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_accounts_with_incorrect_balance(self):
        """Returns all account IDs from balances table where sum of deposits minus withdrawals does not match the recorded balance."""
        with open("query.sql", "r") as file:
            user_query = file.read()

        # The correct answer is account 1002 only
        solution_query = '''
        SELECT b.account_id
        FROM balances b
        LEFT JOIN (
            SELECT 
                account_id, 
                SUM(
                    CASE 
                        WHEN txn_type = 'deposit' THEN amount
                        WHEN txn_type = 'withdrawal' THEN -amount
                        ELSE 0
                    END
                ) AS txn_sum
            FROM bank_transactions
            GROUP BY account_id
        ) t ON b.account_id = t.account_id
        WHERE b.balance <> COALESCE(t.txn_sum, 0);
        '''
        self.cursor.execute(solution_query)
        solution_rows = self.cursor.fetchall()
        solution_ids = set([row[0] for row in solution_rows])

        self.cursor.execute(user_query)
        user_rows = self.cursor.fetchall()
        user_ids = set([row[0] for row in user_rows])

        condition = user_ids == solution_ids
        missing = solution_ids - user_ids
        extra = user_ids - solution_ids
        msg = ""
        if missing:
            msg += f"Missing account_ids: {missing}. "
        if extra:
            msg += f"Unexpected account_ids: {extra}. "
        _dynamic_test(
            self,
            condition,
            "Correctly returns all account IDs where the sum of transactions does not match the recorded balance.",
            msg or "Incorrect account IDs returned."
        )

    def test_handles_accounts_with_no_transactions(self):
        """Handles accounts with zero or no transactions correctly."""
        # Add a new account with no transactions
        self.cursor.execute("INSERT INTO balances (account_id, balance) VALUES (2000, 0)")
        with open("query.sql", "r") as file:
            user_query = file.read()
        self.cursor.execute(user_query)
        user_rows = self.cursor.fetchall()
        returned_ids = set([row[0] for row in user_rows])
        # The account should not appear in the result because balance and txn_sum (0) match
        condition = 2000 not in returned_ids
        _dynamic_test(
            self,
            condition,
            "Accounts with no transactions and zero balance are handled correctly.",
            "Account with no transactions and zero balance should not be returned."
        )

    def test_sum_calculation_logic(self):
        """Correctly calculates the sum by adding deposits and subtracting withdrawals for each account."""
        # Insert a new account with 2 deposits and 1 withdrawal
        self.cursor.execute("INSERT INTO balances (account_id, balance) VALUES (3000, 200)")
        self.cursor.execute("INSERT INTO bank_transactions (account_id, txn_date, amount, txn_type) VALUES (3000, '2024-06-10', 100, 'deposit')")
        self.cursor.execute("INSERT INTO bank_transactions (account_id, txn_date, amount, txn_type) VALUES (3000, '2024-06-11', 100, 'deposit')")
        self.cursor.execute("INSERT INTO bank_transactions (account_id, txn_date, amount, txn_type) VALUES (3000, '2024-06-12', 50, 'withdrawal')")
        with open("query.sql", "r") as file:
            user_query = file.read()
        self.cursor.execute(user_query)
        user_rows = self.cursor.fetchall()
        returned_ids = set([row[0] for row in user_rows])
        # The correct sum is 100+100-50=150, but balance is 200, so this account should be in result
        condition = 3000 in returned_ids
        _dynamic_test(
            self,
            condition,
            "Correctly calculates sum by adding deposits and subtracting withdrawals.",
            "Sum calculation is incorrect for deposits/withdrawals."
        )

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: Reconcile Account Balances

解答