"""Teste US-010 (PRD 5.15): Bulk-fix din lista — selectie multipla -> actiune unica. Acceptance criteria: - test_bulk_remapeaza_selectie: N randuri needs_mapping + aplica cod -> toate -> queued - test_bulk_doar_blocate: randuri sent/sending nu sunt eligibile (sarite silentios) - test_bulk_scoped_cont: 404-before-409 — un cont nu atinge randurile altui cont """ from __future__ import annotations import json import os import re import tempfile import pytest from starlette.testclient import TestClient # --------------------------------------------------------------------------- # Helpere comune (aceeasi conventie ca test_web_submissions.py) # --------------------------------------------------------------------------- def _create_account_user(email: str, name: str = "Service", password: str = "parolasecreta10"): from app.accounts import create_account from app.users import create_user from app.db import get_connection conn = get_connection() try: acct_id = create_account(conn, name, active=True) create_user(conn, acct_id, email, password) return acct_id finally: conn.close() def _login(client, email: str, password: str = "parolasecreta10") -> None: 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, "CSRF token not found on /login" resp = client.post("/login", data={"email": email, "parola": password, "csrf_token": m.group(1)}) assert resp.status_code == 303 def _insert_submission(acct: int, status: str = "needs_mapping", *, payload: dict | None = None) -> int: """Insereaza o trimitere cu payload standard (needs_mapping cu cod_op_service).""" from app.db import get_connection conn = get_connection() try: p = payload if payload is not None else { "vin": "WVWZZZ1KZAW000123", "nr_inmatriculare": "B123TST", "data_prestatie": "2026-06-15", "odometru_final": "50000", "prestatii": [{"cod_op_service": "INTERN1", "denumire": "Schimb ulei"}], } cur = conn.execute( "INSERT INTO submissions (idempotency_key, account_id, status, payload_json) " "VALUES (?, ?, ?, ?)", (f"k-{status}-{os.urandom(6).hex()}", acct, status, json.dumps(p)), ) conn.commit() return int(cur.lastrowid) finally: conn.close() def _get_status(sid: int) -> str | None: """Citeste status-ul curent al unui rand din DB (sursa de adevar).""" from app.db import get_connection conn = get_connection() try: row = conn.execute("SELECT status FROM submissions WHERE id=?", (sid,)).fetchone() return row["status"] if row else None finally: conn.close() def _csrf_from_fragment(client) -> str: """Extrage CSRF token din /_fragments/submissions sau din dashboard (fallback). Submissions fragment include CSRF doar cand exista randuri (form bulk). Dashboard-ul (/) include mereu CSRF in formularul de upload. """ resp = client.get("/_fragments/submissions") assert resp.status_code == 200 m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text) or \ re.search(r'value="([^"]+)"\s+name="csrf_token"', resp.text) if m: return m.group(1) # Fallback: dashboard principal (contine intotdeauna un form cu CSRF dupa login) resp2 = client.get("/") m2 = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp2.text) or \ re.search(r'value="([^"]+)"\s+name="csrf_token"', resp2.text) assert m2, "CSRF token not found in submissions fragment or dashboard" return m2.group(1) # --------------------------------------------------------------------------- # Fixture client # --------------------------------------------------------------------------- @pytest.fixture() def client(monkeypatch): tmp = tempfile.mkdtemp() monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "bulk_fix.db")) monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true") from app.config import get_settings get_settings.cache_clear() from app.web import ratelimit ratelimit._hits.clear() from app.main import app with TestClient(app, follow_redirects=False) as c: yield c ratelimit._hits.clear() get_settings.cache_clear() # --------------------------------------------------------------------------- # Teste US-010 # --------------------------------------------------------------------------- def test_bulk_remapeaza_selectie(client): """US-010 AC principal: N randuri needs_mapping + aplica cod valid -> toate -> queued. OE-1 face parte din nomenclatorul seed (nomenclator_seed.FALLBACK_NOMENCLATOR), incarcat de init_db la startup; nu e nevoie de insert separat. Payload-uri diferite (VIN diferit) ca sa nu colizioneze la recalculul idempotentei. """ acct = _create_account_user("bulk_fix1@test.com") sid1 = _insert_submission(acct, "needs_mapping", payload={ "vin": "WVWZZZ1KZAW000111", "nr_inmatriculare": "B111TST", "data_prestatie": "2026-06-15", "odometru_final": "50000", "prestatii": [{"cod_op_service": "INTERN1", "denumire": "Schimb ulei"}], }) sid2 = _insert_submission(acct, "needs_mapping", payload={ "vin": "WVWZZZ1KZAW000222", "nr_inmatriculare": "B222TST", "data_prestatie": "2026-06-16", "odometru_final": "60000", "prestatii": [{"cod_op_service": "INTERN2", "denumire": "Verificare franare"}], }) _login(client, "bulk_fix1@test.com") csrf = _csrf_from_fragment(client) resp = client.post( "/trimiteri/bulk-fix", data={ "csrf_token": csrf, "submission_id": [str(sid1), str(sid2)], "cod_prestatie": "OE-1", }, ) assert resp.status_code == 200, f"Asteptam 200, primit {resp.status_code}" # Ambele randuri trebuie sa fie acum queued s1 = _get_status(sid1) s2 = _get_status(sid2) assert s1 == "queued", f"sid1 status={s1!r}, asteptam 'queued'" assert s2 == "queued", f"sid2 status={s2!r}, asteptam 'queued'" # Sumar vizibil in raspuns HTML (cel putin unul din: "reusit", "2", "queued") html = resp.text assert "reusit" in html.lower() or "2 " in html or "queued" in html.lower(), \ "Sumar bulk-fix lipseste din raspuns" def test_bulk_doar_blocate(client): """US-010 AC eligibilitate: randuri sent/sending sarite silentios; doar blocate procesate.""" acct = _create_account_user("bulk_fix2@test.com") # Rand sent (read-only, nu trebuie atins) sid_sent = _insert_submission(acct, "sent", payload={ "vin": "WVWZZZ1KZAW000222", "nr_inmatriculare": "B222TST", "data_prestatie": "2026-06-15", "odometru_final": "50000", "prestatii": [{"cod_prestatie": "OE-2", "denumire": "Intretinere"}], }) # Rand needs_mapping (gestionabil, trebuie procesat) sid_blocked = _insert_submission(acct, "needs_mapping") _login(client, "bulk_fix2@test.com") csrf = _csrf_from_fragment(client) # Trimitem ambele id-uri; doar cel blocat trebuie procesat resp = client.post( "/trimiteri/bulk-fix", data={ "csrf_token": csrf, "submission_id": [str(sid_sent), str(sid_blocked)], "cod_prestatie": "OE-1", }, ) assert resp.status_code == 200 # Randul sent ramane sent (read-only — INTERZIS sa fie modificat) assert _get_status(sid_sent) == "sent", \ "Randul sent a fost modificat de bulk-fix — INTERZIS" # Randul blocat a trecut la queued assert _get_status(sid_blocked) == "queued", \ f"Randul needs_mapping nu a trecut la queued: {_get_status(sid_blocked)!r}" def test_bulk_scoped_cont(client): """US-010 AC scope: contul A nu poate modifica randurile contului B. Pattern 404-before-409: randurile cross-account sunt sarite silentios (nu confirmam existenta), raspuns HTTP 200 cu sumar care reflecta 0 reusite. """ acct_a = _create_account_user("bulk_fix_a@test.com", name="Cont A") acct_b = _create_account_user("bulk_fix_b@test.com", name="Cont B") # Randul lui B (alt cont) sid_b = _insert_submission(acct_b, "needs_mapping", payload={ "vin": "WVWZZZ1KZAW000333", "nr_inmatriculare": "B333TST", "data_prestatie": "2026-06-15", "odometru_final": "50000", "prestatii": [{"cod_op_service": "INTERN3", "denumire": "Test extern"}], }) # Logat ca A — incearca sa aplice cod pe randul lui B _login(client, "bulk_fix_a@test.com") csrf = _csrf_from_fragment(client) resp = client.post( "/trimiteri/bulk-fix", data={ "csrf_token": csrf, "submission_id": [str(sid_b)], "cod_prestatie": "OE-1", }, ) # Raspuns 200 (nu 404 expus HTTP — cross-account e sarit silentios ca la bulk-delete) assert resp.status_code == 200 # Randul lui B NEATINS assert _get_status(sid_b) == "needs_mapping", \ "Randul contului B a fost modificat de contul A — INCALCARE SCOPE!"