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_only_expense_category_rows(self):
        """Check that only rows where category is 'Expense' are returned."""
        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)

        # Query the table for correct rows
        self.cursor.execute("SELECT transaction_date, amount, description FROM transactions WHERE category = 'Expense' ORDER BY transaction_date")
        expense_rows = self.cursor.fetchall()
        expense_df = pd.DataFrame(expense_rows, columns=["transaction_date", "amount", "description"])

        # Check that all rows in result_df are in expense_df
        merged = pd.merge(result_df[["transaction_date", "amount", "description"]], expense_df, how="left", on=["transaction_date", "amount", "description"])
        only_expense = merged.notnull().all(axis=1).all()
        _dynamic_test(
            self,
            only_expense,
            "All returned rows have category 'Expense'.",
            "Some rows are not from category 'Expense'."
        )

    def test_columns_present(self):
        """Check that all required columns are present."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required = {"transaction_date", "amount", "description", "running_total"}
        _dynamic_test(
            self,
            required.issubset(set(columns)),
            "All required columns are present.",
            f"Missing columns: {required - set(columns)}"
        )

    def test_running_total_calculation(self):
        """Check that running_total is cumulative sum of amount by 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]
        df = pd.DataFrame(rows, columns=columns)
        df = df.sort_values("transaction_date").reset_index(drop=True)
        # Calculate expected running total
        df["expected"] = df["amount"].cumsum()
        # Convert to numeric before rounding
        df["running_total"] = pd.to_numeric(df["running_total"], errors="coerce")
        df["expected"] = pd.to_numeric(df["expected"], errors="coerce")
        correct = (df["running_total"].round(2) == df["expected"].round(2)).all()
        _dynamic_test(
            self,
            correct,
            "running_total column contains correct cumulative sums.",
            "running_total column does not match expected cumulative sums."
        )

    def test_order_by_transaction_date(self):
        """Check that results are ordered by transaction_date ascending."""
        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)
        ordered = (df["transaction_date"].tolist() == sorted(df["transaction_date"].tolist()))
        _dynamic_test(
            self,
            ordered,
            "Results are ordered by transaction_date ascending.",
            "Results are not ordered by transaction_date ascending."
        )

    def test_row_count_matches_expenses(self):
        """Check that the number of result rows matches the number of expense transactions."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result_rows = self.cursor.fetchall()
        self.cursor.execute("SELECT COUNT(*) FROM transactions WHERE category = 'Expense'")
        expense_count = self.cursor.fetchone()[0]
        _dynamic_test(
            self,
            len(result_rows) == expense_count,
            "Row count matches number of expense transactions.",
            f"Row count ({len(result_rows)}) does not match number of expense transactions ({expense_count})."
        )


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: Calculate Running Totals

Рішення

Challenge: Calculate Running Totals

Рішення