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_mismatched_debits_and_credits(self):
        """Check that only accounts with mismatched debits and credits are returned, with correct totals."""
        with open("query.sql", "r") as file:
            query = file.read()

        # Reference query for validation
        reference_query = '''
        SELECT a.account_name,
               SUM(CASE WHEN e.entry_type = 'debit' THEN e.amount ELSE 0 END) AS total_debits,
               SUM(CASE WHEN e.entry_type = 'credit' THEN e.amount ELSE 0 END) AS total_credits
        FROM accounts a
        JOIN entries e ON a.id = e.account_id
        GROUP BY a.account_name
        HAVING SUM(CASE WHEN e.entry_type = 'debit' THEN e.amount ELSE 0 END) !=
               SUM(CASE WHEN e.entry_type = 'credit' THEN e.amount ELSE 0 END)
        ORDER BY a.account_name
        '''
        self.cursor.execute(reference_query)
        expected = self.cursor.fetchall()
        expected_columns = [desc[0] for desc in self.cursor.description]
        expected_df = pd.DataFrame(expected, columns=expected_columns)

        # User query
        self.cursor.execute(query)
        actual = self.cursor.fetchall()
        actual_columns = [desc[0] for desc in self.cursor.description]
        actual_df = pd.DataFrame(actual, columns=actual_columns)

        # Compare the sets of account names
        expected_names = set(expected_df['account_name'])
        actual_names = set(actual_df['account_name'])
        names_match = expected_names == actual_names

        # Compare total_debits and total_credits for each account
        mismatch = []
        for _, row in expected_df.iterrows():
            acc = row['account_name']
            match_row = actual_df[actual_df['account_name'] == acc]
            if match_row.empty:
                mismatch.append(f"Missing row for account: {acc}")
                continue
            user_debits = float(match_row.iloc[0]['total_debits'])
            user_credits = float(match_row.iloc[0]['total_credits'])
            if not (abs(user_debits - float(row['total_debits'])) < 0.01 and abs(user_credits - float(row['total_credits'])) < 0.01):
                mismatch.append(f"Incorrect totals for account: {acc}")

        condition = names_match and not mismatch
        fail_msg = ""
        if not names_match:
            fail_msg += f"Account names do not match.\\nExpected: {expected_names}\\nGot: {actual_names}\\n"
        if mismatch:
            fail_msg += "\\n".join(mismatch)
        _dynamic_test(
            self,
            condition,
            "Returned accounts and totals are correct for mismatched debits and credits.",
            fail_msg
        )

    def test_balanced_accounts_not_included(self):
        """Check that accounts with equal total debits and credits are not included."""
        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]
        df = pd.DataFrame(rows, columns=columns)

        # There are no accounts in the init data that balance, but let's check for absence anyway
        balanced_accounts = []
        for _, row in df.iterrows():
            if abs(float(row['total_debits']) - float(row['total_credits'])) < 0.01:
                balanced_accounts.append(row['account_name'])
        condition = len(balanced_accounts) == 0
        fail_msg = f"Accounts with balanced debits and credits should not be included: {balanced_accounts}"
        _dynamic_test(
            self,
            condition,
            "No balanced accounts are included in the result.",
            fail_msg
        )

    def test_columns_present(self):
        """Check that the result includes 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 = {'account_name', 'total_debits', 'total_credits'}
        condition = required.issubset(set(columns))
        fail_msg = f"Result columns missing. Required: {required}, Got: {set(columns)}"
        _dynamic_test(
            self,
            condition,
            "All required columns are present.",
            fail_msg
        )

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

解答