"""Teste US-006a/b (PRD 3.3): scoping import web pe sesiune. US-006a: citiri (upload, preview, mapare-coloane) pe contul sesiunii. US-006b: scrieri (confirma) pe contul sesiunii; alt cont -> inaccesibil. C8/OV-2: aceeasi cheie idempotenta prin API si web pe acelasi cont. """ from __future__ import annotations import io import re import os import tempfile import openpyxl import pytest from fastapi.testclient import TestClient def _make_xlsx(rows: list[dict]) -> bytes: wb = openpyxl.Workbook() ws = wb.active if rows: ws.append(list(rows[0].keys())) for r in rows: ws.append(list(r.values())) buf = io.BytesIO() wb.save(buf) return buf.getvalue() _ROWS = [ {"vin": "WVWZZZ1KZAW111111", "nr_inmatriculare": "B111TST", "data_prestatie": "2026-06-01", "odometru_final": "10000", "cod_prestatie": "OE-1"}, ] def _csrf_from(html: str) -> str: """Extrage tokenul CSRF din HTML (hidden input).""" m = re.search(r'name="csrf_token" value="([^"]*)"', html) return m.group(1) if m else "" @pytest.fixture() def env(monkeypatch): tmp = tempfile.mkdtemp() monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "scope.db")) from app.config import get_settings get_settings.cache_clear() from app.main import app with TestClient(app, follow_redirects=False) as c: from app.db import get_connection conn = get_connection() from app.accounts import create_account acct_a = create_account(conn, "Cont A Scope", active=True) acct_b = create_account(conn, "Cont B Scope", active=True) yield c, conn, acct_a, acct_b conn.close() get_settings.cache_clear() def _setup_op_mapping(conn, account_id): """Configureaza maparea operatie cod_op=OE-1 -> cod_prestatie=OE-1 pt. cont.""" conn.execute( "INSERT OR IGNORE INTO nomenclator_rar (cod_prestatie, nume_prestatie) VALUES (?, ?)", ("OE-1", "Operatii electrice"), ) conn.execute( "INSERT OR REPLACE INTO operations_mapping " "(account_id, cod_op_service, cod_prestatie, auto_send) VALUES (?, ?, ?, ?)", (account_id, "OE-1", "OE-1", 1), ) def _mapare_coloane(c, import_id, csrf_token: str = ""): """Salveaza maparea de coloane; mapeaza cod_prestatie -> operatie (canonical). cod_prestatie din xlsx trebuie mapat la 'operatie' (nu la 'cod_prestatie' care nu e camp canonic). resolve_prestatii il rezolva din operations_mapping. """ return c.post( f"/_import/{import_id}/mapare-coloane", data={ "csrf_token": csrf_token, "colname": ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final", "cod_prestatie"], "canon": ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final", "operatie"], }, ) def test_upload_pe_contul_sesiunii(env, monkeypatch): """Upload creeaza batch pe contul din sesiune (nu DEFAULT_ACCOUNT_ID).""" client, conn, acct_a, acct_b = env monkeypatch.setattr("app.web.routes.require_login", lambda r: acct_a) r = client.post("/_import/upload", files={"file": ("a.xlsx", _make_xlsx(_ROWS), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}) assert r.status_code == 200 batch = conn.execute("SELECT id, account_id FROM import_batches").fetchone() assert batch is not None assert batch["account_id"] == acct_a def test_batch_alt_cont_inaccesibil(env, monkeypatch): """Batch-ul contului A nu e accesibil din sesiunea contului B (preview -> eroare).""" client, conn, acct_a, acct_b = env # Upload ca A (sesiune curata, fara csrf_token anterior) monkeypatch.setattr("app.web.routes.require_login", lambda r: acct_a) client.post("/_import/upload", files={"file": ("a.xlsx", _make_xlsx(_ROWS), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}) batch_id = conn.execute("SELECT id FROM import_batches WHERE account_id=?", (acct_a,)).fetchone()["id"] # Preview ca B (GET, fara CSRF) -> trebuie eroare/inaccesibil monkeypatch.setattr("app.web.routes.require_login", lambda r: acct_b) r = client.get(f"/_import/{batch_id}/preview") assert r.status_code == 200 assert "inexistent" in r.text.lower() or "inaccesibil" in r.text.lower() def test_commit_creeaza_submissions_pe_cont(env, monkeypatch): """Confirma creeaza submissions cu account_id-ul sesiunii.""" client, conn, acct_a, acct_b = env _setup_op_mapping(conn, acct_a) monkeypatch.setattr("app.web.routes.require_login", lambda r: acct_a) # Upload — raspunsul contine csrf_token in form (sesiunea l-a creat) r_upload = client.post("/_import/upload", files={"file": ("a.xlsx", _make_xlsx(_ROWS), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}) csrf = _csrf_from(r_upload.text) batch_id = conn.execute("SELECT id FROM import_batches WHERE account_id=?", (acct_a,)).fetchone()["id"] # Mapare coloane cu tokenul din upload r_map = _mapare_coloane(client, batch_id, csrf) csrf = _csrf_from(r_map.text) or csrf # tokenul din preview (stabil per sesiune) # Confirma cu tokenul sesiunii r = client.post(f"/_import/{batch_id}/confirma", data={"n_confirmat": "1", "csrf_token": csrf}) assert r.status_code == 200 sub = conn.execute("SELECT account_id FROM submissions").fetchone() assert sub is not None assert sub["account_id"] == acct_a def test_cheie_identica_api_vs_web_acelasi_cont(env, monkeypatch): """C8/OV-2: import web si API pe acelasi cont produc aceeasi cheie idempotenta.""" from app.idempotency import build_key, canonicalize_row client, conn, acct_a, acct_b = env _setup_op_mapping(conn, acct_a) row = { "vin": "WVWZZZ1KZAW999999", "nr_inmatriculare": "B999TST", "data_prestatie": "2026-06-15", "odometru_final": "99999", "prestatii": [{"cod_prestatie": "OE-1"}], } canon = canonicalize_row(row) key_api = build_key(acct_a, canon) # Upload web pe acelasi cont monkeypatch.setattr("app.web.routes.require_login", lambda r: acct_a) web_row = { "vin": "WVWZZZ1KZAW999999", "nr_inmatriculare": "B999TST", "data_prestatie": "2026-06-15", "odometru_final": "99999", "cod_prestatie": "OE-1", } r_up = client.post("/_import/upload", files={"file": ("w.xlsx", _make_xlsx([web_row]), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}) csrf = _csrf_from(r_up.text) batch_id = conn.execute("SELECT id FROM import_batches WHERE account_id=?", (acct_a,)).fetchone()["id"] r_map = _mapare_coloane(client, batch_id, csrf) csrf = _csrf_from(r_map.text) or csrf client.post(f"/_import/{batch_id}/confirma", data={"n_confirmat": "1", "csrf_token": csrf}) sub = conn.execute("SELECT idempotency_key FROM submissions WHERE account_id=?", (acct_a,)).fetchone() assert sub is not None assert sub["idempotency_key"] == key_api def test_confirma_alt_cont_inaccesibil(env, monkeypatch): """Confirma batch-ul contului A din sesiunea B -> eroare batch inexistent.""" client, conn, acct_a, acct_b = env _setup_op_mapping(conn, acct_a) # Upload + mapare ca A (cu CSRF tokens corecti) monkeypatch.setattr("app.web.routes.require_login", lambda r: acct_a) r_up = client.post("/_import/upload", files={"file": ("a.xlsx", _make_xlsx(_ROWS), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}) csrf = _csrf_from(r_up.text) batch_id = conn.execute("SELECT id FROM import_batches WHERE account_id=?", (acct_a,)).fetchone()["id"] r_map = _mapare_coloane(client, batch_id, csrf) csrf = _csrf_from(r_map.text) or csrf # Confirma ca B cu tokenul din sesiune (acelasi cookie jar, token valid CSRF) # dar batch apartine lui A -> "inexistent sau expirat" monkeypatch.setattr("app.web.routes.require_login", lambda r: acct_b) r = client.post(f"/_import/{batch_id}/confirma", data={"n_confirmat": "1", "csrf_token": csrf}) assert r.status_code == 200 assert "inexistent" in r.text.lower() or "inaccesibil" in r.text.lower() or "expirat" in r.text.lower()