@@ -94,9 +95,20 @@
{% if e.denumire and e.denumire != e.cod_op_service %}
{{ e.denumire }}
{% endif %}
- {% if e.suggestions %}
+ {% if e.suggestions or e.sugestie_principala or (e.surse_sugestie and e.surse_sugestie.nul) %}
sugestii:
+ {# 5.18 US-007: badge sursa pe sugestia sistemului — confirmat (GOLD) / similar
+ (SILVER+embedding k-NN) / non-operatie (pre-filtru NUL). Suggestion-only. #}
+ {% if e.sugestie_principala %}
+ {% if e.sugestie_principala.sursa == 'gold_partajat' %}
+ confirmat
+ {% else %}
+ similar
+ {% endif %}
+ {% elif e.surse_sugestie and e.surse_sugestie.nul %}
+ non-operatie
+ {% endif %}
{% for s in e.suggestions[:3] %}
{{ s.cod_prestatie }} ({{ s.score|round|int }}%){% if not loop.last %}, {% endif %}
{% endfor %}
diff --git a/tests/test_web_preview_paritate_mapari.py b/tests/test_web_preview_paritate_mapari.py
new file mode 100644
index 0000000..47237a7
--- /dev/null
+++ b/tests/test_web_preview_paritate_mapari.py
@@ -0,0 +1,221 @@
+"""Paritate editor mapare: panoul inline din preview-ul de import == pagina /mapari.
+
+Panoul "Operatii de mapat la cod RAR" din preview (`_collect_unmapped_ops` via
+`_web_compute_preview`) trebuie sa produca ACEEASI `sugestie_principala` +
+`surse_sugestie` ca `pending_unmapped` (functia care randeaza /mapari), pentru
+aceeasi denumire de operatie. Fara asta, cele doua editoare diverg (preview arata
+doar fuzzy, /mapari arata GOLD partajat > SILVER > embeddings k-NN + badge sursa).
+
+Test TARE (Eng finding F6): NU reproba doar determinismul lui enrich (a chema enrich
+de doua ori), ci probeaza WIRING-ul real — `conn` pasat din `_web_compute_preview`,
+corpusul indexat o data, campurile atasate — construind un batch de import cu randuri
+needs_mapping si comparand rezultatul cu `pending_unmapped`, cate un caz per sursa
+(gold / silver / embedding / nul).
+
+Suggestion-only (#13): enrichment NU intra in resolve_prestatii/load_mapping.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+import json
+import os
+import re
+import tempfile
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+# --------------------------------------------------------------------------- #
+# Cazuri: cate o operatie per sursa de sugestie. #
+# (op, denumire, cod_asteptat, sursa_asteptata) #
+# --------------------------------------------------------------------------- #
+_CAZ_GOLD = ("OP-GOLD", "Revizie gold speciala", "OE-1", "gold_partajat")
+_CAZ_SILVER = ("OP-SILVER", "Reparatie motor silver", "OE-2", "silver")
+_CAZ_EMB = ("OP-EMB", "Diagnoza semantica embedding", "OE-3", "embedding")
+_CAZ_NUL = ("OP-ITP", "ITP CT 99 XYZ", None, None) # pre-filtru NUL -> fara cod
+
+_CAZURI = [_CAZ_GOLD, _CAZ_SILVER, _CAZ_EMB, _CAZ_NUL]
+
+
+@pytest.fixture()
+def env(monkeypatch):
+ tmp = tempfile.mkdtemp()
+ monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "paritate_preview.db"))
+ monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "false")
+ # Embeddings ON ca sursa "embedding" sa fie exercitata; modulul e mock-uit mai jos
+ # (fara lazy-load al modelului ~230MB).
+ monkeypatch.setenv("AUTOPASS_EMBEDDINGS_ENABLED", "true")
+ from app.config import get_settings
+ get_settings.cache_clear()
+ from app.crypto import reset_cache
+ reset_cache()
+ from app.db import init_db
+ init_db()
+ yield monkeypatch
+ get_settings.cache_clear()
+ reset_cache()
+
+
+@pytest.fixture()
+def mock_emb(monkeypatch):
+ """Mock modulul embeddings: has_corpus True, suggest_nearest da OE-3 doar pt textul
+ care contine EMBEDDING, index_corpus/corpus_signature inofensive (fara model real)."""
+ import app.embeddings as emb
+
+ def _suggest(text, top_k=1):
+ # text = denumire NORMALIZATA (upper, fara diacritice) — enrich normalizeaza.
+ if "EMBEDDING" in (text or ""):
+ return [{"cod": "OE-3", "is_nul": False, "similaritate": 0.99}]
+ return []
+
+ monkeypatch.setattr(emb, "has_corpus", lambda: True)
+ monkeypatch.setattr(emb, "suggest_nearest", _suggest)
+ monkeypatch.setattr(emb, "corpus_signature", lambda: "")
+ monkeypatch.setattr(emb, "index_corpus", lambda items, signature=None: None)
+ return emb
+
+
+@pytest.fixture()
+def client(env):
+ from app.main import app
+ with TestClient(app) as c:
+ yield c
+
+
+def _csv_bytes(rows: list[dict]) -> bytes:
+ buf = io.StringIO()
+ writer = csv.DictWriter(buf, fieldnames=list(rows[0].keys()), delimiter=";")
+ writer.writeheader()
+ writer.writerows(rows)
+ return buf.getvalue().encode("utf-8")
+
+
+def _seed_surse(conn):
+ """Semeaza nomenclator + GOLD + SILVER pentru cazurile de test."""
+ from app.shared_store import record_human_validation, seed_suggestions
+
+ conn.executemany(
+ "INSERT OR IGNORE INTO nomenclator_rar (cod_prestatie, nume_prestatie) VALUES (?, ?)",
+ [("OE-1", "REVIZIE"), ("OE-2", "REPARATIE MOTOR"), ("OE-3", "DIAGNOZA")],
+ )
+ # GOLD partajat (shared_mappings) — NU intra in operations_mapping, deci op ramane needs_mapping.
+ record_human_validation(conn, _CAZ_GOLD[1], _CAZ_GOLD[2])
+ # SILVER (mapping_suggestions).
+ seed_suggestions(conn, [
+ {"denumire": _CAZ_SILVER[1], "cod_prestatie": _CAZ_SILVER[2], "source": "llm", "confidence": 0.9},
+ ])
+ conn.commit()
+
+
+def _insert_submissions_needs_mapping(conn):
+ """Insereaza cate un submission needs_mapping per caz, cu ACEEASI (op, denumire)
+ ca randurile de import — ca `pending_unmapped` sa vada aceleasi operatii."""
+ for i, (op, den, _cod, _sursa) in enumerate(_CAZURI):
+ conn.execute(
+ "INSERT INTO submissions (account_id, status, payload_json, idempotency_key) "
+ "VALUES (1, 'needs_mapping', ?, ?)",
+ (
+ json.dumps({
+ "vin": f"WVWZZZ1KZAW00{i:04d}",
+ "prestatii": [{"cod_op_service": op, "denumire": den}],
+ }),
+ f"paritate-sub-{i}",
+ ),
+ )
+ conn.commit()
+
+
+def _upload_batch(client: TestClient) -> int:
+ """Upload CSV cu cele 4 operatii nemapate + salveaza maparea de coloane. -> import_id."""
+ rows = [
+ {
+ "VIN": f"WVWZZZ1KZAW01{i:04d}",
+ "Nr": f"B{i:03d}TST",
+ "Data": "2026-06-15",
+ "KM": str(100000 + i),
+ "Operatie": op,
+ "Denumire": den,
+ }
+ for i, (op, den, _c, _s) in enumerate(_CAZURI)
+ ]
+ data = _csv_bytes(rows)
+ r = client.post("/_import/upload", files={"file": ("t.csv", io.BytesIO(data), "text/csv")})
+ assert r.status_code == 200, r.text
+ m = re.search(r"/_import/(\d+)/", r.text)
+ assert m, r.text[:400]
+ iid = int(m.group(1))
+ if f"/_import/{iid}/mapare-coloane" in r.text:
+ r2 = client.post(
+ f"/_import/{iid}/mapare-coloane",
+ data={
+ "colname": ["VIN", "Nr", "Data", "KM", "Operatie", "Denumire"],
+ "canon": ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final",
+ "operatie", "denumire_op"],
+ "format_data": "YYYY-MM-DD",
+ },
+ )
+ assert r2.status_code == 200, r2.text
+ return iid
+
+
+def test_preview_unmapped_ops_paritate_cu_pending_unmapped(client, mock_emb):
+ """`_web_compute_preview(...)["unmapped_ops"]` == `pending_unmapped(conn, account)`
+ pe `sugestie_principala` + `surse_sugestie`, per sursa (gold/silver/embedding/nul)."""
+ from app.db import get_connection
+ from app.mapping import pending_unmapped
+ from app.web.routes import _web_compute_preview
+
+ conn = get_connection()
+ try:
+ _seed_surse(conn)
+ iid = _upload_batch(client)
+ _insert_submissions_needs_mapping(conn)
+
+ # Calea PREVIEW (panou inline import) — foloseste conn (wiring nou).
+ preview = _web_compute_preview(conn, iid, 1)
+ assert isinstance(preview, dict), preview
+ preview_ops = {e["cod_op_service"]: e for e in preview["unmapped_ops"]}
+
+ # Calea /mapari (functia canonica de randare a editorului).
+ pending_ops = {e["cod_op_service"]: e for e in pending_unmapped(conn, 1)}
+
+ # Ambele cai trebuie sa vada exact aceleasi operatii.
+ assert set(preview_ops) == set(pending_ops) == {c[0] for c in _CAZURI}
+
+ # Paritate 1:1 pe sugestia principala + sursele, per operatie.
+ for op in preview_ops:
+ assert preview_ops[op]["sugestie_principala"] == pending_ops[op]["sugestie_principala"], op
+ assert preview_ops[op]["surse_sugestie"] == pending_ops[op]["surse_sugestie"], op
+
+ # Corectitudine per sursa (nu doar egalitate reciproca): fiecare caz da ce trebuie.
+ for op, _den, cod, sursa in _CAZURI:
+ sp = preview_ops[op]["sugestie_principala"]
+ surse = preview_ops[op]["surse_sugestie"]
+ if sursa is None:
+ # NUL: fara cod, badge non-operatie.
+ assert sp is None, op
+ assert surse["nul"] is True, op
+ else:
+ assert sp == {"cod_prestatie": cod, "sursa": sursa}, op
+ assert surse[sursa] == cod, op
+ finally:
+ conn.close()
+
+
+def test_collect_unmapped_ops_conn_none_contract_template(env):
+ """Fara conn, `_collect_unmapped_ops` init-eaza totusi `sugestie_principala`=None +
+ `surse_sugestie` default -> contractul catre template ramane identic (fara KeyError)."""
+ from app.web.routes import _collect_unmapped_ops
+
+ preview_rows = [{
+ "resolved_status": "needs_mapping",
+ "resolved": {"prestatii": [{"cod_op_service": "OP-X", "denumire": "Ceva"}]},
+ }]
+ out = _collect_unmapped_ops(preview_rows, [], conn=None)
+ assert len(out) == 1
+ e = out[0]
+ assert e["sugestie_principala"] is None
+ assert e["surse_sugestie"] == {"gold_partajat": None, "silver": None, "embedding": None, "nul": False}