rar-autopass/tests/test_import_review.py

"""Teste US-007 (PRD 5.12): gate review in modal — import_rows.reviewed + confirmare.

TDD RED: testele sunt scrise inainte de implementare.

Scenarii:
  1. Migrare: import_rows.reviewed INTEGER DEFAULT 0, idempotent.
  2. Rand needs_review exclus din "gata de trimis" pana la confirmare explicita.
  3. POST confirma-review seteaza reviewed=1 → randul devine ok la recalcul.
  4. `reviewed` NU intra in payload/idempotency (marcaj separat).
  5. Editarea unei valori pe un rand confirmat reseteaza reviewed=0 (D#9).
"""

from __future__ import annotations

import io
import os
import re
import tempfile

import pytest
from fastapi.testclient import TestClient


# --------------------------------------------------------------------------- #
# Fixtures                                                                      #
# --------------------------------------------------------------------------- #

@pytest.fixture()
def client(monkeypatch):
    """Client fara autentificare web obligatorie (cont 1 implicit)."""
    tmp = tempfile.mkdtemp()
    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "ir.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()


# --------------------------------------------------------------------------- #
# Helpere                                                                       #
# --------------------------------------------------------------------------- #

def _seed_op1(account_id: int = 1) -> None:
    """Semeaza nomenclator + mapare OP-1 → R-FRANE."""
    from app.db import get_connection
    conn = get_connection()
    try:
        conn.execute(
            "INSERT OR REPLACE INTO nomenclator_rar (cod_prestatie, nume_prestatie) "
            "VALUES ('R-FRANE','Reparatie frane')"
        )
        conn.execute(
            "INSERT OR IGNORE INTO operations_mapping "
            "(account_id, cod_op_service, cod_prestatie, auto_send) "
            "VALUES (?, 'OP-1', 'R-FRANE', 1)",
            (account_id,),
        )
        conn.commit()
    finally:
        conn.close()


def _csv_bytes(rows: list[dict], sep: str = ";") -> bytes:
    import csv as _csv
    buf = io.StringIO()
    writer = _csv.DictWriter(buf, fieldnames=list(rows[0].keys()), delimiter=sep)
    writer.writeheader()
    writer.writerows(rows)
    return buf.getvalue().encode("utf-8")


# Rand care declanseaza needs_review: data in format ambiguu (DD.MM.YYYY cu zi<=12)
# si format_data=None -> col_fmt="ambiguous" -> is_ambiguous_date=True
_ROWS_NEEDS_REVIEW = [
    {
        "VIN": "WVWZZZ1KZAW000123",
        "Nr": "B001TST",
        "Data": "05.06.2026",   # Format ambiguu: zi=5 <= 12, luna=6 <= 12
        "KM": "123456",
        "Operatie": "OP-1",
    },
]

_MAP_COLS = {
    "VIN": "vin",
    "Nr": "nr_inmatriculare",
    "Data": "data_prestatie",
    "KM": "odometru_final",
    "Operatie": "operatie",
}


def _get_csrf(client: TestClient) -> str:
    r = client.get("/")
    m = re.search(r'name="csrf_token"\s+value="([^"]+)"', r.text) or \
        re.search(r'value="([^"]+)"\s+name="csrf_token"', r.text)
    return m.group(1) if m else ""


def _upload_and_preview_needs_review(client: TestClient) -> int:
    """Upload CSV cu data ambigua + salveaza mapare fara format_data → preview.

    format_data=None → col_fmt='ambiguous' → data '05.06.2026' → is_ambiguous=True
    → flag needs_review in _resolve_row_for_preview.

    Intoarce import_id.
    """
    rows = _ROWS_NEEDS_REVIEW
    csv_data = _csv_bytes(rows)
    csrf = _get_csrf(client)
    r = client.post(
        "/_import/upload",
        files={"file": ("test.csv", io.BytesIO(csv_data), "text/csv")},
        data={"csrf_token": csrf},
    )
    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]
    csrf2 = _get_csrf(client)
    # IMPORTANT: format_data="" (gol/None) → date_col_format={} → col_fmt="ambiguous"
    r2 = client.post(f"/_import/{iid}/mapare-coloane", data={
        "colname": colnames,
        "canon": canons,
        "format_data": "",   # fara format -> ambiguous
        "csrf_token": csrf2,
    })
    assert r2.status_code == 200, r2.text
    return iid


def _get_reviewed(import_id: int, row_index: int) -> int:
    """Citeste valoarea reviewed din DB pentru un rand."""
    from app.db import get_connection
    conn = get_connection()
    try:
        row = conn.execute(
            "SELECT reviewed FROM import_rows WHERE batch_id=? AND row_index=?",
            (import_id, row_index),
        ).fetchone()
        return int(row["reviewed"]) if row else -1
    finally:
        conn.close()


def _get_idempotency_key(import_id: int, row_index: int) -> str | None:
    """Citeste cheia de idempotenta a unui rand prin endpoint-ul de preview."""
    from app.db import get_connection
    from app.crypto import decrypt_creds
    from app.api.v1.import_router import (
        _resolve_row_for_preview, _signature, _build_idempotency_key,
    )
    import json
    from app.mapping import account_or_default, load_mapping_meta, load_nomenclator_codes, load_text_rules
    from app.import_parse import parse_date_value

    conn = get_connection()
    try:
        acct = account_or_default(1)
        row_db = conn.execute(
            "SELECT raw_json, override_json, reviewed FROM import_rows "
            "WHERE batch_id=? AND row_index=?",
            (import_id, row_index),
        ).fetchone()
        if not row_db:
            return None
        raw = decrypt_creds(row_db["raw_json"]) or {}
        ov = decrypt_creds(row_db["override_json"]) if row_db["override_json"] else None

        col_names = list(raw.keys())
        sig = _signature(col_names)
        mapping_row = conn.execute(
            "SELECT json_mapare, format_data FROM column_mappings "
            "WHERE account_id=? AND signature_coloane=?",
            (acct, sig),
        ).fetchone()
        if not mapping_row:
            return None
        json_mapare = json.loads(mapping_row["json_mapare"])
        format_data = mapping_row["format_data"]
        date_col_format: dict[str, str] = {}
        if format_data:
            for col_f, camp_c in json_mapare.items():
                if camp_c == "data_prestatie":
                    date_col_format[col_f] = format_data

        mapping_meta = load_mapping_meta(conn, acct)
        mapping = {op: meta["cod_prestatie"] for op, meta in mapping_meta.items()}
        valid_codes = load_nomenclator_codes(conn) or None
        text_rules = load_text_rules(conn, acct)

        info = _resolve_row_for_preview(
            raw_row=raw,
            json_mapare=json_mapare,
            date_col_format=date_col_format,
            coercion_flags=[],
            mapping=mapping,
            mapping_meta=mapping_meta,
            formula_columns=[],
            override=ov or None,
            valid_codes=valid_codes,
            text_rules=text_rules,
            reviewed=bool(row_db["reviewed"]),
        )
        try:
            return _build_idempotency_key(1, info["resolved"])
        except Exception:
            return None
    finally:
        conn.close()


# --------------------------------------------------------------------------- #
# Teste                                                                         #
# --------------------------------------------------------------------------- #

def test_migrare_adauga_coloana_reviewed_idempotent():
    """_migrate adauga coloana import_rows.reviewed INTEGER DEFAULT 0.

    Idempotent: a doua invocare nu ridica eroare.
    """
    tmp = tempfile.mkdtemp()
    import os
    os.environ["AUTOPASS_DB_PATH"] = os.path.join(tmp, "m.db")
    from app.config import get_settings
    get_settings.cache_clear()
    from app.crypto import reset_cache
    reset_cache()
    try:
        from app.db import get_connection, _migrate

        # Initializare DB (inclusiv _migrate)
        from app.db import init_db
        init_db()

        conn = get_connection()
        try:
            cols = {r["name"] for r in conn.execute("PRAGMA table_info(import_rows)").fetchall()}
            assert "reviewed" in cols, \
                "import_rows trebuie sa aiba coloana 'reviewed' dupa init_db()"

            # Verifica DEFAULT 0
            row_info = {r["name"]: r for r in conn.execute("PRAGMA table_info(import_rows)").fetchall()}
            reviewed_col = row_info.get("reviewed")
            assert reviewed_col is not None
            # dflt_value poate fi "0" sau 0 in functie de SQLite versiune
            assert str(reviewed_col["dflt_value"]) == "0", \
                f"import_rows.reviewed trebuie sa aiba DEFAULT 0, got: {reviewed_col['dflt_value']}"

            # Idempotenta: a doua invocare a _migrate nu ridica eroare
            _migrate(conn)  # no exception
        finally:
            conn.close()
    finally:
        del os.environ["AUTOPASS_DB_PATH"]
        get_settings.cache_clear()
        reset_cache()


def test_needs_review_exclus_din_gata_pana_la_confirmare(client):
    """Un rand needs_review nu intra in 'gata de trimis' (n_confirmat = 0) pana la confirmare.

    US-007 Q1: randul cu data ambigua apare cu pill 'Verifica valori' si este
    EXCLUS din n_confirmat. Bannerul de discoverability (T1) trebuie sa fie prezent.
    """
    _seed_op1()
    iid = _upload_and_preview_needs_review(client)

    # Verifica preview via GET /_import/{id}/preview
    r = client.get(f"/_import/{iid}/preview")
    assert r.status_code == 200, r.text
    html = r.text

    # Randul are starea needs_review
    assert "needs_review" in html, \
        "Randul cu data ambigua trebuie sa apara cu starea needs_review in preview"
    # Pill-ul cu eticheta "Verifica valori"
    assert "Verifica valori" in html or "verifica valori" in html.lower(), \
        "Pill-ul 'Verifica valori' trebuie sa apara pentru randul needs_review"

    # n_confirmat = 0 (randul NU e in ok)
    # Cautam valoarea campului n_confirmat (value="0")
    n_match = re.search(r'id="n-confirmat"[^>]*value="(\d+)"', html) or \
              re.search(r'name="n_confirmat"[^>]*value="(\d+)"', html)
    if n_match:
        n_val = int(n_match.group(1))
        assert n_val == 0, \
            f"n_confirmat trebuie sa fie 0 cand randul e needs_review (negasit), got {n_val}"

    # Bannerul de discoverability (T1) — prezent cand summary.needs_review > 0
    # Trebuie sa contina un mesaj despre faptul ca randurile nu pleaca pana la confirmare
    # (ex. 'nu pleaca la RAR' sau 'confirmi in modal' sau 'Verifica valori')
    banner_present = (
        "nu pleaca" in html.lower() or
        "confirmi in modal" in html.lower() or
        "preview-needs-review-banner" in html
    )
    assert banner_present, \
        "Bannerul de discoverability (T1) trebuie sa fie prezent cand exista randuri needs_review"


def test_confirmare_in_modal_seteaza_reviewed_si_devine_ok(client):
    """POST /_import/{id}/rand/0/confirma-review seteaza reviewed=1.

    US-007 T2: operatorul apasa 'Confirma valorile' in modal →
    reviewed=1 in DB → randul devine ok la recalcul preview.

    Verifica:
    - Raspuns 200
    - reviewed=1 in DB
    - HX-Trigger: randSalvat cu noua stare 'Gata de trimis' (pentru toast)
    - HX-Trigger: reincarcaPreview + HX-Trigger-After-Settle: inchideModal
    """
    import json as _json
    _seed_op1()
    iid = _upload_and_preview_needs_review(client)

    # Verifica ca randul e needs_review inainte de confirmare
    assert _get_reviewed(iid, 0) == 0, "reviewed trebuie sa fie 0 inainte de confirmare"

    # POST confirma-review
    csrf = _get_csrf(client)
    r = client.post(f"/_import/{iid}/rand/0/confirma-review", data={"csrf_token": csrf})
    assert r.status_code == 200, r.text

    # reviewed=1 in DB
    assert _get_reviewed(iid, 0) == 1, \
        "reviewed trebuie sa fie 1 in DB dupa confirmare"

    # Contractul nou: reload preview + randSalvat cu noua stare (nu OOB pe <tr>).
    trig = _json.loads(r.headers.get("HX-Trigger", "{}"))
    assert trig.get("reincarcaPreview") is True, "confirma-review trebuie sa ceara reincarcaPreview"
    assert trig.get("randSalvat", {}).get("stare") == "Gata de trimis", \
        "Dupa confirmare, randSalvat.stare trebuie sa fie 'Gata de trimis' (pentru toast)"

    # Modal se inchide
    trigger = r.headers.get("HX-Trigger-After-Settle", "")
    assert "inchideModal" in trigger, \
        f"confirma-review trebuie sa emita inchideModal, got: '{trigger}'"


def test_reviewed_nu_intra_in_payload_sau_idempotency(client):
    """reviewed NU intra in payload, override_json sau cheia de idempotenta.

    US-007 marcaj separat (D#8): reviewed e DOAR un flag de confirmare umana,
    nu un camp de continut. Cheia de idempotenta trebuie sa fie identica inainte
    si dupa confirmare.
    """
    _seed_op1()
    iid = _upload_and_preview_needs_review(client)

    # Cheia inainte de confirmare
    key_before = _get_idempotency_key(iid, 0)
    assert key_before is not None, "Cheia de idempotenta trebuie sa existe inainte de confirmare"

    # Confirma
    csrf = _get_csrf(client)
    r = client.post(f"/_import/{iid}/rand/0/confirma-review", data={"csrf_token": csrf})
    assert r.status_code == 200, r.text

    # Cheia dupa confirmare
    key_after = _get_idempotency_key(iid, 0)
    assert key_after is not None, "Cheia de idempotenta trebuie sa existe dupa confirmare"

    assert key_before == key_after, \
        f"Cheia de idempotenta NU trebuie sa se schimbe la confirmare! " \
        f"Inainte: {key_before}, dupa: {key_after}. " \
        "'reviewed' NU trebuie sa intre in payload sau cheia de idempotenta."

    # Verifica ca 'reviewed' nu e in override_json
    from app.db import get_connection
    from app.crypto import decrypt_creds
    conn = get_connection()
    try:
        row = conn.execute(
            "SELECT override_json FROM import_rows WHERE batch_id=? AND row_index=?",
            (iid, 0),
        ).fetchone()
        if row and row["override_json"]:
            ov = decrypt_creds(row["override_json"]) or {}
            assert "reviewed" not in ov, \
                "'reviewed' NU trebuie sa fie in override_json (camp separat, nu camp de continut)"
    finally:
        conn.close()


def test_editare_valoare_pe_needs_review_reseteaza_reviewed(client):
    """Editarea unei valori pe un rand confirmat reseteaza reviewed=0 (D#9).

    Cand operatorul SCHIMBA o valoare (via POST editeaza) pe un rand deja confirmat
    (reviewed=1), apply_row_override trebuie sa reseteze reviewed=0.
    → Randul se intoarce in starea needs_review si cere re-confirmare.
    """
    _seed_op1()
    iid = _upload_and_preview_needs_review(client)

    # Confirma randul
    csrf = _get_csrf(client)
    r = client.post(f"/_import/{iid}/rand/0/confirma-review", data={"csrf_token": csrf})
    assert r.status_code == 200, r.text
    assert _get_reviewed(iid, 0) == 1, "reviewed trebuie sa fie 1 dupa confirmare"

    # Editeaza o valoare (odometru_final)
    csrf2 = _get_csrf(client)
    r2 = client.post(f"/_import/{iid}/rand/0/editeaza", data={
        "odometru_final": "200000",
        "csrf_token": csrf2,
    })
    assert r2.status_code == 200, r2.text

    # reviewed trebuie sa fie 0 (resetat de apply_row_override)
    assert _get_reviewed(iid, 0) == 0, \
        "reviewed trebuie sa fie resetat la 0 dupa editarea unei valori (D#9 — re-cere confirmare)"


def test_confirma_review_guard_committed_409(client):
    """POST confirma-review pe batch deja comis → 409 (guard committed)."""
    _seed_op1()
    iid = _upload_and_preview_needs_review(client)

    # Marcheaza batch ca committed
    from app.db import get_connection
    conn = get_connection()
    try:
        conn.execute("UPDATE import_batches SET status='committed' WHERE id=?", (iid,))
        conn.commit()
    finally:
        conn.close()

    csrf = _get_csrf(client)
    r = client.post(f"/_import/{iid}/rand/0/confirma-review", data={"csrf_token": csrf})
    assert r.status_code == 409, \
        f"confirma-review pe batch committed trebuie sa returneze 409, got {r.status_code}"


def test_confirma_review_scoped_404_alt_cont():
    """POST confirma-review pe un rand al altui cont → 404 (scoping JOIN)."""
    tmp = tempfile.mkdtemp()
    env_patch = {
        "AUTOPASS_DB_PATH": os.path.join(tmp, "scope.db"),
        "AUTOPASS_WEB_AUTH_REQUIRED": "true",
    }
    for k, v in env_patch.items():
        os.environ[k] = v
    from app.config import get_settings
    get_settings.cache_clear()
    from app.crypto import reset_cache
    reset_cache()
    from app.web import ratelimit
    ratelimit._hits.clear()
    from app.main import app

    try:
        with TestClient(app, follow_redirects=False) as c:
            from app.db import get_connection
            from app.accounts import create_account
            from app.users import create_user
            conn = get_connection()
            try:
                acct1 = create_account(conn, "Firma A", active=True)
                create_user(conn, acct1, "userA@test.com", "parola123secure")
                acct2 = create_account(conn, "Firma B", active=True)
                create_user(conn, acct2, "userB@test.com", "parola123secure")
            finally:
                conn.close()

            # Semeaza operatii pentru acct1
            from app.db import get_connection as gcn
            conn2 = gcn()
            try:
                conn2.execute(
                    "INSERT OR REPLACE INTO nomenclator_rar (cod_prestatie, nume_prestatie) "
                    "VALUES ('R-FRANE','Reparatie frane')"
                )
                conn2.execute(
                    "INSERT OR IGNORE INTO operations_mapping "
                    "(account_id, cod_op_service, cod_prestatie, auto_send) "
                    "VALUES (?, 'OP-1', 'R-FRANE', 1)", (acct1,)
                )
                conn2.commit()
            finally:
                conn2.close()

            # Login ca userA, creeaza batch
            def _login(client, email, pwd="parola123secure"):
                resp = client.get("/login")
                m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text) or \
                    re.search(r'value="([^"]+)"\s+name="csrf_token"', resp.text)
                assert m
                client.post("/login", data={
                    "email": email, "parola": pwd, "csrf_token": m.group(1),
                })

            _login(c, "userA@test.com")

            rows = _ROWS_NEEDS_REVIEW
            csv_data = _csv_bytes(rows)

            def _csrf():
                r = c.get("/")
                m = re.search(r'name="csrf_token"\s+value="([^"]+)"', r.text) or \
                    re.search(r'value="([^"]+)"\s+name="csrf_token"', r.text)
                return m.group(1) if m else ""

            csrf = _csrf()
            r = c.post(
                "/_import/upload",
                files={"file": ("test.csv", io.BytesIO(csv_data), "text/csv")},
                data={"csrf_token": csrf},
            )
            assert r.status_code == 200
            m = re.search(r"/_import/(\d+)/mapare-coloane", r.text)
            assert m
            iid = int(m.group(1))
            colnames = list(rows[0].keys())
            canons = [_MAP_COLS[c] for c in colnames]
            csrf2 = _csrf()
            c.post(f"/_import/{iid}/mapare-coloane", data={
                "colname": colnames, "canon": canons,
                "format_data": "", "csrf_token": csrf2,
            })

            # Login ca userB si incearca confirma-review pe batch-ul lui A
            _login(c, "userB@test.com")
            csrf3 = _csrf()
            r2 = c.post(f"/_import/{iid}/rand/0/confirma-review", data={"csrf_token": csrf3})
            assert r2.status_code == 404, \
                f"confirma-review cross-account trebuie sa returneze 404, got {r2.status_code}"
    finally:
        for k in env_patch:
            if k in os.environ:
                del os.environ[k]
        ratelimit._hits.clear()
        get_settings.cache_clear()
        reset_cache()


# --------------------------------------------------------------------------- #
# Teste markup Bug 1: confirma-review form swap (B1)                           #
# --------------------------------------------------------------------------- #

def test_confirma_review_form_nu_foloseste_hx_swap_none():
    """Bug B1 (markup): formularul confirma-review NU trebuie sa foloseasca
    hx-swap='none' — cu none, htmx aplica doar OOB-urile dar NU executa scriptul
    din continutul principal → updateN() nu ruleaza → n_confirmat stale → 422.

    Forma corecta: hx-target='#detaliu-modal-body' + hx-swap='innerHTML'.
    """
    import pathlib
    template = (
        pathlib.Path(__file__).parent.parent
        / "app/web/templates/_editare_preview_modal.html"
    )
    html = template.read_text(encoding="utf-8")

    # Gasim sectiunea formularului confirma-review (dupa marcajul T2)
    idx = html.find("confirma-review")
    assert idx >= 0, "Formularul confirma-review nu a fost gasit in template"
    form_section = html[idx:]

    # NU trebuie sa existe hx-swap="none" pe formularul confirma-review
    assert 'hx-swap="none"' not in form_section, (
        "Formularul confirma-review NU trebuie sa foloseasca hx-swap='none'. "
        "Cu none, scriptul updateN() nu ruleaza → n_confirmat stale → gate 422."
    )

    # TREBUIE sa tinteasca #detaliu-modal-body cu innerHTML
    assert 'hx-target="#detaliu-modal-body"' in form_section, (
        "Formularul confirma-review trebuie sa aiba hx-target='#detaliu-modal-body' "
        "ca scriptul updateN sa fie executat (identic cu formularul editeaza)."
    )
    assert 'hx-swap="innerHTML"' in form_section, (
        "Formularul confirma-review trebuie sa aiba hx-swap='innerHTML' "
        "ca scriptul updateN sa fie executat."
    )


def test_confirma_review_cere_reincarcarea_preview(client):
    """Contractul nou (dogfood 5.13): confirma-review NU mai depinde de scriptul updateN
    din payload (care, cu OOB pe <tr> rupt, lasa randul stale). Acum cere reincarcaPreview,
    iar preview-ul reincarcat re-randeaza contorul si butonul de confirmare cu n_confirmat
    corect server-side — deci problema B1 (n_confirmat stale -> 422) dispare structural.

    Verifica:
    - Raspuns 200
    - HX-Trigger contine reincarcaPreview (reincarca contorul/confirmarea, fresh)
    """
    import json as _json
    _seed_op1()
    iid = _upload_and_preview_needs_review(client)

    csrf = _get_csrf(client)
    r = client.post(f"/_import/{iid}/rand/0/confirma-review", data={"csrf_token": csrf})
    assert r.status_code == 200, r.text

    trig = _json.loads(r.headers.get("HX-Trigger", "{}"))
    assert trig.get("reincarcaPreview") is True, (
        "confirma-review trebuie sa ceara reincarcaPreview — preview-ul reincarcat aduce "
        "n_confirmat corect server-side (fara dependenta de scriptul updateN din payload)."
    )