rar-autopass/tests/test_web_preview_paritate_mapari.py

"""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).

Testul nu verifica doar determinismul lui enrich, ci 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: 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, "embedding_similaritate": None, "nul": False,
    }