"""US-004: persistarea `exclus` prin salvare (import + Trimiteri), round-trip complet. Acopera: - resolve_prestatii: item-exclus EXPLICIT supravietuieste re-rezolvarii (treapta noua de precedenta) FARA sa strice restul precedentei (IRON RULE, regresiune-cheie). - Import (web_editeaza_rand / apply_row_override): `exclus` persistat in override_json, chip Nedeclarat reapare la redeschiderea modalului, x il readuce la warning. - Rand cu toate operatiile excluse -> stare preview 'excluded' ("Nedeclarat"). - Trimiteri (post_corectie_trimitere): acelasi round-trip pe chips_state; toate excluse -> needs_data cu motiv explicit. - Sentinelul __NEDECLARAT__ NU ajunge in payload daca apare direct ca cod_prestatie in starea hidden (nu doar ales din select), pe AMBELE cai. """ from __future__ import annotations import json import os import re import tempfile import pytest from starlette.testclient import TestClient # --------------------------------------------------------------------------- # # Test unitar (fara client): precedenta resolve_prestatii # # --------------------------------------------------------------------------- # def test_exclus_supravietuieste_resolve_chiar_cu_mapare_pe_cont(): """IRON RULE: item cu `exclus=1` explicit pe item RAMANE exclus dupa resolve_prestatii, chiar daca operatia are (a) o mapare exacta pe cont SAU (b) un cod_prestatie deja valid pe rand. Regresiune-cheie: restul precedentei (cod valid > exclus de cont > mapare exacta > reguli text > nemapat) trebuie sa ramana IDENTICA pentru itemii FARA `exclus` explicit — altfel stricam maparile existente. """ from app.mapping import resolve_prestatii valid_codes = {"OE-1", "IG-1", "FRN1"} mapping = {"Schimb-Ulei": "OE-1"} # mapare exacta pe cont pentru Schimb-Ulei excluded_ops_cont = {"Filtru-Aer"} # regula de excludere PE CONT (alt mecanism) prestatii = [ # (a) exclus explicit + mapare exacta pe cont exista -> exclus CASTIGA {"cod_op_service": "Schimb-Ulei", "denumire": "Schimb ulei motor", "exclus": 1}, # (b) exclus explicit + cod_prestatie deja valid pe rand -> exclus CASTIGA {"cod_op_service": "Frana", "denumire": "Reparatie frana", "cod_prestatie": "FRN1", "exclus": 1}, # regresiune: FARA exclus explicit, mapare exacta trebuie sa functioneze ca inainte {"cod_op_service": "Schimb-Ulei", "denumire": "Schimb ulei motor 2"}, # regresiune: FARA exclus explicit, regula de excludere PE CONT trebuie sa functioneze {"cod_op_service": "Filtru-Aer", "denumire": "Filtru aer"}, # regresiune: cod deja valid pe rand, FARA exclus -> pastrat ca atare (cod_valid) {"cod_op_service": "", "denumire": "", "cod_prestatie": "IG-1"}, # regresiune: nemapat ramane nemapat {"cod_op_service": "Necunoscut-Op", "denumire": "Operatie necunoscuta"}, ] resolved, unmapped = resolve_prestatii( prestatii, mapping, valid_codes=valid_codes, text_rules=None, excluded_ops=excluded_ops_cont, ) item_a = resolved[0] assert item_a.get("exclus") is True, f"exclus explicit trebuia sa supravietuiasca: {item_a}" assert not item_a.get("cod_prestatie"), f"item exclus explicit nu trebuie sa aiba cod_prestatie: {item_a}" item_b = resolved[1] assert item_b.get("exclus") is True, f"exclus explicit trebuia sa domine cod_prestatie valid: {item_b}" assert not item_b.get("cod_prestatie"), f"cod_prestatie valid trebuia sters de exclus explicit: {item_b}" # Regresiune: mapare exacta pe cont NESCHIMBATA pentru itemul fara exclus item_c = resolved[2] assert item_c.get("cod_prestatie") == "OE-1", f"mapare exacta pe cont stricata: {item_c}" assert not item_c.get("exclus"), f"item fara exclus explicit nu trebuie marcat exclus: {item_c}" # Regresiune: regula de excludere PE CONT (mecanism vechi) NESCHIMBATA item_d = resolved[3] assert item_d.get("exclus") is True, f"regula de excludere pe cont stricata: {item_d}" assert not item_d.get("cod_prestatie") # Regresiune: cod deja valid pastrat ca atare item_e = resolved[4] assert item_e.get("cod_prestatie") == "IG-1", f"cod valid pastrat stricat: {item_e}" # Regresiune: nemapat ramane nemapat assert len(unmapped) == 1 assert unmapped[0]["cod_op_service"] == "Necunoscut-Op" def test_precedenta_neschimbata_pentru_itemi_fara_exclus(): """Regresiune suplimentara: fara `exclus` explicit pe niciun item, comportamentul lui resolve_prestatii e IDENTIC cu cel dinaintea treptei noi (cod valid > exclus de cont > mapare exacta > reguli text > nemapat). """ from app.mapping import resolve_prestatii valid_codes = {"OE-1"} mapping = {"Op-Mapata": "OE-1"} text_rules = [{"pattern": "frana", "cod_prestatie": "OE-1", "auto_send": 1, "priority": 1}] excluded_ops = {"Op-Exclusa"} prestatii = [ {"cod_op_service": "Op-Mapata", "denumire": "Operatie mapata"}, {"cod_op_service": "Op-Exclusa", "denumire": "Operatie exclusa"}, {"cod_op_service": "Op-Text", "denumire": "Reparatie frana fata"}, {"cod_op_service": "Op-Nemapat", "denumire": "Chestie noua"}, ] resolved, unmapped = resolve_prestatii(prestatii, mapping, valid_codes, text_rules, excluded_ops) assert resolved[0]["cod_prestatie"] == "OE-1" assert not resolved[0].get("exclus") assert resolved[1]["exclus"] is True assert not resolved[1].get("cod_prestatie") assert resolved[2]["cod_prestatie"] == "OE-1" # via regula text assert resolved[2].get("cod_sursa") assert len(unmapped) == 1 assert unmapped[0]["cod_op_service"] == "Op-Nemapat" # --------------------------------------------------------------------------- # # Fixtures + helpere HTTP (import + Trimiteri, fara auth — cont 1 implicit) # # --------------------------------------------------------------------------- # @pytest.fixture() def client(monkeypatch): tmp = tempfile.mkdtemp() monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "editare_rand.db")) monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "false") from app.config import get_settings get_settings.cache_clear() from app.crypto import reset_cache reset_cache() from app.main import app with TestClient(app) as c: yield c get_settings.cache_clear() reset_cache() def _seed_cod(cod: str, denumire: str = "Prestatie test") -> None: from app.db import get_connection conn = get_connection() try: conn.execute( "INSERT OR REPLACE INTO nomenclator_rar (cod_prestatie, nume_prestatie) VALUES (?, ?)", (cod, denumire), ) conn.commit() finally: conn.close() def _csv_bytes(rows: list[dict]) -> bytes: import csv as _csv import io buf = io.StringIO() writer = _csv.DictWriter(buf, fieldnames=list(rows[0].keys()), delimiter=";") writer.writeheader() writer.writerows(rows) return buf.getvalue().encode("utf-8") _MAP_COLS = { "VIN": "vin", "Nr": "nr_inmatriculare", "Data": "data_prestatie", "KM": "odometru_final", "Operatie": "operatie", } def _upload_and_preview(client: TestClient, rows: list[dict]) -> int: import io as _io csv_data = _csv_bytes(rows) r = client.post( "/_import/upload", files={"file": ("test.csv", _io.BytesIO(csv_data), "text/csv")}, data={}, ) assert r.status_code == 200, r.text m = re.search(r"/_import/(\d+)/mapare-coloane", r.text) assert m, f"import_id negasit in raspuns: {r.text[:300]}" iid = int(m.group(1)) colnames = list(rows[0].keys()) canons = [_MAP_COLS[c] for c in colnames] r2 = client.post(f"/_import/{iid}/mapare-coloane", data={ "colname": colnames, "canon": canons, "format_data": "YYYY-MM-DD", }) assert r2.status_code == 200, r2.text return iid def _chips_state_din_html(html: str) -> dict: m = re.search(r'name="chips_state"\s+value=\'([^\']*)\'', html) assert m, f"hidden chips_state lipsa din HTML: {html[:400]}" return json.loads(m.group(1).replace(""", '"')) # --- Trimiteri: helpere submissions direct in DB (cont 1, fara auth) --------- # def _insert_submission(*, status: str, payload: dict) -> int: from app.db import get_connection conn = get_connection() try: cur = conn.execute( "INSERT INTO submissions (idempotency_key, account_id, status, payload_json) " "VALUES (?, ?, ?, ?)", (f"k-{os.urandom(6).hex()}", 1, status, json.dumps(payload)), ) conn.commit() return int(cur.lastrowid) finally: conn.close() def _submission_row(sid: int): from app.db import get_connection conn = get_connection() try: return conn.execute("SELECT * FROM submissions WHERE id=?", (sid,)).fetchone() finally: conn.close() def _submission_payload(sid: int) -> dict: from app.db import get_connection conn = get_connection() try: r = conn.execute("SELECT payload_json FROM submissions WHERE id=?", (sid,)).fetchone() return json.loads(r["payload_json"]) finally: conn.close() def _payload_cu_ops(vin: str, ops: list[tuple[str, str]]) -> dict: return { "vin": vin, "nr_inmatriculare": "B100AAA", "data_prestatie": "2026-06-10", "odometru_final": "50000", "prestatii": [{"cod_op_service": op, "denumire": den} for op, den in ops], } # --------------------------------------------------------------------------- # # Import: salvare + round-trip (web_editeaza_rand / apply_row_override) # # --------------------------------------------------------------------------- # _SAMPLE_ROW_1OP = [{ "VIN": "WVWZZZ1KZAW000999", "Nr": "B009TST", "Data": "2026-06-10", "KM": "123456", "Operatie": "Op-Nemapata", }] def test_salvare_pastreaza_exclus_in_override(client): """Salvarea unui chip Nedeclarat in modalul de editare (import) persista `exclus=1` in override_json — la reincarcarea preview-ului randul ramane 'Nedeclarat'.""" iid = _upload_and_preview(client, _SAMPLE_ROW_1OP) chips_state = json.dumps({"v": 1, "items": [ {"cod": "", "op": "Op-Nemapata", "den": "Op-Nemapata", "exclus": 1}, ]}) r = client.post(f"/_import/{iid}/rand/0/editeaza", data={"chips_state": chips_state}) assert r.status_code == 200, r.text trigger_settle = r.headers.get("HX-Trigger-After-Settle", "") assert "inchideModal" in trigger_settle, f"Salvarea trebuia sa reuseasca: {r.text[:400]}" r2 = client.get(f"/_import/{iid}/rand/0/editare-modal") assert r2.status_code == 200 payload = _chips_state_din_html(r2.text) item = payload["items"][0] assert item.get("exclus") == 1, f"exclus nu a fost persistat: {payload}" assert not item.get("cod"), f"item exclus nu trebuie sa aiba cod: {payload}" def test_redeschidere_modal_arata_chip_nedeclarat_si_x_readuce_warning(client): """Round-trip complet: dupa salvare, redeschiderea modalului arata chipul 'Nedeclarat' (nu warning); actiunea x (/form-chips remove) il readuce la warning.""" iid = _upload_and_preview(client, _SAMPLE_ROW_1OP) chips_state = json.dumps({"v": 1, "items": [ {"cod": "", "op": "Op-Nemapata", "den": "Op-Nemapata", "exclus": 1}, ]}) r = client.post(f"/_import/{iid}/rand/0/editeaza", data={"chips_state": chips_state}) assert r.status_code == 200 r2 = client.get(f"/_import/{iid}/rand/0/editare-modal") assert r2.status_code == 200 html = r2.text assert "Nedeclarat" in html, f"Chipul Nedeclarat trebuia sa reapara la redeschidere: {html[:600]}" assert "chip-nedeclarat" in html # x pe chip Nedeclarat -> /form-chips remove index 0 -> revine la warning. payload = _chips_state_din_html(html) r3 = client.post("/form-chips", data={ "chips_state": json.dumps(payload), "chips_action": "remove", "chips_remove_index": "0", }) assert r3.status_code == 200, r3.text assert "Nedeclarat" not in r3.text, "Dupa x, chipul Nedeclarat nu mai trebuie sa apara" payload3 = _chips_state_din_html(r3.text) item3 = payload3["items"][0] assert not item3.get("exclus"), f"x trebuia sa readuca operatia la warning (exclus=0): {payload3}" assert not item3.get("cod"), f"operatia trebuia sa ramana fara cod dupa x: {payload3}" def test_rand_toate_operatiile_excluse_devine_excluded(client): """Rand cu unica operatie exclusa -> stare preview 'excluded' ('Nedeclarat').""" iid = _upload_and_preview(client, _SAMPLE_ROW_1OP) chips_state = json.dumps({"v": 1, "items": [ {"cod": "", "op": "Op-Nemapata", "den": "Op-Nemapata", "exclus": 1}, ]}) r = client.post(f"/_import/{iid}/rand/0/editeaza", data={"chips_state": chips_state}) assert r.status_code == 200, r.text trigger = json.loads(r.headers.get("HX-Trigger", "{}")) stare = trigger.get("randSalvat", {}).get("stare", "") assert "Nedeclarat" in stare, f"randSalvat.stare trebuia sa arate Nedeclarat, got: {stare!r}" # --------------------------------------------------------------------------- # # Trimiteri: salvare + round-trip (post_corectie_trimitere) # # --------------------------------------------------------------------------- # def test_salvare_trimitere_cu_exclus(client): """/trimitere/{id}/corecteaza cu un item mapat + un item exclus -> queued, payload_json pastreaza AMBELE (itemul exclus adnotat exclus=1, fara cod_prestatie).""" _seed_cod("OE-1", "Schimb ulei") sid = _insert_submission(status="needs_mapping", payload=_payload_cu_ops( "WVWZZZ1JZXW0EX001", [("SchimbUlei", "Schimb ulei motor"), ("Filtru-Aer", "Filtru aer")], )) chips_state = json.dumps({"v": 1, "items": [ {"cod": "OE-1", "op": "SchimbUlei", "den": "Schimb ulei motor", "exclus": 0}, {"cod": "", "op": "Filtru-Aer", "den": "Filtru aer", "exclus": 1}, ]}) r = client.post(f"/trimitere/{sid}/corecteaza", data={"chips_state": chips_state}) assert r.status_code == 200, r.text[:500] row = _submission_row(sid) assert row["status"] == "queued", f"status asteptat queued, got {row['status']}: {r.text[:500]}" prestatii = _submission_payload(sid)["prestatii"] assert len(prestatii) == 2, f"ambele itemi trebuiau pastrate in payload: {prestatii}" by_op = {p.get("cod_op_service"): p for p in prestatii} assert by_op["SchimbUlei"].get("cod_prestatie") == "OE-1" assert by_op["Filtru-Aer"].get("exclus") is True, f"itemul exclus trebuia adnotat: {prestatii}" assert not by_op["Filtru-Aer"].get("cod_prestatie"), f"itemul exclus nu trebuie sa aiba cod: {prestatii}" def test_trimitere_toate_excluse_devine_needs_data(client): """Toate operatiile excluse -> needs_data cu motiv explicit (nu se trimite la RAR).""" sid = _insert_submission(status="needs_mapping", payload=_payload_cu_ops( "WVWZZZ1JZXW0EX002", [("Filtru-Aer", "Filtru aer")], )) chips_state = json.dumps({"v": 1, "items": [ {"cod": "", "op": "Filtru-Aer", "den": "Filtru aer", "exclus": 1}, ]}) r = client.post(f"/trimitere/{sid}/corecteaza", data={"chips_state": chips_state}) assert r.status_code == 200, r.text[:500] row = _submission_row(sid) assert row["status"] == "needs_data", f"status asteptat needs_data, got {row['status']}" assert "excluse" in (row["rar_error"] or "").lower() prestatii = _submission_payload(sid)["prestatii"] assert prestatii[0].get("exclus") is True, "itemul exclus trebuie pastrat in payload pt afisare motiv" # --------------------------------------------------------------------------- # # Sentinel __NEDECLARAT__ respins explicit daca apare ca cod_prestatie # # --------------------------------------------------------------------------- # def test_sentinel_nedeclarat_nu_ajunge_in_payload_import(client): """Daca cineva fabrica direct chips_state cu cod=__NEDECLARAT__ (nu prin selectul care seteaza exclus), salvarea trebuie respinsa — sentinelul nu e in nomenclator.""" iid = _upload_and_preview(client, _SAMPLE_ROW_1OP) chips_state = json.dumps({"v": 1, "items": [ {"cod": "__NEDECLARAT__", "op": "Op-Nemapata", "den": "Op-Nemapata", "exclus": 0}, ]}) r = client.post(f"/_import/{iid}/rand/0/editeaza", data={"chips_state": chips_state}) assert r.status_code == 200 assert "necunoscut" in r.text.lower(), f"Sentinelul trebuia respins ca cod necunoscut: {r.text[:400]}" trigger_settle = r.headers.get("HX-Trigger-After-Settle", "") assert "inchideModal" not in trigger_settle, "Salvarea NU trebuia sa reuseasca cu sentinelul in payload" # Confirma ca sentinelul NU a ajuns in override (redeschiderea nu-l arata ca si cod). r2 = client.get(f"/_import/{iid}/rand/0/editare-modal") payload2 = _chips_state_din_html(r2.text) coduri = [it.get("cod") for it in payload2["items"]] assert "__NEDECLARAT__" not in coduri, f"Sentinelul a ajuns in starea persistata: {payload2}" def test_sentinel_nedeclarat_nu_ajunge_in_payload_trimitere(client): """Aceeasi respingere pe calea Trimiteri (post_corectie_trimitere).""" sid = _insert_submission(status="needs_mapping", payload=_payload_cu_ops( "WVWZZZ1JZXW0SN001", [("Op-Test", "Operatie test")], )) old_status = _submission_row(sid)["status"] chips_state = json.dumps({"v": 1, "items": [ {"cod": "__NEDECLARAT__", "op": "Op-Test", "den": "Operatie test", "exclus": 0}, ]}) r = client.post(f"/trimitere/{sid}/corecteaza", data={"chips_state": chips_state}) assert r.status_code == 200 assert "necunoscut" in r.text.lower(), f"Sentinelul trebuia respins ca cod necunoscut: {r.text[:400]}" row = _submission_row(sid) assert row["status"] == old_status, f"Statusul nu trebuia sa se schimbe: {row['status']}" prestatii = _submission_payload(sid)["prestatii"] coduri = [p.get("cod_prestatie") for p in prestatii] assert "__NEDECLARAT__" not in coduri, f"Sentinelul a ajuns in payload: {prestatii}"