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_discrepancy_accounts_correct(self):
        """Check that only accounts with mismatched balances are returned, and the correct columns are present."""
        with open("query.sql", "r") as file:
            query = file.read()

        # Compute expected results
        solution_query = '''
            SELECT b.account_id, b.balance AS recorded_balance,
                   COALESCE(SUM(CASE WHEN t.txn_type = 'deposit' THEN t.amount WHEN t.txn_type = 'withdrawal' THEN -t.amount ELSE 0 END), 0) AS txn_sum
            FROM balances b
            LEFT JOIN bank_transactions t ON b.account_id = t.account_id
            GROUP BY b.account_id, b.balance
            HAVING b.balance <> COALESCE(SUM(CASE WHEN t.txn_type = 'deposit' THEN t.amount WHEN t.txn_type = 'withdrawal' THEN -t.amount ELSE 0 END), 0);
        '''
        self.cursor.execute(solution_query)
        expected_rows = self.cursor.fetchall()
        expected_cols = [desc[0] for desc in self.cursor.description]
        expected_df = pd.DataFrame(expected_rows, columns=expected_cols)

        self.cursor.execute(query)
        user_rows = self.cursor.fetchall()
        user_cols = [desc[0] for desc in self.cursor.description]
        user_df = pd.DataFrame(user_rows, columns=user_cols)

        # Check columns
        required_cols = set(["account_id", "recorded_balance", "txn_sum"])
        has_required_cols = required_cols.issubset(set(user_cols))

        # Check that all expected rows are present
        missing_rows = []
        for _, expected_row in expected_df.iterrows():
            mask = (user_df[list(required_cols)] == expected_row[list(required_cols)]).all(axis=1)
            if not mask.any():
                missing_rows.append(expected_row.to_dict())

        # Check that user did not return extra rows
        extra_rows = []
        for _, user_row in user_df.iterrows():
            mask = (expected_df[list(required_cols)] == user_row[list(required_cols)]).all(axis=1)
            if not mask.any():
                extra_rows.append(user_row.to_dict())

        condition = has_required_cols and len(missing_rows) == 0 and len(extra_rows) == 0
        msg = ""
        if not has_required_cols:
            msg += f"Missing required columns. Expected columns: {required_cols}. "
        if missing_rows:
            msg += f"The following expected rows are missing: {missing_rows}. "
        if extra_rows:
            msg += f"The following extra rows were returned: {extra_rows}."

        _dynamic_test(
            self,
            condition,
            "Query returns exactly the accounts with mismatched balances and correct columns.",
            msg
        )

    def test_query_returns_empty_when_balances_match(self):
        """Check that query returns no rows when all balances are correct."""
        # Fix all balances to match calculated sums
        fix_query = '''
        UPDATE balances SET balance = (
            SELECT COALESCE(SUM(CASE WHEN txn_type = 'deposit' THEN amount WHEN txn_type = 'withdrawal' THEN -amount ELSE 0 END), 0)
            FROM bank_transactions WHERE account_id = balances.account_id
        );
        '''
        self.cursor.execute(fix_query)
        self.connection.commit()

        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        condition = len(rows) == 0
        _dynamic_test(
            self,
            condition,
            "Query returns no rows when all balances match their transaction sums.",
            "Query should return no rows when balances are correct, but some rows were returned."
        )

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: Detect Duplicate Transactions

Рішення

Challenge: Detect Duplicate Transactions

Рішення