rar-autopass/tests/test_held_ingestie.py

"""Teste PRD 5.19 US-002 — snapshot `held` la ingestie din comutatorul de cont.

Acopera TOATE situ-rile care scriu `status='queued'` prin ingestie (chokepoint
`held_for_account`): enqueue API, reactivare error->queued, dedup echo, commit import,
reresolve `needs_mapping`. `held` NU intra in idempotency_key.

Contul default id=1 e fortat pe Auto ON de fixtura autouse din conftest; testele care
vor Auto OFF il pun explicit pe 0 (`_set_auto_send(client, False)`).
"""

from __future__ import annotations

import io
import os
import tempfile

import openpyxl
import pytest
from fastapi.testclient import TestClient


@pytest.fixture()
def client(monkeypatch):
    tmp = tempfile.mkdtemp()
    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "held.db"))
    monkeypatch.setenv("AUTOPASS_LOG_DIR", os.path.join(tmp, "logs"))
    monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "false")
    from app.config import get_settings
    get_settings.cache_clear()
    from app.main import app
    with TestClient(app) as c:
        yield c
    get_settings.cache_clear()


def _set_auto_send(enabled: bool, account_id: int = 1) -> None:
    from app.db import get_connection
    conn = get_connection()
    try:
        conn.execute(
            "UPDATE accounts SET auto_send_enabled=? WHERE id=?",
            (1 if enabled else 0, account_id),
        )
        conn.commit()
    finally:
        conn.close()


def _row(sid: int) -> dict:
    from app.db import get_connection
    conn = get_connection()
    try:
        r = conn.execute("SELECT * FROM submissions WHERE id=?", (sid,)).fetchone()
        return dict(r) if r else {}
    finally:
        conn.close()


def _force_status(sid: int, status: str) -> None:
    from app.db import get_connection
    conn = get_connection()
    try:
        conn.execute("UPDATE submissions SET status=? WHERE id=?", (status, sid))
        conn.commit()
    finally:
        conn.close()


def _body(**over) -> dict:
    prez = {
        "vin": "WVWZZZ1KZAW000123",
        "nr_inmatriculare": "B999TST",
        "data_prestatie": "2026-06-15",
        "odometru_final": "123456",
        "prestatii": [{"cod_prestatie": "OE-1"}],
    }
    prez.update(over)
    return {"prezentari": [prez]}


# --------------------------------------------------------------------------- #
# Enqueue API                                                                 #
# --------------------------------------------------------------------------- #

def test_auto_off_ingestie_held(client):
    """Cont Auto OFF -> POST valid -> rand queued, held=1; raspuns held=true + motiv non-null."""
    _set_auto_send(False)
    r = client.post("/v1/prezentari", json=_body())
    assert r.status_code == 200, r.text
    res = r.json()["results"][0]
    assert res["status"] == "queued"
    assert res["held"] is True
    assert res["motiv"], "motiv trebuie sa fie non-null cand randul e tinut"
    assert _row(res["submission_id"])["held"] == 1


def test_auto_on_ingestie_not_held(client):
    """Cont Auto ON -> POST valid -> rand queued, held=0; raspuns held=false, motiv None."""
    _set_auto_send(True)
    r = client.post("/v1/prezentari", json=_body())
    assert r.status_code == 200, r.text
    res = r.json()["results"][0]
    assert res["status"] == "queued"
    assert res["held"] is False
    assert res["motiv"] is None
    assert _row(res["submission_id"])["held"] == 0


# --------------------------------------------------------------------------- #
# Reactivare error->queued (bug de bypass Eng Finding A)                       #
# --------------------------------------------------------------------------- #

def test_reactivare_respecta_auto_off(client):
    """Un rand `error` re-POST-at pe cont Auto OFF re-intra queued CU held=1 (nu pastreaza vechiul)."""
    _set_auto_send(True)
    r = client.post("/v1/prezentari", json=_body())
    sid = r.json()["results"][0]["submission_id"]
    assert _row(sid)["held"] == 0
    # Contul trece pe OFF si randul cade in error; re-POST-ul trebuie sa-l tina.
    _set_auto_send(False)
    _force_status(sid, "error")
    r2 = client.post("/v1/prezentari", json=_body())
    res = r2.json()["results"][0]
    assert res["submission_id"] == sid
    assert res["reactivated"] is True
    assert res["held"] is True
    assert res["motiv"]
    assert _row(sid)["held"] == 1


# --------------------------------------------------------------------------- #
# Dedup echo                                                                   #
# --------------------------------------------------------------------------- #

def test_dedup_echo_held(client):
    """Re-POST acelasi continut (Auto OFF) -> deduped, dar held propagat corect (nu queued curat fals)."""
    _set_auto_send(False)
    r1 = client.post("/v1/prezentari", json=_body())
    sid = r1.json()["results"][0]["submission_id"]
    r2 = client.post("/v1/prezentari", json=_body())
    res = r2.json()["results"][0]
    assert res["submission_id"] == sid
    assert res["deduped"] is True
    assert res["held"] is True
    assert res["motiv"]


# --------------------------------------------------------------------------- #
# Idempotenta: held nu schimba cheia                                           #
# --------------------------------------------------------------------------- #

def test_held_nu_schimba_idempotency_key(client):
    """Acelasi continut Auto ON vs OFF -> aceeasi idempotency_key (held e pur coada)."""
    _set_auto_send(True)
    sid_on = client.post("/v1/prezentari", json=_body()).json()["results"][0]["submission_id"]
    key_on = _row(sid_on)["idempotency_key"]
    # Simulam un al doilea DB logic: acelasi continut, cont OFF -> dedup pe aceeasi cheie.
    _set_auto_send(False)
    r2 = client.post("/v1/prezentari", json=_body())
    res2 = r2.json()["results"][0]
    assert res2["deduped"] is True
    assert res2["submission_id"] == sid_on
    assert _row(res2["submission_id"])["idempotency_key"] == key_on


# --------------------------------------------------------------------------- #
# Reresolve needs_mapping -> queued                                           #
# --------------------------------------------------------------------------- #

def test_reresolve_held(client):
    """Cont Auto OFF, submission needs_mapping -> reresolve dupa mapare -> queued cu held=1."""
    _set_auto_send(False)
    # Cod intern necunoscut -> needs_mapping (nu se trimite).
    body = _body(prestatii=[{"cod_op_service": "REV-INTERN", "denumire": "Revizie interna"}])
    r = client.post("/v1/prezentari", json=body)
    res = r.json()["results"][0]
    sid = res["submission_id"]
    assert _row(sid)["status"] == "needs_mapping"
    # Salvare mapare -> reresolve automat (POST /v1/mapari cheama reresolve_account).
    rc = client.post("/v1/mapari", json={
        "cod_op_service": "REV-INTERN", "cod_prestatie": "OE-1", "auto_send": True,
    })
    assert rc.status_code == 200, rc.text
    row = _row(sid)
    assert row["status"] == "queued"
    assert row["held"] == 1


def test_reresolve_held_auto_on(client):
    """Cont Auto ON, reresolve -> queued cu held=0."""
    _set_auto_send(True)
    body = _body(prestatii=[{"cod_op_service": "REV-INTERN", "denumire": "Revizie interna"}])
    sid = client.post("/v1/prezentari", json=body).json()["results"][0]["submission_id"]
    assert _row(sid)["status"] == "needs_mapping"
    client.post("/v1/mapari", json={
        "cod_op_service": "REV-INTERN", "cod_prestatie": "OE-1", "auto_send": True,
    })
    row = _row(sid)
    assert row["status"] == "queued"
    assert row["held"] == 0


# --------------------------------------------------------------------------- #
# Commit import                                                                #
# --------------------------------------------------------------------------- #

_HEADER = ["VIN", "Nr inmatriculare", "Data prestatie", "Odometru final", "Operatie"]
_ROW = ["WVWZZZ1KZAW001111", "B100TST", "2026-06-15", "123456", "Revizie"]
_COLMAP = {
    "VIN": "vin", "Nr inmatriculare": "nr_inmatriculare",
    "Data prestatie": "data_prestatie", "Odometru final": "odometru_final",
    "Operatie": "operatie",
}


def _make_xlsx(rows: list[list]) -> bytes:
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.title = "Sheet1"
    for row in rows:
        ws.append(row)
    buf = io.BytesIO()
    wb.save(buf)
    return buf.getvalue()


def _commit_import(client) -> int:
    data = _make_xlsx([_HEADER, _ROW])
    import_id = client.post(
        "/v1/import",
        files={"file": ("t.xlsx", io.BytesIO(data), "application/octet-stream")},
    ).json()["import_id"]
    client.post(f"/v1/import/{import_id}/column-mapping", json={"json_mapare": _COLMAP})
    client.post("/v1/mapari", json={"cod_op_service": "Revizie", "cod_prestatie": "OE-1", "auto_send": True})
    rp = client.get(f"/v1/import/{import_id}/preview")
    assert rp.status_code == 200 and rp.json()["summary"].get("ok", 0) == 1, rp.text
    rc = client.post(f"/v1/import/{import_id}/commit", json={
        "n_confirmat": 1, "reviewed_rows": [], "confirmed_by": "t@e2e.ro",
    })
    assert rc.status_code == 200, rc.text
    return rc.json()["submissions"][0]["submission_id"]


def test_import_commit_held_auto_off(client):
    """Commit import pe cont Auto OFF -> rand queued cu held=1."""
    _set_auto_send(False)
    sid = _commit_import(client)
    row = _row(sid)
    assert row["status"] == "queued"
    assert row["held"] == 1


def test_import_commit_not_held_auto_on(client):
    """Commit import pe cont Auto ON -> rand queued cu held=0."""
    _set_auto_send(True)
    sid = _commit_import(client)
    row = _row(sid)
    assert row["status"] == "queued"
    assert row["held"] == 0