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_balance_unchanged_after_rollback(self):
        """Check that the balance for account_id=1001 remains unchanged after the simulated failed transaction."""
        # Get initial balance
        self.cursor.execute("SELECT balance FROM balances WHERE account_id = 1001;")
        initial_balance = self.cursor.fetchone()[0]

        # Run the user's query
        with open("query.sql", "r") as file:
            query = file.read()
        try:
            self.cursor.execute(query)
        except Exception:
            self.connection.rollback()  # Ensure rollback in case of error

        # Check balance again
        self.cursor.execute("SELECT balance FROM balances WHERE account_id = 1001;")
        final_balance = self.cursor.fetchone()[0]
        condition = initial_balance == final_balance
        _dynamic_test(
            self,
            condition,
            "The balance for account_id=1001 remains unchanged after the rollback.",
            f"The balance for account_id=1001 changed after the transaction. Initial: {initial_balance}, Final: {final_balance}."
        )

    def test_no_changes_committed_on_error(self):
        """Check that no balances are changed if an error occurs during the transaction."""
        # Get all initial balances
        self.cursor.execute("SELECT account_id, balance FROM balances ORDER BY account_id;")
        initial_balances = self.cursor.fetchall()

        # Run the user's query
        with open("query.sql", "r") as file:
            query = file.read()
        try:
            self.cursor.execute(query)
        except Exception:
            self.connection.rollback()  # Ensure rollback in case of error

        # Get all balances again
        self.cursor.execute("SELECT account_id, balance FROM balances ORDER BY account_id;")
        final_balances = self.cursor.fetchall()

        condition = initial_balances == final_balances
        _dynamic_test(
            self,
            condition,
            "No balances changed after the failed transaction.",
            f"Some balances were changed after the failed transaction. Initial: {initial_balances}, Final: {final_balances}."
        )

    def test_simulated_error_causes_rollback(self):
        """Check that the transaction contains a simulated error (e.g., reference to a non-existent column) that causes a rollback."""
        with open("query.sql", "r") as file:
            query = file.read()
        error_triggered = False
        try:
            self.cursor.execute(query)
        except Exception as e:
            error_triggered = True
        _dynamic_test(
            self,
            error_triggered,
            "A simulated error was triggered and caused a rollback.",
            "No simulated error was found in the transaction, or the transaction did not fail as expected."
        )

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: Simulate a Transaction Rollback

解答