"""Teste US-004 (PRD 5.8) — sectiunea „Reguli automate (text)" din pagina Mapari. Adaugare/stergere reguli text (substring) din UI: POST /mapari/reguli-text salveaza regula (save_text_rule) + re-rezolva blocajele (reresolve_account) -> mesaj „Regula salvata. Deblocate: N" + trigger trimiteriChanged. POST /mapari/reguli-text/sterge sterge regula. Ambele scoped pe contul sesiunii (require_login), CSRF obligatoriu. Cod absent din nomenclator -> respins inline. """ from __future__ import annotations import json import os import re import tempfile import pytest from starlette.testclient import TestClient 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) assert m resp = client.post("/login", data={"email": email, "parola": password, "csrf_token": m.group(1)}) assert resp.status_code == 303 def _csrf(client) -> str: resp = client.get("/?tab=mapari") m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text) assert m, "csrf_token negasit" return m.group(1) def _seed_needs_mapping(acct: int, *, op: str, denumire: str | None = None) -> int: """Submission needs_mapping pe canal API (batch_id NULL) cu o operatie nemapata.""" from app.db import get_connection conn = get_connection() try: cur = conn.execute( "INSERT INTO submissions (idempotency_key, account_id, status, payload_json, rar_error) " "VALUES (?, ?, 'needs_mapping', ?, ?)", ( f"k-{os.urandom(6).hex()}", acct, json.dumps({ "vin": "WVWZZZ1JZXW000111", "nr_inmatriculare": "B11AAA", "data_prestatie": "2026-06-18", "odometru_final": "12345", "prestatii": [{"cod_prestatie": None, "cod_op_service": op, "denumire": denumire or op}], }), json.dumps({"unmapped": [{"cod_op_service": op}]}), ), ) conn.commit() return int(cur.lastrowid) finally: conn.close() def _status_of(sid: int) -> str: from app.db import get_connection conn = get_connection() try: return conn.execute("SELECT status FROM submissions WHERE id=?", (sid,)).fetchone()["status"] finally: conn.close() def _text_rules(acct: int) -> list[dict]: from app.db import get_connection from app.mapping import load_text_rules conn = get_connection() try: return load_text_rules(conn, acct) finally: conn.close() def _seed_text_rule(acct: int, pattern: str, cod: str, auto_send: int = 0) -> None: from app.db import get_connection conn = get_connection() try: conn.execute( "INSERT INTO operation_text_rules (account_id, pattern, cod_prestatie, auto_send) " "VALUES (?, ?, ?, ?)", (acct, pattern, cod, auto_send), ) conn.commit() finally: conn.close() @pytest.fixture() def client(monkeypatch): tmp = tempfile.mkdtemp() monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "reguli_text.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() def test_post_regula_text_salveaza_si_rerezolva(client): """POST salveaza regula + re-rezolva blocajele; mesaj „Deblocate: N" + trigger.""" acct = _create_account_user("rt@test.com") sid = _seed_needs_mapping(acct, op="DIVERSE VERIFICARI 159004") assert _status_of(sid) == "needs_mapping" _login(client, "rt@test.com") csrf = _csrf(client) resp = client.post("/mapari/reguli-text", data={ "pattern": "verificari", "cod_prestatie": "OE-2", "auto_send": "true", "csrf_token": csrf, }) assert resp.status_code == 200 assert "Regula salvata" in resp.text assert "Deblocate" in resp.text assert resp.headers.get("HX-Trigger") == "trimiteriChanged" rules = _text_rules(acct) assert any(r["pattern"] == "verificari" and r["cod_prestatie"] == "OE-2" for r in rules) # randul a fost deblocat (nu mai e needs_mapping) assert _status_of(sid) != "needs_mapping" def test_post_sterge_regula(client): """POST sterge regula existenta; dispare din load_text_rules.""" acct = _create_account_user("del@test.com") _seed_text_rule(acct, "verificare", "OE-2") assert len(_text_rules(acct)) == 1 _login(client, "del@test.com") csrf = _csrf(client) resp = client.post("/mapari/reguli-text/sterge", data={ "pattern": "verificare", "csrf_token": csrf, }) assert resp.status_code == 200 assert _text_rules(acct) == [] def test_regula_text_scoped_pe_cont_sesiune(client): """Salvarea creeaza regula DOAR pe contul sesiunii, nu pe alt cont.""" acct_a = _create_account_user("a@test.com", name="Cont A") acct_b = _create_account_user("b@test.com", name="Cont B") _login(client, "b@test.com") csrf = _csrf(client) resp = client.post("/mapari/reguli-text", data={ "pattern": "schimb ulei", "cod_prestatie": "OE-1", "csrf_token": csrf, }) assert resp.status_code == 200 assert any(r["pattern"] == "schimb ulei" for r in _text_rules(acct_b)) assert _text_rules(acct_a) == [] def test_csrf_necesar(client): """Fara token CSRF valid -> 403, fara regula salvata.""" acct = _create_account_user("cf@test.com") _login(client, "cf@test.com") resp = client.post("/mapari/reguli-text", data={ "pattern": "verificare", "cod_prestatie": "OE-2", "csrf_token": "gresit", }) assert resp.status_code == 403 assert _text_rules(acct) == [] def test_cod_invalid_respins(client): """Cod absent din nomenclator -> mesaj inline, fara salvare.""" acct = _create_account_user("ci@test.com") _login(client, "ci@test.com") csrf = _csrf(client) resp = client.post("/mapari/reguli-text", data={ "pattern": "verificare", "cod_prestatie": "NU-EXISTA", "csrf_token": csrf, }) assert resp.status_code == 200 assert "necunoscut" in resp.text.lower() assert _text_rules(acct) == []