romfast/gomag-vending
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sync): gate CUI invalid/ANAF-notFound → ERROR inainte de import Oracle

Browse Source 
Incident 22.04.2026 (#485225171 NONA ROYAL SRL): clientul a inversat
cod_fiscal cu registru in GoMag → sistem a creat partener cu CUI=J1994000194225.

Adauga evaluate_cui_gate() care blocheaza comanda (ERROR) daca:
- CUI format invalid (ex: J.. in loc de cifre)
- CUI nu trece cifra de control
- ANAF returneaza explicit notFound (scpTVA=None + denumire_anaf="")

ANAF down (anaf_data=None) → fallback pass, comportament existent pastrat.
_record_order_error() DRY helper evita duplicarea upsert/add_items.
Contract ANAF down/notFound/found documentat in anaf_service._call_anaf_api.
9 teste unit (inclusiv T5 CRITIC: ANAF down nu blocheaza) + T7 COALESCE.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Agent
2026-04-22 09:40:07 +00:00
parent 7e30523242
commit 6620b28ed1
3 changed files with 351 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

256
api/tests/test_sync_cui_gate.py Normal file
View File

@@ -0,0 +1,256 @@
"""
CUI Gate Tests
==============
Unit tests for evaluate_cui_gate() and _record_order_error() in sync_service.
Tests 1-6: pure predicate, no IO.
Test 7: integration — _record_order_error with pre-seeded SQLite IMPORTED row
verifies COALESCE preserves existing id_partener.
Run:
cd api && python -m pytest tests/test_sync_cui_gate.py -v
"""
import os
import sys
import tempfile
import pytest
pytestmark = pytest.mark.unit
_tmpdir = tempfile.mkdtemp()
os.environ.setdefault("FORCE_THIN_MODE", "true")
os.environ.setdefault("SQLITE_DB_PATH", os.path.join(_tmpdir, "test_cui_gate.db"))
os.environ.setdefault("ORACLE_DSN", "dummy")
os.environ.setdefault("ORACLE_USER", "dummy")
os.environ.setdefault("ORACLE_PASSWORD", "dummy")
os.environ.setdefault("JSON_OUTPUT_DIR", _tmpdir)
_api_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _api_dir not in sys.path:
sys.path.insert(0, _api_dir)
from app import database
from app.services import sqlite_service
from app.services.sync_service import evaluate_cui_gate, _record_order_error
from app.services.order_reader import OrderBilling, OrderShipping, OrderData, OrderItem
from app.constants import OrderStatus
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_VALID_ANAF_FOUND = {"scpTVA": True, "denumire_anaf": "NONA ROYAL SRL", "checked_at": "2026-04-22T10:00:00"}
_ANAF_NOT_FOUND = {"scpTVA": None, "denumire_anaf": "", "checked_at": "2026-04-22T10:00:00"}
# A CUI with valid format and valid checksum (MATTEO&OANA CAFFE 2022 SRL)
_VALID_CUI = "49033051"
# Same body but last digit modified → fails checksum
_BAD_CHECKSUM_CUI = "49033052"
# J-format — the incident CUI (registru number in the CUI field)
_J_FORMAT = "J1994000194225"
def _make_pj_order(company_code=_VALID_CUI, number="O-001"):
billing = OrderBilling(
firstname="Ion", lastname="Pop", phone="0700", email="x@x.ro",
address="Str A 1", city="Cluj", region="Cluj", country="Romania",
company_name="TEST SRL", company_code=company_code,
company_reg="J12/123/2020", is_company=True,
)
shipping = OrderShipping(
firstname="Ion", lastname="Pop", phone="0700", email="x@x.ro",
address="Str A 1", city="Cluj", region="Cluj", country="Romania",
)
return OrderData(
id=number, number=number, date="2026-04-22",
billing=billing, shipping=shipping,
items=[OrderItem(sku="SKU1", name="Prod", price=10.0, quantity=1, vat=19)],
)
def _make_pf_order(number="O-PF-1"):
billing = OrderBilling(
firstname="Ana", lastname="Pop", phone="0700", email="a@x.ro",
address="Str B 2", city="Iasi", region="Iasi", country="Romania",
is_company=False,
)
shipping = OrderShipping(
firstname="Ana", lastname="Pop", phone="0700", email="a@x.ro",
address="Str B 2", city="Iasi", region="Iasi", country="Romania",
)
return OrderData(
id=number, number=number, date="2026-04-22",
billing=billing, shipping=shipping,
items=[OrderItem(sku="SKU1", name="Prod", price=10.0, quantity=1, vat=19)],
)
# ---------------------------------------------------------------------------
# Tests 1-6: pure predicate — no IO
# ---------------------------------------------------------------------------
class TestEvaluateCuiGate:
def test_format_invalid_incident_case(self):
"""Test 1: J-format in cod_fiscal field (the 22-Apr-2026 incident) → blocked."""
# bare_cui from sanitize_cui("J1994000194225") = "J1994000194225" (not digits)
result = evaluate_cui_gate(
is_ro_company=True,
company_code_raw=_J_FORMAT,
bare_cui=_J_FORMAT,
anaf_data=_VALID_ANAF_FOUND,
)
assert result is not None
assert "format" in result
def test_checksum_invalid(self):
"""Test 2: valid format, wrong check digit → blocked."""
result = evaluate_cui_gate(
is_ro_company=True,
company_code_raw=_BAD_CHECKSUM_CUI,
bare_cui=_BAD_CHECKSUM_CUI,
anaf_data=_VALID_ANAF_FOUND,
)
assert result is not None
assert "cifra de control" in result
def test_anaf_not_found_explicit(self):
"""Test 3: ANAF explicit notFound → blocked with registry hint."""
result = evaluate_cui_gate(
is_ro_company=True,
company_code_raw=_VALID_CUI,
bare_cui=_VALID_CUI,
anaf_data=_ANAF_NOT_FOUND,
)
assert result is not None
assert "nu exista in registrul ANAF" in result
assert "registrul comertului" in result
def test_anaf_found_vat_payer_passes(self):
"""Test 4: ANAF found + platitor TVA → pass."""
result = evaluate_cui_gate(
is_ro_company=True,
company_code_raw=_VALID_CUI,
bare_cui=_VALID_CUI,
anaf_data=_VALID_ANAF_FOUND,
)
assert result is None
def test_anaf_down_fallback_passes(self):
"""Test 5 [CRITICAL REGRESSION]: ANAF down (anaf_data=None) + valid CUI → pass.
If this test fails, the gate is breaking the ANAF-down fallback and all
RO company orders would error when ANAF is unavailable.
"""
result = evaluate_cui_gate(
is_ro_company=True,
company_code_raw=_VALID_CUI,
bare_cui=_VALID_CUI,
anaf_data=None, # ANAF down / transient error
)
assert result is None, (
"ANAF down must NOT block orders — gate must only block on explicit notFound"
)
def test_pf_always_passes(self):
"""Test 6: PF order (is_ro_company=False) → always pass, regardless of CUI."""
result = evaluate_cui_gate(
is_ro_company=False,
company_code_raw=_J_FORMAT,
bare_cui=_J_FORMAT,
anaf_data=_ANAF_NOT_FOUND,
)
assert result is None
def test_no_company_code_passes(self):
"""PJ without company_code → pass (nothing to validate)."""
result = evaluate_cui_gate(
is_ro_company=True,
company_code_raw=None,
bare_cui="",
anaf_data=None,
)
assert result is None
def test_anaf_found_non_vat_passes(self):
"""ANAF found non-platitor TVA (scpTVA=False) → pass."""
result = evaluate_cui_gate(
is_ro_company=True,
company_code_raw=_VALID_CUI,
bare_cui=_VALID_CUI,
anaf_data={"scpTVA": False, "denumire_anaf": "FIRMA SRL", "checked_at": "2026-04-22T10:00:00"},
)
assert result is None
# ---------------------------------------------------------------------------
# Test 7: integration — COALESCE preserves id_partener on gate block
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _init_db():
database.init_sqlite()
@pytest.mark.asyncio
async def test_record_order_error_preserves_id_partener():
"""Test 7: _record_order_error called with id_partener=None preserves existing id_partener
via SQLite COALESCE in upsert_order.
Scenario: order was previously IMPORTED with id_partener=9001.
At resync the gate blocks it (bad CUI). _record_order_error passes id_partener=None.
After upsert, the row should have status=ERROR and id_partener=9001 (preserved).
"""
order = _make_pj_order(company_code=_J_FORMAT, number="O-COALESCE-1")
run_id = "test-run-coalesce"
# Seed an existing IMPORTED row with id_partener=9001
db = await sqlite_service.get_sqlite()
try:
await db.execute(
"""INSERT OR REPLACE INTO orders
(order_number, order_date, customer_name, status, id_partener, items_count)
VALUES (?, ?, ?, ?, ?, ?)""",
(order.number, order.date, "TEST SRL", OrderStatus.IMPORTED.value, 9001, 1),
)
await db.commit()
finally:
await db.close()
# Gate fires → calls _record_order_error with id_partener=None (gate doesn't know it)
await _record_order_error(
run_id=run_id,
order=order,
customer="TEST SRL",
shipping_name="Ion Pop",
billing_name="TEST SRL",
payment_method="card",
delivery_method="curier",
discount_split_json=None,
order_items_data=[{
"sku": "SKU1", "product_name": "Prod", "quantity": 1,
"price": 10.0, "baseprice": None, "vat": 19,
"mapping_status": "direct", "codmat": None, "id_articol": None, "cantitate_roa": None,
}],
reason=f"CUI invalid (format): {_J_FORMAT!r}",
id_partener=None, # gate doesn't have it
)
# Verify: status=ERROR, id_partener=9001 (COALESCE preserved), error_message populated
db = await sqlite_service.get_sqlite()
try:
row = await db.execute(
"SELECT status, id_partener, error_message FROM orders WHERE order_number = ?",
(order.number,),
)
row = await row.fetchone()
finally:
await db.close()
assert row is not None, "Order row missing after _record_order_error"
assert row[0] == OrderStatus.ERROR.value, f"Expected ERROR, got {row[0]}"
assert row[1] == 9001, f"Expected id_partener=9001 (preserved by COALESCE), got {row[1]}"
assert row[2] and "format" in row[2], f"Expected error_message with 'format', got {row[2]!r}"