merge: confirmare valori needs_review bulk+rapid (PR3 / Issue B)

tests/test_import_review.py

@@ -583,6 +583,228 @@ def test_confirma_review_form_nu_foloseste_hx_swap_none():
     )
 
 
+def _upload_and_preview_rows(client: TestClient, rows: list[dict]) -> int:
+    """Upload CSV cu `rows` + salveaza mapare fara format_data -> preview.
+
+    Generalizarea lui `_upload_and_preview_needs_review` pentru mai multe randuri
+    (fiecare cu data ambigua -> needs_review). Intoarce import_id.
+    """
+    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)
+    r2 = client.post(f"/_import/{iid}/mapare-coloane", data={
+        "colname": colnames,
+        "canon": canons,
+        "format_data": "",   # fara format -> ambiguous -> needs_review
+        "csrf_token": csrf2,
+    })
+    assert r2.status_code == 200, r2.text
+    return iid
+
+
+# Trei randuri, VIN-uri distincte (nu se dedupe ca duplicat), toate cu data ambigua.
+_ROWS_MULTI_NEEDS_REVIEW = [
+    {"VIN": "WVWZZZ1KZAW000123", "Nr": "B001TST", "Data": "05.06.2026", "KM": "123456", "Operatie": "OP-1"},
+    {"VIN": "WVWZZZ1KZAW000456", "Nr": "B002TST", "Data": "07.08.2026", "KM": "223456", "Operatie": "OP-1"},
+    {"VIN": "WVWZZZ1KZAW000789", "Nr": "B003TST", "Data": "09.10.2026", "KM": "323456", "Operatie": "OP-1"},
+]
+
+
+def test_confirma_toate_review_marcheaza_toate(client):
+    """B1 bulk: POST /_import/{id}/confirma-toate-review seteaza reviewed=1 pe TOATE
+    randurile needs_review din batch cu un singur click.
+
+    Verifica:
+    - Raspuns 200 cu preview re-randat (fragment #import-section)
+    - reviewed=1 in DB pentru toate cele 3 randuri
+    - dupa recalcul, randurile nu mai sunt needs_review (devin ok -> gata de trimis)
+    """
+    _seed_op1()
+    iid = _upload_and_preview_rows(client, _ROWS_MULTI_NEEDS_REVIEW)
+
+    # Toate randurile pornesc needs_review (reviewed=0)
+    for i in range(3):
+        assert _get_reviewed(iid, i) == 0, f"randul {i} trebuie sa fie reviewed=0 initial"
+
+    csrf = _get_csrf(client)
+    r = client.post(f"/_import/{iid}/confirma-toate-review", data={"csrf_token": csrf})
+    assert r.status_code == 200, r.text
+    # Raspunsul re-randeaza sectiunea de preview
+    assert "import-section" in r.text, "bulk trebuie sa re-randeze #import-section"
+
+    # Toate randurile confirmate in DB
+    for i in range(3):
+        assert _get_reviewed(iid, i) == 1, \
+            f"randul {i} trebuie sa fie reviewed=1 dupa confirmarea in bloc"
+
+    # Recalcul: niciun rand nu mai e needs_review (toate ok)
+    r2 = client.get(f"/_import/{iid}/preview")
+    assert r2.status_code == 200, r2.text
+    assert 'data-status="needs_review"' not in r2.text, \
+        "dupa confirmarea in bloc, niciun rand nu mai trebuie sa fie needs_review"
+
+
+def test_confirma_toate_review_guard_committed_409(client):
+    """POST confirma-toate-review pe batch deja comis -> 409."""
+    _seed_op1()
+    iid = _upload_and_preview_rows(client, _ROWS_MULTI_NEEDS_REVIEW)
+
+    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}/confirma-toate-review", data={"csrf_token": csrf})
+    assert r.status_code == 409, \
+        f"confirma-toate-review pe batch committed trebuie sa returneze 409, got {r.status_code}"
+
+
+def test_confirma_toate_review_scoped_404_alt_cont():
+    """POST confirma-toate-review pe batch-ul altui cont -> 404, iar randurile
+    contului A raman needs_review (reviewed=0): bulk-ul altui cont NU le atinge."""
+    tmp = tempfile.mkdtemp()
+    env_patch = {
+        "AUTOPASS_DB_PATH": os.path.join(tmp, "scope_bulk.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")
+                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)", (acct1,)
+                )
+                conn.commit()
+            finally:
+                conn.close()
+
+            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)})
+
+            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 ""
+
+            # userA creeaza batch cu randuri needs_review
+            _login(c, "userA@test.com")
+            rows = _ROWS_MULTI_NEEDS_REVIEW
+            csrf = _csrf()
+            r = c.post(
+                "/_import/upload",
+                files={"file": ("test.csv", io.BytesIO(_csv_bytes(rows)), "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,
+            })
+
+            # userB incearca bulk pe batch-ul lui A -> 404
+            _login(c, "userB@test.com")
+            csrf3 = _csrf()
+            r2 = c.post(f"/_import/{iid}/confirma-toate-review", data={"csrf_token": csrf3})
+            assert r2.status_code == 404, \
+                f"confirma-toate-review cross-account trebuie sa returneze 404, got {r2.status_code}"
+
+            # Randurile lui A raman needs_review (reviewed=0): nu au fost atinse.
+            conn = get_connection()
+            try:
+                n = conn.execute(
+                    "SELECT COUNT(*) AS c FROM import_rows WHERE batch_id=? AND reviewed=1",
+                    (iid,),
+                ).fetchone()["c"]
+                assert n == 0, "bulk-ul altui cont NU trebuie sa marcheze reviewed pe randurile lui A"
+            finally:
+                conn.close()
+    finally:
+        for k in env_patch:
+            if k in os.environ:
+                del os.environ[k]
+        ratelimit._hits.clear()
+        get_settings.cache_clear()
+        reset_cache()
+
+
+def test_confirm_rapid_buton_prezent_in_tabel(client):
+    """B1 quick: randul needs_review are butonul de confirm rapid direct in tabel,
+    care POST-eaza pe ruta /confirma-review (fara a deschide modalul)."""
+    _seed_op1()
+    iid = _upload_and_preview_needs_review(client)
+
+    r = client.get(f"/_import/{iid}/preview")
+    assert r.status_code == 200, r.text
+    html = r.text
+    assert "btn-confirma-rapid" in html, \
+        "randul needs_review trebuie sa aiba butonul de confirm rapid in tabel"
+    assert f"/_import/{iid}/rand/0/confirma-review" in html, \
+        "butonul de confirm rapid trebuie sa POST-eze pe ruta /confirma-review"
+
+
+def test_confirm_rapid_per_rand_seteaza_reviewed(client):
+    """B1 quick: POST /confirma-review din tabel (fara modal) seteaza reviewed=1 pe un
+    singur rand si NU atinge celelalte randuri needs_review din batch."""
+    _seed_op1()
+    iid = _upload_and_preview_rows(client, _ROWS_MULTI_NEEDS_REVIEW)
+
+    csrf = _get_csrf(client)
+    r = client.post(f"/_import/{iid}/rand/1/confirma-review", data={"csrf_token": csrf})
+    assert r.status_code == 200, r.text
+
+    assert _get_reviewed(iid, 1) == 1, "randul confirmat rapid trebuie sa fie reviewed=1"
+    assert _get_reviewed(iid, 0) == 0, "confirmul per-rand NU trebuie sa atinga alte randuri"
+    assert _get_reviewed(iid, 2) == 0, "confirmul per-rand NU trebuie sa atinga alte randuri"
+
+
 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,