rar-autopass/tests/test_web_mapare_op.py

"""Teste US-004 — un singur „Salveaza maparile" pe panoul de operatii nemapate.

Ruta noua: POST /_import/{id}/mapare-operatii (plural)
  - primeste perechi (cod_op_service, cod_prestatie) ca liste paralele
  - apeleaza save_mapping pentru fiecare pereche cu cod ales (reuse exact)
  - ignora perechile cu cod_prestatie gol (nu eroare, nu salvare)
  - D#12: validare per-item — cod invalid -> skip + sumar, restul salvate
  - O singura _web_compute_preview + re-randare #import-section la final
  - CSRF + scoped sesiune + guard batch committed (409) pastrate
"""

from __future__ import annotations

import csv as csv_mod
import io
import os
import re
import tempfile

import pytest


@pytest.fixture()
def client(monkeypatch):
    tmp = tempfile.mkdtemp()
    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
    # Mod dev: fallback cont 1, fara login/CSRF (ca in test_import_mapare_operatie).
    monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "false")
    from app.config import get_settings

    get_settings.cache_clear()
    from app.main import app
    from fastapi.testclient import TestClient

    with TestClient(app) as c:
        yield c
    get_settings.cache_clear()


# ------------------------------------------------------------------ #
# Helpers de setup                                                     #
# ------------------------------------------------------------------ #

_HEADER = ["VIN", "Nr inmatriculare", "Data prestatie", "Odometru final",
           "Cod operatie", "Denumire"]
_CANON = ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final",
          "operatie", "denumire_op"]

# Doua operatii distincte: OP-REV si OP-FR
_ROWS_2OPS = [
    ["WVWZZZ1KZAW001111", "B100TST", "2026-06-15", "123456", "OP-REV", "Revizie periodica"],
    ["WVWZZZ1KZAW002222", "CJ200AB", "2026-05-20", "98765",  "OP-REV", "Revizie periodica"],
    ["WVWZZZ1KZAW003333", "IS300CD", "2026-04-10", "50000",  "OP-FR",  "Franare"],
]

# O singura operatie
_ROWS_1OP = [
    ["WVWZZZ1KZAW001111", "B100TST", "2026-06-15", "123456", "OP-REV", "Revizie periodica"],
    ["WVWZZZ1KZAW002222", "CJ200AB", "2026-05-20", "98765",  "OP-REV", "Revizie periodica"],
]


def _csv_bytes(header, rows, sep=";") -> bytes:
    buf = io.StringIO()
    w = csv_mod.writer(buf, delimiter=sep)
    w.writerow(header)
    for r in rows:
        w.writerow(r)
    return buf.getvalue().encode("utf-8")


def _upload(client, rows=None) -> int:
    """Incarca fisier CSV si intoarce import_id."""
    rows = _ROWS_2OPS if rows is None else rows
    r = client.post(
        "/_import/upload",
        files={"file": ("t.csv", _csv_bytes(_HEADER, rows), "text/csv")},
    )
    assert r.status_code == 200, r.text
    m = re.search(r"/_import/(\d+)/mapare-coloane", r.text)
    assert m, f"form mapare-coloane lipsa: {r.text[:300]}"
    return int(m.group(1))


def _map_columns(client, import_id, canon=None):
    return client.post(
        f"/_import/{import_id}/mapare-coloane",
        data={
            "colname": _HEADER,
            "canon": canon or _CANON,
            "format_data": "YYYY-MM-DD",
        },
    )


def _get_batch_counts(import_id):
    from app.db import get_connection
    conn = get_connection()
    try:
        return conn.execute(
            "SELECT ok, needs_mapping FROM import_batches WHERE id=?", (import_id,)
        ).fetchone()
    finally:
        conn.close()


def _get_row_statuses(import_id):
    from app.db import get_connection
    conn = get_connection()
    try:
        rows = conn.execute(
            "SELECT resolved_status FROM import_rows WHERE batch_id=? ORDER BY row_index",
            (import_id,),
        ).fetchall()
        return [r["resolved_status"] for r in rows]
    finally:
        conn.close()


def _get_mapping(cod_op_service, account_id=1):
    from app.db import get_connection
    conn = get_connection()
    try:
        return conn.execute(
            "SELECT cod_prestatie FROM operations_mapping WHERE account_id=? AND cod_op_service=?",
            (account_id, cod_op_service),
        ).fetchone()
    finally:
        conn.close()


# ------------------------------------------------------------------ #
# 1. Salveaza multiple operatii intr-un singur POST                   #
# ------------------------------------------------------------------ #

def test_mapare_operatii_salveaza_multiple_intr_un_post(client):
    """POST mapare-operatii cu 2 operatii alese -> ambele salvate, randurile trec la ok."""
    import_id = _upload(client)
    r = _map_columns(client, import_id)
    assert r.status_code == 200

    # Inainte: 0 ok, 3 needs_mapping
    b = _get_batch_counts(import_id)
    assert b["needs_mapping"] == 3
    assert b["ok"] == 0

    # Un singur POST cu ambele operatii
    rm = client.post(
        f"/_import/{import_id}/mapare-operatii",
        data={
            "cod_op_service": ["OP-REV", "OP-FR"],
            "cod_prestatie": ["OE-3", "OE-1"],
        },
    )
    assert rm.status_code == 200, rm.text

    # Raspuns e preview re-randat cu #import-section
    assert "import-section" in rm.text

    # Ambele mapari persistate
    assert _get_mapping("OP-REV") is not None
    assert _get_mapping("OP-REV")["cod_prestatie"] == "OE-3"
    assert _get_mapping("OP-FR") is not None
    assert _get_mapping("OP-FR")["cod_prestatie"] == "OE-1"

    # Toate randurile trecute la ok
    b2 = _get_batch_counts(import_id)
    assert b2["ok"] == 3, b2
    assert b2["needs_mapping"] == 0, b2

    statuses = _get_row_statuses(import_id)
    assert all(s == "ok" for s in statuses), statuses


def test_panoul_mapare_are_un_singur_form(client):
    """Preview-ul randeaza panoul de mapare cu un singur <form> si un buton Salveaza maparile."""
    import_id = _upload(client)
    r = _map_columns(client, import_id)
    assert r.status_code == 200

    # Ruta noua mapare-operatii (plural) prezenta in form
    assert f"/_import/{import_id}/mapare-operatii" in r.text
    # Un singur buton Salveaza
    assert "Salveaza maparile" in r.text
    # NU apare ruta singular mapare-operatie ca target de form (panoul unificat)
    # (poate apare ca ruta pastrata, dar nu ca hx-post al formularului de mapare in bulk)


# ------------------------------------------------------------------ #
# 2. Ignora operatiile fara cod ales                                  #
# ------------------------------------------------------------------ #

def test_mapare_operatii_ignora_randuri_neselectate(client):
    """Operatia cu cod_prestatie gol e ignorata (nu eroare, nu salvare)."""
    import_id = _upload(client, rows=_ROWS_2OPS)
    _map_columns(client, import_id)

    # OP-REV cu cod ales, OP-FR fara cod ales (string gol = "— alege cod RAR —")
    rm = client.post(
        f"/_import/{import_id}/mapare-operatii",
        data={
            "cod_op_service": ["OP-REV", "OP-FR"],
            "cod_prestatie": ["OE-3", ""],
        },
    )
    assert rm.status_code == 200, rm.text

    # OP-REV salvat
    assert _get_mapping("OP-REV") is not None

    # OP-FR nesal vat (nu eroare, nu mapare)
    assert _get_mapping("OP-FR") is None

    # Randurile OP-REV (2) sunt ok, OP-FR (1) raman needs_mapping
    b = _get_batch_counts(import_id)
    assert b["ok"] == 2, b
    assert b["needs_mapping"] == 1, b

    # Panoul mai arata OP-FR (inca nemapat)
    assert "OP-FR" in rm.text


def test_mapare_operatii_fara_nicio_selectie_nu_eroare(client):
    """POST cu toate cod_prestatie goale -> nici o eroare, nici o salvare, preview re-randat."""
    import_id = _upload(client, rows=_ROWS_1OP)
    _map_columns(client, import_id)

    rm = client.post(
        f"/_import/{import_id}/mapare-operatii",
        data={
            "cod_op_service": ["OP-REV"],
            "cod_prestatie": [""],
        },
    )
    assert rm.status_code == 200, rm.text

    # Nicio mapare salvata
    assert _get_mapping("OP-REV") is None

    # Randurile raman needs_mapping
    b = _get_batch_counts(import_id)
    assert b["needs_mapping"] == 2, b
    assert b["ok"] == 0, b


# ------------------------------------------------------------------ #
# 3. Re-rezolva randurile blocate cu needs_mapping                    #
# ------------------------------------------------------------------ #

def test_mapare_operatii_re_rezolva_blocatele(client):
    """Operatiile cu cod ales trec din needs_mapping la ok (re-rezolvare imediata)."""
    import_id = _upload(client, rows=_ROWS_1OP)
    _map_columns(client, import_id)

    # Inainte: 2 needs_mapping
    b = _get_batch_counts(import_id)
    assert b["needs_mapping"] == 2, b
    assert b["ok"] == 0, b

    rm = client.post(
        f"/_import/{import_id}/mapare-operatii",
        data={
            "cod_op_service": ["OP-REV"],
            "cod_prestatie": ["OE-3"],
        },
    )
    assert rm.status_code == 200

    # Dupa: 2 ok, 0 needs_mapping
    b2 = _get_batch_counts(import_id)
    assert b2["ok"] == 2, b2
    assert b2["needs_mapping"] == 0, b2

    statuses = _get_row_statuses(import_id)
    assert all(s == "ok" for s in statuses), statuses

    # Preview randat nu mai arata panoul de mapare
    assert "Operatii de mapat la cod RAR" not in rm.text


# ------------------------------------------------------------------ #
# 4. D#12 — validare per-item: cod invalid skip + sumar, restul ok   #
# ------------------------------------------------------------------ #

def test_mapare_operatii_cod_invalid_skip_salveaza_restul(client):
    """D#12: daca un cod ales e invalid (1 din 2), skip-ul + sumar, celalalt salvat, 1 re-render."""
    import_id = _upload(client, rows=_ROWS_2OPS)
    _map_columns(client, import_id)

    # OP-REV cod valid, OP-FR cod inexistent
    rm = client.post(
        f"/_import/{import_id}/mapare-operatii",
        data={
            "cod_op_service": ["OP-REV", "OP-FR"],
            "cod_prestatie": ["OE-3", "COD-INEXISTENT"],
        },
    )
    assert rm.status_code == 200, rm.text

    # OP-REV salvat (codul valid)
    assert _get_mapping("OP-REV") is not None
    assert _get_mapping("OP-REV")["cod_prestatie"] == "OE-3"

    # OP-FR nesalvat (cod invalid)
    assert _get_mapping("OP-FR") is None

    # Sumar in mesaj: cod invalid mentionat
    assert "COD-INEXISTENT" in rm.text or "necunoscut" in rm.text.lower()

    # Randurile OP-REV (2) ok, OP-FR (1) inca needs_mapping
    b = _get_batch_counts(import_id)
    assert b["ok"] == 2, b
    assert b["needs_mapping"] == 1, b

    # O singura re-randare (200, nu redirect, nu multiple #import-section)
    assert rm.text.count("import-section") >= 1


# ------------------------------------------------------------------ #
# 5. Guard batch committed (409)                                      #
# ------------------------------------------------------------------ #

def test_mapare_operatii_batch_committed_409(client):
    """Batch deja comis -> 409 Conflict."""
    import_id = _upload(client, rows=_ROWS_1OP)
    _map_columns(client, import_id)

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

    rm = client.post(
        f"/_import/{import_id}/mapare-operatii",
        data={
            "cod_op_service": ["OP-REV"],
            "cod_prestatie": ["OE-3"],
        },
    )
    assert rm.status_code == 409, rm.text


# ------------------------------------------------------------------ #
# 6. Guard scoped sesiune (404 cross-account)                         #
# ------------------------------------------------------------------ #

def test_mapare_operatii_scoped_404_alt_cont(monkeypatch):
    """Import apartinand altui cont -> 404 (scoping corect)."""
    tmp = tempfile.mkdtemp()
    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "scope.db"))
    monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true")
    from app.config import get_settings
    get_settings.cache_clear()

    from app.accounts import create_account
    from app.users import create_user
    from app.db import get_connection
    from app.main import app
    from fastapi.testclient import TestClient

    with TestClient(app, follow_redirects=False) as c:
        conn = get_connection()
        try:
            acct_a = create_account(conn, "ServiceA", active=True)
            create_user(conn, acct_a, "a@test.com", "parolasecreta10")
            acct_b = create_account(conn, "ServiceB", active=True)
            create_user(conn, acct_b, "b@test.com", "parolasecreta10")
        finally:
            conn.close()

        def _login_and_get_csrf(c, email, password="parolasecreta10"):
            resp = c.get("/login")
            m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text)
            assert m
            csrf = m.group(1)
            c.post("/login", data={"email": email, "parola": password, "csrf_token": csrf})
            # Get a fresh CSRF token for next request
            resp2 = c.get("/?tab=import")
            m2 = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp2.text)
            return m2.group(1) if m2 else csrf

        # Login cu cont A, upload fisier (batch apartine cont A)
        csrf_a = _login_and_get_csrf(c, "a@test.com")

        r = c.post(
            "/_import/upload",
            files={"file": ("t.csv", _csv_bytes(_HEADER, _ROWS_1OP), "text/csv")},
            data={"csrf_token": csrf_a},
        )
        assert r.status_code == 200, r.text
        m2 = re.search(r"/_import/(\d+)/mapare-coloane", r.text)
        assert m2
        import_id = int(m2.group(1))

        # Extrage CSRF din raspuns pentru mapare-coloane
        m_csrf = re.search(r'name="csrf_token"\s+value="([^"]+)"', r.text)
        csrf_for_map = m_csrf.group(1) if m_csrf else csrf_a
        c.post(f"/_import/{import_id}/mapare-coloane",
               data={"colname": _HEADER, "canon": _CANON, "format_data": "YYYY-MM-DD",
                     "csrf_token": csrf_for_map})

        # Login cu cont B, incearca mapare pe batch-ul lui A
        csrf_b = _login_and_get_csrf(c, "b@test.com")

        rm = c.post(
            f"/_import/{import_id}/mapare-operatii",
            data={
                "cod_op_service": ["OP-REV"],
                "cod_prestatie": ["OE-3"],
                "csrf_token": csrf_b,
            },
        )
        assert rm.status_code == 404, f"expected 404 got {rm.status_code}: {rm.text[:200]}"

    get_settings.cache_clear()