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

feat(import): T6 gate auto_send pe coduri nou-mapate (OV-1)

Browse Source 
- load_mapping_meta: {cod_op_service -> {cod_prestatie, auto_send}}
- has_no_auto_send: verifica daca vreun item rezolvat via mapping are auto_send=0
- reresolve_account: auto_send=0 -> ramane needs_mapping (review_manual stat),
  NU trece pe queued; previne FINALIZATA eronat permanent
- reresolve_account primeste batch_id optional (pregatire T7, urmeaza)
- POST /v1/prezentari: auto_send=0 -> needs_mapping + motiv explicit
- 9 teste: load_mapping_meta, has_no_auto_send, reresolve (zero/unu), POST API

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Agent
2026-06-16 20:21:32 +00:00
parent 12f0ca3a81
commit 61a7b4ea1c
3 changed files with 282 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
app/api/v1/router.py
View File

@@ -25,7 +25,8 @@ from ...db import get_connection
from ...idempotency import build_key, canonicalize_row, idempotency_key
from ...mapping import (
account_or_default,
load_mapping,
has_no_auto_send,
load_mapping_meta,
pending_unmapped,
reresolve_account,
resolve_prestatii,
@@ -60,7 +61,9 @@ def create_prezentari(
conn = get_connection()
results: list[SubmissionResult] = []
try:
mapping = load_mapping(conn, acct)
# T6/OV-1: load_mapping_meta include auto_send per op (gate pentru coduri noi).
mapping_meta = load_mapping_meta(conn, acct)
mapping = {op: meta["cod_prestatie"] for op, meta in mapping_meta.items()}
for prez in req.prezentari:
content = prez.model_dump()
# T9/OV-2: canonicalize_row inaintea build_key (odometru strip ".0", VIN upper).
@@ -104,6 +107,14 @@ def create_prezentari(
errors = validate_prezentare(content)
if errors:
status, rar_error = "needs_data", json.dumps(errors, ensure_ascii=False)
elif has_no_auto_send(resolved, mapping_meta):
# T6/OV-1: cod rezolvat cu auto_send=0 -> nu trimite automat.
# Randul ramane 'needs_mapping' pana userul confirma manual (sau comuta auto_send=1).
status = "needs_mapping"
rar_error = json.dumps(
{"auto_send": "cod mapat cu auto_send=0; review manual inainte de trimitere"},
ensure_ascii=False,
)
else:
status, rar_error = "queued", None

75
app/mapping.py
View File

@@ -188,6 +188,35 @@ def load_mapping(conn, account_id: int | None) -> dict[str, str]:
return {r["cod_op_service"]: r["cod_prestatie"] for r in rows}
def load_mapping_meta(conn, account_id: int | None) -> dict[str, dict]:
"""{cod_op_service -> {cod_prestatie, auto_send}} pentru un cont.
T6/OV-1: varianta extinsa care include si flagul auto_send per operatie.
"""
acct = account_or_default(account_id)
rows = conn.execute(
"SELECT cod_op_service, cod_prestatie, auto_send FROM operations_mapping WHERE account_id=?",
(acct,),
).fetchall()
return {
r["cod_op_service"]: {"cod_prestatie": r["cod_prestatie"], "auto_send": bool(r["auto_send"])}
for r in rows
}
def has_no_auto_send(resolved: list[dict], mapping_meta: dict[str, dict]) -> bool:
"""Verifica daca vreun item rezolvat via mapping are auto_send=0.
T6/OV-1: un cod nou-mapat cu auto_send=0 nu trebuie trimis automat.
Items cu cod_prestatie direct (nu via cod_op_service) nu sunt afectate.
"""
for item in resolved:
op = (item.get("cod_op_service") or "").strip()
if op and op in mapping_meta and not mapping_meta[op]["auto_send"]:
return True
return False
def pending_unmapped(conn) -> list[dict]:
"""Operatii distincte nemapate, agregate din submission-urile `needs_mapping`.
@@ -250,22 +279,39 @@ def save_mapping(conn, account_id: int | None, cod_op_service: str, cod_prestati
)
def reresolve_account(conn, account_id: int | None) -> dict[str, int]:
def reresolve_account(conn, account_id: int | None, batch_id: int | None = None) -> dict[str, int]:
"""Re-rezolva submission-urile `needs_mapping` ale unui cont dupa o noua mapare.
Pentru fiecare: aplica maparea curenta; daca nu mai raman op-uri nemapate ->
ruleaza validarea de continut (T3) si trece pe `queued` (sau `needs_data` cu
motiv), resetand backoff-ul. Daca raman nemapate, ramane `needs_mapping` cu
motivul actualizat. Intoarce {requeued, still_blocked, needs_data}.
motivul actualizat. Intoarce {requeued, still_blocked, needs_data, review_manual}.
T6/OV-1: auto_send=0 pe un cod nou-mapat -> nu trece pe 'queued' (ramane
'needs_mapping' cu motiv "review manual"); previne FINALIZATA eronat permanent.
T7: batch_id != None -> scope la seria comitata (NU cross-batch).
batch_id is None -> re-rezolva toti (canal API, batch_id IS NULL inclus).
"""
acct = account_or_default(account_id)
mapping = load_mapping(conn, acct)
rows = conn.execute(
"SELECT id, payload_json FROM submissions WHERE status='needs_mapping' AND account_id=?",
(acct,),
).fetchall()
mapping_meta = load_mapping_meta(conn, acct)
mapping = {op: meta["cod_prestatie"] for op, meta in mapping_meta.items()}
stats = {"requeued": 0, "still_blocked": 0, "needs_data": 0}
if batch_id is not None:
# T7: scope la batch-ul specificat
rows = conn.execute(
"SELECT id, payload_json FROM submissions "
"WHERE status='needs_mapping' AND account_id=? AND batch_id=?",
(acct, batch_id),
).fetchall()
else:
# Canal API (batch_id IS NULL) + legacy (batch_id nesetat)
rows = conn.execute(
"SELECT id, payload_json FROM submissions WHERE status='needs_mapping' AND account_id=?",
(acct,),
).fetchall()
stats = {"requeued": 0, "still_blocked": 0, "needs_data": 0, "review_manual": 0}
for r in rows:
try:
content = json.loads(r["payload_json"])
@@ -283,6 +329,19 @@ def reresolve_account(conn, account_id: int | None) -> dict[str, int]:
stats["still_blocked"] += 1
continue
# T6/OV-1: verifica auto_send inainte de re-queuing
if has_no_auto_send(resolved, mapping_meta):
conn.execute(
"UPDATE submissions SET payload_json=?, rar_error=?, updated_at=datetime('now') WHERE id=?",
(
payload_json,
json.dumps({"auto_send": "cod mapat cu auto_send=0; review manual inainte de trimitere"}, ensure_ascii=False),
r["id"],
),
)
stats["review_manual"] += 1
continue
errors = validate_prezentare(content)
if errors:
conn.execute(

202
tests/test_t6_auto_send.py Normal file
View File

@@ -0,0 +1,202 @@
"""Teste T6: gate auto_send pe coduri nou-mapate (OV-1).
Verify:
(a) cod nou-mapat cu auto_send=0 -> nu auto-send, review manual.
(b) REGRESIE: mapare existenta cu auto_send=1 tot se requeue ca azi.
"""
from __future__ import annotations
import json
import os
import tempfile
import pytest
@pytest.fixture()
def env(monkeypatch):
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t6.db"))
from app.config import get_settings
get_settings.cache_clear()
from app.db import init_db
init_db()
yield monkeypatch
get_settings.cache_clear()
@pytest.fixture()
def conn(env):
from app.db import get_connection
c = get_connection()
yield c
c.close()
@pytest.fixture()
def client(env):
from app.main import app
from fastapi.testclient import TestClient
with TestClient(app) as c:
yield c
def _insert_needs_mapping(conn, account_id=1, cod_op="ITP-CHECK"):
content = {
"vin": "WVWZZZ1KZAW000123", "nr_inmatriculare": "B1",
"data_prestatie": "2026-06-15", "odometru_final": "123456",
"prestatii": [{"cod_op_service": cod_op, "denumire": "Inspectie tehnica"}],
}
cur = conn.execute(
"INSERT INTO submissions (idempotency_key, account_id, status, payload_json) VALUES (?, ?, ?, ?)",
(f"k-{os.urandom(4).hex()}", account_id, "needs_mapping", json.dumps(content)),
)
return int(cur.lastrowid)
def _add_mapping(conn, account_id=1, cod_op="ITP-CHECK", cod_prestatie="OE-1", auto_send=True):
conn.execute(
"INSERT OR REPLACE INTO nomenclator_rar (cod_prestatie, nume_prestatie) VALUES (?, ?)",
(cod_prestatie, "Operatie test"),
)
conn.execute(
"INSERT OR REPLACE INTO operations_mapping (account_id, cod_op_service, cod_prestatie, auto_send) "
"VALUES (?, ?, ?, ?)",
(account_id, cod_op, cod_prestatie, 1 if auto_send else 0),
)
# --- load_mapping_meta ---
def test_load_mapping_meta_returns_auto_send(conn):
from app.mapping import load_mapping_meta
_add_mapping(conn, cod_op="ITP-1", cod_prestatie="OE-1", auto_send=True)
_add_mapping(conn, cod_op="ITP-2", cod_prestatie="OE-2", auto_send=False)
conn.execute("INSERT OR IGNORE INTO nomenclator_rar (cod_prestatie, nume_prestatie) VALUES ('OE-2', 'Test2')")
meta = load_mapping_meta(conn, 1)
assert meta["ITP-1"]["auto_send"] is True
assert meta["ITP-2"]["auto_send"] is False
# --- has_no_auto_send ---
def test_has_no_auto_send_detecteaza_false(conn):
from app.mapping import has_no_auto_send
mapping_meta = {
"ITP-1": {"cod_prestatie": "OE-1", "auto_send": False},
}
resolved = [{"cod_op_service": "ITP-1", "cod_prestatie": "OE-1"}]
assert has_no_auto_send(resolved, mapping_meta) is True
def test_has_no_auto_send_trece_cu_true(conn):
from app.mapping import has_no_auto_send
mapping_meta = {
"ITP-1": {"cod_prestatie": "OE-1", "auto_send": True},
}
resolved = [{"cod_op_service": "ITP-1", "cod_prestatie": "OE-1"}]
assert has_no_auto_send(resolved, mapping_meta) is False
def test_has_no_auto_send_direct_cod_prestatie(conn):
"""Item cu cod_prestatie direct (fara cod_op_service) nu e afectat de auto_send."""
from app.mapping import has_no_auto_send
mapping_meta = {}
resolved = [{"cod_prestatie": "OE-1"}]
assert has_no_auto_send(resolved, mapping_meta) is False
# --- reresolve_account cu auto_send=0 ---
def test_reresolve_auto_send_zero_nu_requeue(conn):
"""(a) cod nou-mapat cu auto_send=0 -> ramane needs_mapping (nu trece pe queued)."""
from app.mapping import reresolve_account
sid = _insert_needs_mapping(conn, cod_op="ITP-CHECK")
_add_mapping(conn, cod_op="ITP-CHECK", cod_prestatie="OE-1", auto_send=False)
stats = reresolve_account(conn, 1)
assert stats["review_manual"] == 1
assert stats["requeued"] == 0
row = conn.execute("SELECT status, rar_error FROM submissions WHERE id=?", (sid,)).fetchone()
assert row["status"] == "needs_mapping"
err = json.loads(row["rar_error"])
assert "auto_send" in err
def test_reresolve_auto_send_unu_requeue(conn):
"""(b) REGRESIE: mapare cu auto_send=1 tot se requeue ca azi."""
from app.mapping import reresolve_account
sid = _insert_needs_mapping(conn, cod_op="ITP-CHECK")
_add_mapping(conn, cod_op="ITP-CHECK", cod_prestatie="OE-1", auto_send=True)
stats = reresolve_account(conn, 1)
assert stats["requeued"] == 1
assert stats["review_manual"] == 0
row = conn.execute("SELECT status FROM submissions WHERE id=?", (sid,)).fetchone()
assert row["status"] == "queued"
# --- POST /v1/prezentari cu auto_send=0 ---
def _body_with_op(cod_op="ITP-CHECK"):
return {
"rar_credentials": {"email": "x@y.ro", "password": "s"},
"prezentari": [{
"vin": "WVWZZZ1KZAW000123",
"nr_inmatriculare": "B999TST",
"data_prestatie": "2026-06-15",
"odometru_final": "123456",
"prestatii": [{"cod_op_service": cod_op, "denumire": "Test"}],
}],
}
def test_post_auto_send_zero_nu_queued(client, env):
"""(a) Via API: cod nou-mapat cu auto_send=0 -> nu 'queued', review manual."""
from app.db import get_connection
conn2 = get_connection()
try:
_add_mapping(conn2, cod_op="ITP-X", cod_prestatie="OE-1", auto_send=False)
finally:
conn2.close()
r = client.post("/v1/prezentari", json=_body_with_op("ITP-X"))
assert r.status_code == 200
status = r.json()["results"][0]["status"]
assert status != "queued", f"auto_send=0 nu trebuie sa fie queued, e: {status}"
assert status == "needs_mapping"
def test_post_auto_send_unu_queued(client, env):
"""(b) REGRESIE: mapare existenta cu auto_send=1 -> queued ca azi."""
from app.db import get_connection
conn2 = get_connection()
try:
_add_mapping(conn2, cod_op="ITP-Y", cod_prestatie="OE-1", auto_send=True)
finally:
conn2.close()
r = client.post("/v1/prezentari", json=_body_with_op("ITP-Y"))
assert r.status_code == 200
status = r.json()["results"][0]["status"]
assert status == "queued", f"auto_send=1 trebuie queued, e: {status}"
def test_post_cod_prestatie_direct_queued(client):
"""Cod RAR direct (fara cod_op_service) -> queued indiferent de mapping."""
body = {
"rar_credentials": {"email": "x@y.ro", "password": "s"},
"prezentari": [{
"vin": "WVWZZZ1KZAW000123",
"nr_inmatriculare": "B999TST",
"data_prestatie": "2026-06-15",
"odometru_final": "123456",
"prestatii": [{"cod_prestatie": "OE-1"}],
}],
}
r = client.post("/v1/prezentari", json=body)
assert r.status_code == 200
assert r.json()["results"][0]["status"] == "queued"