"""Teste US-010 (PRD 5.20): badge mediu RAR in liste/detaliu/audit/ecou API. Teste: test_badge_in_lista -- /_fragments/submissions afiseaza PRODUCȚIE / Testare per rand test_audit_contine_rar_env -- GET /v1/audit/export (CSV) contine coloana rar_env cu valoarea test_get_ecou_rar_env -- GET /v1/prezentari si /v1/prezentari/{id} includ campul rar_env """ from __future__ import annotations import csv import io import os import re import tempfile import pytest from cryptography.fernet import Fernet from starlette.testclient import TestClient # --------------------------------------------------------------------------- # Fixture client izolat (DB temp + cheie Fernet; fara autentificare API-key) # --------------------------------------------------------------------------- @pytest.fixture() def client(monkeypatch): """Client izolat cu DB temporara + cheie Fernet. AUTOPASS_REQUIRE_API_KEY=false.""" tmp = tempfile.mkdtemp() monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t_badge.db")) monkeypatch.setenv("AUTOPASS_CREDS_KEY", Fernet.generate_key().decode()) monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "false") monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true") from app.config import get_settings from app import crypto get_settings.cache_clear() crypto.reset_cache() from app.main import app with TestClient(app, follow_redirects=False) as c: yield c get_settings.cache_clear() crypto.reset_cache() # --------------------------------------------------------------------------- # Helpere — preluate din test_cont_medii.py (acelasi pattern izolat) # --------------------------------------------------------------------------- def _create_account_user( name: str = "Service Badge SRL", email: str = "badge@test.com", password: str = "parolasecreta10", ): """Creeaza cont + user. Returneaza (acct_id, user_id).""" 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) user_id = create_user(conn, acct_id, email, password) return acct_id, user_id finally: conn.close() def _login(client, email: str, password: str) -> None: """Login real prin HTTP — seteaza cookie sesiune pe client.""" resp = client.get("/login") assert resp.status_code == 200 m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text) if not m: m = re.search(r'value="([^"]+)"\s+name="csrf_token"', resp.text) assert m, "csrf_token negasit pe /login" csrf = m.group(1) resp = client.post( "/login", data={"email": email, "parola": password, "csrf_token": csrf}, ) assert resp.status_code == 303, f"Login esuat: {resp.status_code} {resp.text[:200]}" def _insert_submission(acct_id: int, rar_env: str, vin: str = "WVWZZZ1KZAW000123") -> int: """Insereaza un submission direct in DB cu rar_env dat. Returneaza id-ul randului.""" import json from app.db import get_connection from app.idempotency import build_key payload = { "vin": vin, "nr_inmatriculare": "B999TST", "data_prestatie": "2026-06-15", "odometru_final": "123456", "prestatii": [{"cod_prestatie": "OE-1", "denumire": "Revizie"}], } ikey = build_key(acct_id, payload) conn = get_connection() try: cur = conn.execute( "INSERT INTO submissions " "(idempotency_key, account_id, status, payload_json, rar_env) " "VALUES (?, ?, 'queued', ?, ?)", (ikey, acct_id, json.dumps(payload), rar_env), ) row_id = cur.lastrowid conn.commit() return row_id finally: conn.close() # --------------------------------------------------------------------------- # Teste # --------------------------------------------------------------------------- def test_badge_in_lista(client): """/_fragments/submissions afiseaza PRODUCȚIE pentru rar_env='prod' si Testare pt 'test'.""" acct_id, _ = _create_account_user("Firma Badge 1", "b1@test.com") _login(client, "b1@test.com", "parolasecreta10") # Submission cu mediu Productie _insert_submission(acct_id, rar_env="prod", vin="WVWZZZ1KZAW000001") # Submission cu mediu Testare _insert_submission(acct_id, rar_env="test", vin="WVWZZZ1KZAW000002") resp = client.get("/_fragments/submissions") assert resp.status_code == 200, f"/_fragments/submissions: {resp.status_code}" html = resp.text assert "PRODUC" in html.upper(), \ f"Badge 'PRODUCȚIE' lipsa in lista submissions: {html[:800]}" assert "testare" in html.lower(), \ f"Badge 'Testare' lipsa in lista submissions: {html[:800]}" def test_audit_contine_rar_env(client): """GET /v1/audit/export?status=all (CSV) contine coloana rar_env cu valoarea corecta.""" acct_id, _ = _create_account_user("Firma Badge 2", "b2@test.com") # Insereaza un submission cu rar_env='prod' si altul cu 'test' _insert_submission(acct_id, rar_env="prod", vin="WVWZZZ1KZAW000003") _insert_submission(acct_id, rar_env="test", vin="WVWZZZ1KZAW000004") # Exportul de audit functioneaza cu AUTOPASS_REQUIRE_API_KEY=false (cont id=1 implicit) # Contul creat e acct_id; cont default e 1 (creat de schema). # Inseram pe cont 1 explicit ca sa testam exportul fara cheie. from app.db import get_connection import json conn = get_connection() try: payload = { "vin": "WVWZZZ1KZAW000099", "nr_inmatriculare": "B001AUD", "data_prestatie": "2026-06-15", "odometru_final": "10000", "prestatii": [{"cod_prestatie": "OE-1", "denumire": "Revizie"}], } conn.execute( "INSERT INTO submissions " "(idempotency_key, account_id, status, payload_json, rar_env) " "VALUES ('audit-test-key', 1, 'sent', ?, 'prod')", (json.dumps(payload),), ) conn.commit() finally: conn.close() resp = client.get("/v1/audit/export?status=all") assert resp.status_code == 200, f"audit/export: {resp.status_code} {resp.text[:200]}" reader = csv.DictReader(io.StringIO(resp.text)) assert "rar_env" in (reader.fieldnames or []), \ f"Coloana 'rar_env' lipseste din CSV audit. Coloane: {reader.fieldnames}" rows = list(reader) assert rows, "Exportul CSV este gol — nicio trimitere?" # Cel putin un rand are rar_env setat env_values = {r["rar_env"] for r in rows if r.get("rar_env")} assert env_values, f"Toate randurile au rar_env gol: {rows}" assert "prod" in env_values, f"'prod' asteptat in valorile rar_env, gasit: {env_values}" def test_get_ecou_rar_env(client): """GET /v1/prezentari si /v1/prezentari/{id} includ campul rar_env in JSON.""" # Insereaza direct pe cont 1 (default fara cheie cu AUTOPASS_REQUIRE_API_KEY=false) import json from app.db import get_connection conn = get_connection() try: payload = { "vin": "WVWZZZ1KZAW000077", "nr_inmatriculare": "B002ECO", "data_prestatie": "2026-06-20", "odometru_final": "55000", "prestatii": [{"cod_prestatie": "OE-1", "denumire": "Revizie"}], } cur = conn.execute( "INSERT INTO submissions " "(idempotency_key, account_id, status, payload_json, rar_env) " "VALUES ('ecou-test-key-prod', 1, 'queued', ?, 'prod')", (json.dumps(payload),), ) sub_id = cur.lastrowid conn.commit() finally: conn.close() # GET /v1/prezentari (lista) resp_lista = client.get("/v1/prezentari") assert resp_lista.status_code == 200, f"GET /v1/prezentari: {resp_lista.status_code}" lista = resp_lista.json().get("submissions", []) assert lista, "Lista de submission-uri este goala" # Fiecare rand din lista trebuie sa aiba campul rar_env sub = next((s for s in lista if s.get("id") == sub_id), None) assert sub is not None, f"Submission {sub_id} negasit in lista" assert "rar_env" in sub, f"Campul 'rar_env' lipseste din raspunsul listei: {sub}" assert sub["rar_env"] == "prod", f"rar_env asteptat 'prod', gasit {sub['rar_env']!r}" # GET /v1/prezentari/{id} (detaliu) resp_det = client.get(f"/v1/prezentari/{sub_id}") assert resp_det.status_code == 200, f"GET /v1/prezentari/{sub_id}: {resp_det.status_code}" det = resp_det.json() assert "rar_env" in det, f"Campul 'rar_env' lipseste din detaliul submission-ului: {det}" assert det["rar_env"] == "prod", f"rar_env asteptat 'prod', gasit {det['rar_env']!r}"