test_pipeline.py

"""End-to-end pipeline tests (pytest).

Deterministic only -- no model required. Vision-extraction accuracy is checked
manually against the scan fixture (see README).
"""

import json
import os
import sys

import pytest

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
SAMPLES = os.path.join(ROOT, "samples")

from categorize import categorize_rules, enforce_closed_set  # noqa: E402
from constants import CATEGORY_SET, SUSPENSE  # noqa: E402
from excel_export import write_excel  # noqa: E402
from extraction import extract_from_pdf, parse_date  # noqa: E402
from tally_export import generate_tally_xml  # noqa: E402
from validate import reconcile  # noqa: E402


@pytest.fixture(scope="module")
def truth():
    with open(os.path.join(SAMPLES, "sample_digital_truth.json"), encoding="utf-8") as f:
        return json.load(f)


# 1. Extraction (digital) ---------------------------------------------------- #
def test_extraction_digital_exact(truth):
    txns, meta = extract_from_pdf(os.path.join(SAMPLES, "sample_digital.pdf"))
    assert meta["path"] == "text-layer"
    assert meta["gpu_used"] is False
    assert len(txns) == len(truth) == 40
    for got, want in zip(txns, truth):
        assert got["date"] == want["date"]
        assert got["debit"] == want["debit"]
        assert got["credit"] == want["credit"]
        assert got["balance"] == want["balance"]
        assert got["narration"].strip() == want["narration"].strip()


# 2. Balance reconciliation -------------------------------------------------- #
def test_reconcile_clean(truth):
    txns = [dict(t) for t in truth]
    result = reconcile(txns)
    assert result["reconciled"] == result["total"]
    assert result["total"] == 39  # 40 rows, 39 consecutive pairs
    assert "✅" in result["banner"]


def test_reconcile_flags_corrupted_row(truth):
    txns = [dict(t) for t in truth]
    txns[10]["balance"] = txns[10]["balance"] + 999.0  # corrupt one balance
    result = reconcile(txns)
    assert txns[10]["flags"], "corrupted row must be flagged"
    # the mismatch is detected on the corrupted row and its neighbour
    assert result["reconciled"] < result["total"]


# 3. Date parsing ------------------------------------------------------------ #
@pytest.mark.parametrize("raw", ["01/04/2026", "01-04-26", "1 Apr 2026",
                                 "2026-04-01", "1-April-2026", "01.04.2026"])
def test_date_parsing(raw):
    assert parse_date(raw) == "2026-04-01"


# 4. Categorization rules (deterministic) ------------------------------------ #
CANNED = [
    ("SALARY APR 2026 - STAFF PAYROLL", None, 50000.0, "Salary & Wages"),
    ("UPI/DR/123/SHARMA RENT/office rent", 35000.0, None, "Office Rent"),
    ("GST PMT-CBIC-GSTR3B", 18900.0, None, "GST Payment"),
    ("TDS PMT-CPC Q4", 9400.0, None, "TDS Payment"),
    ("INCOME TAX-ADVANCE TAX CBDT", 35000.0, None, "Income Tax Payment"),
    ("ATM CSH WDL/AXIS", 10000.0, None, "Cash Withdrawal"),
    ("CSH DEP/BRANCH CASH DEPOSIT", None, 50000.0, "Cash Deposit"),
    ("BSNL BROADBAND INTERNET BILL", 1499.0, None, "Telephone & Internet"),
    ("ELECTRICITY KESCO BILL", 8920.0, None, "Electricity & Utilities"),
    ("BANK CHRG-SMS CHGS", 23.6, None, "Bank Charges"),
    ("LIC OF INDIA PREMIUM POLICY", 9800.0, None, "Insurance"),
    ("HDFC HOME LOAN EMI", 28750.0, None, "Loan EMI"),
    ("INT CREDIT-FIXED DEPOSIT INTEREST", None, 4521.0, "Interest Received"),
    ("IRCTC RAIL TICKET TRAVEL", 3450.0, None, "Travel & Conveyance"),
    ("DRAWINGS-PROPRIETOR PERSONAL", 30000.0, None, "Drawings"),
]


@pytest.mark.parametrize("narration,debit,credit,expected", CANNED)
def test_categorization_rules(narration, debit, credit, expected):
    txn = {"narration": narration, "debit": debit, "credit": credit}
    result = categorize_rules(txn)
    assert result is not None, f"no rule matched: {narration}"
    assert result[0] == expected


# 5. Closed-set guarantee ---------------------------------------------------- #
def test_closed_set_rejects_invented_category():
    cat, conf = enforce_closed_set("Crypto Trading", 0.99)
    assert cat == SUSPENSE
    cat, conf = enforce_closed_set("Office Rent", 0.3)  # below floor
    assert cat == SUSPENSE
    cat, conf = enforce_closed_set("Office Rent", 0.9)
    assert cat == "Office Rent"
    assert cat in CATEGORY_SET


# 6. Tally XML --------------------------------------------------------------- #
def test_tally_xml_wellformed_and_balanced(truth):
    from lxml import etree
    txns = [dict(t, category="Suspense / Unclassified",
                 voucher_type=("Payment" if t["debit"] else "Receipt"))
            for t in truth]
    xml = generate_tally_xml(txns)
    root = etree.fromstring(xml.encode("utf-8"))
    vouchers = root.findall(".//VOUCHER")
    assert len(vouchers) == len(truth)
    for v in vouchers:
        amounts = [float(a.text) for a in v.findall(".//AMOUNT")]
        assert len(amounts) == 2
        assert abs(sum(amounts)) < 0.01, "debit/credit must net to zero per voucher"


# 7. Excel ------------------------------------------------------------------- #
def test_excel_export(truth, tmp_path):
    from openpyxl import load_workbook
    txns = [dict(t, category="Suspense / Unclassified") for t in truth]
    out = os.path.join(tmp_path, "out.xlsx")
    write_excel(txns, out)
    wb = load_workbook(out)
    ws = wb.active
    # header + 40 rows + totals row
    assert ws.max_row == 1 + len(truth) + 1
    total_debit = sum(t["debit"] or 0 for t in truth)
    # totals row, Debit column (4)
    assert abs(ws.cell(row=ws.max_row, column=4).value - round(total_debit, 2)) < 0.01