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_all_accounts_present(self):
        """Check that all accounts are present in the query result."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)
        self.cursor.execute("SELECT id, account_name FROM accounts ORDER BY id;")
        accounts = self.cursor.fetchall()
        for acc_id, acc_name in accounts:
            mask = (result_df["account_id"] == acc_id) & (result_df["account_name"] == acc_name)
            _dynamic_test(
                self,
                mask.any(),
                f"Account '{acc_name}' (ID {acc_id}) is present in result.",
                f"Account '{acc_name}' (ID {acc_id}) is missing in result."
            )

    def test_accounts_with_no_transactions_have_nulls(self):
        """Check that accounts with no transactions have NULLs for entry columns."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)
        # Find accounts with no entries
        self.cursor.execute("""
            SELECT a.id, a.account_name FROM accounts a
            LEFT JOIN entries e ON a.id = e.account_id
            GROUP BY a.id, a.account_name
            HAVING COUNT(e.id) = 0;
        """)
        no_entry_accounts = self.cursor.fetchall()
        for acc_id, acc_name in no_entry_accounts:
            mask = (result_df["account_id"] == acc_id)
            subset = result_df[mask]
            has_nulls = (
                subset["entry_id"].isnull().all() and
                subset["entry_date"].isnull().all() and
                subset["amount"].isnull().all() and
                subset["entry_type"].isnull().all()
            )
            _dynamic_test(
                self,
                has_nulls,
                f"Account '{acc_name}' (ID {acc_id}) with no transactions has NULLs for entry columns.",
                f"Account '{acc_name}' (ID {acc_id}) with no transactions does not have NULLs for entry columns."
            )

    def test_transaction_rows_match_entries(self):
        """Check that every transaction of every account is present as a row."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)
        self.cursor.execute("SELECT id, account_id, entry_date, amount, entry_type FROM entries;")
        entries = self.cursor.fetchall()
        missing = []
        for eid, acc_id, edate, amount, etype in entries:
            mask = (
                (result_df["entry_id"] == eid) &
                (result_df["account_id"] == acc_id) &
                (result_df["entry_date"] == edate) &
                (result_df["amount"] == amount) &
                (result_df["entry_type"] == etype)
            )
            if not mask.any():
                missing.append(f"Entry {eid} for account {acc_id} is missing.")
        _dynamic_test(
            self,
            len(missing) == 0,
            "All transactions are present as rows.",
            "The following transactions are missing from result:\\n" + "\\n".join(missing)
        )

    def test_columns_present(self):
        """Check that all required columns are present in the result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        expected = {"account_id", "account_name", "entry_id", "entry_date", "amount", "entry_type"}
        missing = expected - set(columns)
        _dynamic_test(
            self,
            not missing,
            "All required columns are present.",
            f"Missing columns: {', '.join(missing)}"
        )

    def test_ordering_by_account_and_date(self):
        """Check that results are ordered by account_id and entry_date."""
        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]
        result_df = pd.DataFrame(rows, columns=columns)
        # For rows with NULL entry_date, treat as earliest
        result_df["entry_date_rank"] = result_df["entry_date"].apply(lambda x: pd.Timestamp.min if pd.isnull(x) else x)
        sorted_df = result_df.sort_values(["account_id", "entry_date_rank"]).reset_index(drop=True)
        # Compare ordering
        is_ordered = result_df[["account_id", "entry_date_rank"]].reset_index(drop=True).equals(
            sorted_df[["account_id", "entry_date_rank"]].reset_index(drop=True)
        )
        _dynamic_test(
            self,
            is_ordered,
            "Rows are ordered by account_id and entry_date.",
            "Rows are not ordered by account_id and entry_date."
        )

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: Account Transaction Listing

Рішення

Challenge: Account Transaction Listing

Рішення