feat(5.12): modal editare + cont obligatoriu la import; design.md + PRD 5.13 revizuit (/autoplan)

5.12 (livrat): editare in modal a randurilor de preview, cont obligatoriu inainte de import, formular editare extras (_form_editare, _editare_preview_modal), plus suita de teste aferenta (preview edit/compact, mapare op, form editare, signup, admin panel). Design + planificare: - docs/design.md: sistem de design (tokeni, breakpoints, scara control, componente, a11y). - docs/prd/prd-5.12-* si prd-5.13-* (5.13 cu raport /autoplan: CEO+Design+Eng, audit trail). Curatare: sterse PNG-urile de test/mockup temporare din radacina. Nota: implementarea CSS 5.13 (responsive compact + sistem butoane) NU e inca facuta — planul revizuit cere refactorul testelor fragile din test_web_responsive.py INAINTE de CSS. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-27 18:52:20 +00:00
parent 283299ff20
commit b26dbb79e1
44 changed files with 4852 additions and 305 deletions
--- a/tests/test_web_mapare_op.py
+++ b/tests/test_web_mapare_op.py
@@ -0,0 +1,421 @@
+"""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()