""" Month-end controls: each one must actually FAIL when its defect is present. A control that only ever passes is worse than no control — it is a green light wired on. The previous "Reconciled" status was exactly that (see core/controls.py), so every test here introduces the real defect and asserts the close is blocked and no figure is published. """ from __future__ import annotations import os import tempfile import pytest from app.core import controls from app.core.money import USD, CurrencyMismatch, Total from app.core.settlements import aggregate _TMP = tempfile.mkdtemp() # --------------------------------------------------------------------- money / C6 def test_total_refuses_to_mix_currencies(): t = Total() t.add(100.0, "EUR") with pytest.raises(CurrencyMismatch): t.add(100.0, "GBP") def test_total_converts_to_usd(): t = Total(USD) t.add_converted(230089.62, 1.185665) t.add_converted(661888.13, 1.368908) assert t.value == round(230089.62 * 1.185665 + 661888.13 * 1.368908, 2) def test_c6_detects_a_local_currency_sum(): """The Jan-2026 defect: locals summed (2,278,406.86) vs converted (2,723,065.30).""" r = controls.c6_currency_integrity(2_278_406.86, 2_723_065.30) assert r.status == controls.FAIL and r.blocking assert "444,658.44" in r.detail def test_c6_passes_when_surfaces_agree(): assert controls.c6_currency_integrity(2_723_065.30, 2_723_065.30).status == controls.PASS # --------------------------------------------------------------------- C1 class _Meta: def __init__(self, name, last_row, header_row, imported, helper=0, blank=0): self.filename, self.sheet_last_row, self.header_row = name, last_row, header_row self.imported_rows, self.helper_rows_skipped, self.blank_rows_skipped = ( imported, helper, blank) @property def expected_data_rows(self): return max(0, self.sheet_last_row - self.header_row) @property def rows_accounted_for(self): return self.imported_rows + self.helper_rows_skipped + self.blank_rows_skipped def test_c1_fails_when_rows_go_missing(): """A file that silently stops parsing half-way is invisible to a self-referential identity.""" r = controls.c1_source_row_count([_Meta("half.xlsx", 1008, 8, imported=500)]) assert r.status == controls.FAIL and r.blocking assert "500 unexplained" in r.evidence[0] def test_c1_passes_when_every_row_is_accounted_for(): metas = [_Meta("ok.xlsx", 1008, 8, imported=990, helper=1, blank=9)] assert controls.c1_source_row_count(metas).status == controls.PASS def test_c1_accepts_a_genuinely_empty_file(): """Turkey Jan-2026: dimension A1:T7, header row 7, zero data rows — empty, not broken.""" assert controls.c1_source_row_count( [_Meta("Turkey.xlsx", 7, 7, imported=0)]).status == controls.PASS # --------------------------------------------------------------------- C2 def _journal(lines_total: float) -> dict: return {"marketplace": "USA", "lines": [{"key": "Sales", "total": lines_total}]} def test_c2_fails_when_an_amount_column_is_unmapped(): """The Australia `fulfilment by amazon fees` class of bug: a column missing from the GL.""" r = controls.c2_column_completeness(_journal(1000.0), uploaded_total=1002_427.43) assert r.status == controls.FAIL and r.blocking def test_c2_passes_when_lines_reconcile_to_total(): assert controls.c2_column_completeness( _journal(1000.0), uploaded_total=1000.0).status == controls.PASS # --------------------------------------------------------------------- C3 def _rec(**kw): base = {"settlement_id": "100", "txn_type": "Order", "account_type": "Standard Orders", "total": 10.0, "_date": None, "_type_en": "Order", "_marketplace": "USA"} base.update(kw) return base def test_c3_fails_on_an_unrecognized_account_type(): agg = aggregate([_rec(account_type="Mystery Orders", total=5000.0)]) r = controls.c3_bucket_completeness(agg) assert r.status == controls.FAIL and r.blocking assert "5,000.00" in r.detail def test_c3_fails_on_a_non_numeric_settlement_id(): agg = aggregate([_rec(settlement_id="", total=1234.56)]) r = controls.c3_bucket_completeness(agg) assert r.status == controls.FAIL and r.blocking assert "1,234.56" in r.detail def test_c3_passes_on_clean_rows(): agg = aggregate([_rec(), _rec(settlement_id="101", account_type="Invoiced Orders")]) assert controls.c3_bucket_completeness(agg).status == controls.PASS def test_c3_ignores_transfers_without_an_account_type(): """Amazon leaves account type blank on payouts everywhere except the USA — that's normal.""" agg = aggregate([_rec(txn_type="Transfer", _type_en="Transfer", account_type="", total=-9999.0)]) assert controls.c3_bucket_completeness(agg).status == controls.PASS # --------------------------------------------------------------------- C5 class _Fx: def __init__(self, marketplace, currency, rate, confirmed_by="", confirmed_month="", source="default (Jan-26 workbook)"): self.marketplace, self.currency, self.rate = marketplace, currency, rate self.confirmed_by, self.confirmed_month, self.source = ( confirmed_by, confirmed_month, source) def test_c5_blocks_a_seeded_default_rate(): r = controls.c5_fx_confirmed([_Fx("Germany", "EUR", 1.185665)], "2026-07", ["Germany"]) assert r.status == controls.FAIL and r.blocking assert "not confirmed" in r.evidence[0] def test_c5_blocks_a_rate_confirmed_for_another_month(): """The stale-FX defect: January's rate silently valuing a July close.""" fx = [_Fx("Germany", "EUR", 1.185665, confirmed_by="cfo", confirmed_month="2026-01")] r = controls.c5_fx_confirmed(fx, "2026-07", ["Germany"]) assert r.status == controls.FAIL assert "2026-01" in r.evidence[0] def test_c5_passes_when_confirmed_for_this_month(): fx = [_Fx("Germany", "EUR", 1.16, confirmed_by="cfo", confirmed_month="2026-07")] assert controls.c5_fx_confirmed(fx, "2026-07", ["Germany"]).status == controls.PASS def test_c5_does_not_ask_anyone_to_confirm_usd_at_parity(): fx = [_Fx("USA", "USD", 1.0)] assert controls.c5_fx_confirmed(fx, "2026-07", ["USA"]).status == controls.PASS # --------------------------------------------------------------------- blocking def test_only_error_severity_blocks(): warn = controls.c4_dual_method([ {"marketplace": "USA", "currency": "USD", "settlement_closing": 100.0, "closing_local": 900.0}, ]) assert warn.status == controls.FAIL and not warn.blocking # warning, not a block assert controls.blocking_summary([warn]) == "" def test_blocking_summary_names_the_failed_control(): bad = controls.c6_currency_integrity(1.0, 2.0) assert "C6" in controls.blocking_summary([bad]) # --------------------------------------------------------------------- advertising def test_advertising_detected_from_description_not_type(): """ Amazon books advertising as type "Service Fee" with only the DESCRIPTION saying "Cost of advertising". Canada Jan-2026 had CA$36,948.43 of it sitting in "Other service charges" with the Advertising line at 0.00 because the old check looked at the type only. """ from app.core.journal import _contribute_components, COMPONENT_KEYS comp = {k: 0.0 for k in COMPONENT_KEYS} _contribute_components( {"txn_type": "Service Fee", "description": "Cost of advertising", "other": -36948.43, "total": -36948.43}, comp, "Service Fee") assert comp["advertising"] == -36948.43 assert comp["other_service_charges"] == 0.0 def test_localized_advertising_descriptions_route_to_advertising(): from app.core.i18n import is_advertising_like for desc in ("Cost of advertising", "Werbekosten", "Coût de la publicité", "Costo della pubblicità", "Coste de publicidad", "Sponsored Products charge"): assert is_advertising_like("Service Fee", desc), desc # a plain service fee must NOT be classified as advertising assert not is_advertising_like("Service Fee", "Subscription fee") assert not is_advertising_like("Service Fee", None) def test_uk_advertising_in_other_transaction_fees_column(): """UK books "Cost of Advertising" in `other transaction fees` (UK Jan-26: -153,176.09), not in `other` like North America — both columns must route to Advertising.""" from app.core.journal import _contribute_components, COMPONENT_KEYS comp = {k: 0.0 for k in COMPONENT_KEYS} _contribute_components( {"txn_type": "Service Fee", "description": "Cost of Advertising", "other_transaction_fees": -153176.09, "total": -153176.09}, comp, "Service Fee") assert comp["advertising"] == -153176.09 assert comp["other_transaction_fees"] == 0.0 # …and a NON-advertising row keeps its other_transaction_fees where they belong. comp2 = {k: 0.0 for k in COMPONENT_KEYS} _contribute_components( {"txn_type": "Order", "description": "some product", "other_transaction_fees": -10.0, "total": -10.0}, comp2, "Order") assert comp2["other_transaction_fees"] == -10.0 assert comp2["advertising"] == 0.0 # --------------------------------------------------------------------- definitions def test_every_component_has_a_definition(): """The (i) buttons must cover every line the Finance Summary can show. A new component without a definition ships an unexplained number — fail here instead.""" from app.core.definitions import DEFINITIONS from app.core.journal import COMPONENT_KEYS missing = [k for k in COMPONENT_KEYS if k not in DEFINITIONS] assert not missing, f"components with no (i) definition: {missing}" # …and the aggregate/tab-level figures the UI explains. for key in ("gross_revenue", "net_revenue", "opening_balance", "closing_receivable", "settlement_closing", "disbursements", "in_transit_payouts", "closing_receivable_usd", "receivable_orders", "paid_orders", "transfers_total", "aging_basis", "settlement_status", "uploaded_total"): assert key in DEFINITIONS, key for k, d in DEFINITIONS.items(): assert d.get("formula") and d.get("source"), f"{k} definition is incomplete"