56ebe1f7cb
Inlocuieste cele 3 controale de asociere cod RAR cu un singur rand de chips + un singur select care adauga instant la change. Stare = un hidden chips_state JSON versionat; post_form_chips redus la 2 actiuni (add/remove). Optgroup Sugestii (fuzzy/k-NN) + optiune "Nu se declara la RAR" in select, cu tinta implicita evidentiata si placeholder care o numeste. exclus persistat prin payload_json (treapta noua de precedenta in resolve_prestatii, round-trip complet). Siguranta: itemii exclusi sunt scosi din payload la momentul trimiterii (worker split_prestatii_excluse inainte de build_rar_payload + filtru defensiv), ca sa nu ajunga NICIODATA la RAR ca codPrestatie:null. Suita: 1680 passed, 1 skipped. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
422 lines
18 KiB
Python
422 lines
18 KiB
Python
"""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}"
|