3b5cf7a7d9
Comutator accounts.auto_send_enabled per cont: Auto OFF (default) tine randurile
la ingestie (submissions.held=1), worker-ul (claim_one AND held=0) le sare pana la
eliberare umana (per rand/bulk/auto-release OFF->ON). Snapshot held prin chokepoint
unic held_for_account pe toate caile de ingestie (API, import, reresolve, reactivare).
- schema/migrare: coloana held + index partial idx_submissions_held; auto_send_enabled
- API: echo onest held+motiv (US-010), ruta /prezentari/{id}/trimite-acum
- web: toggle header, modal confirmare tipata, buton Trimite per rand + Trimite toate,
banner coada tinuta imbatranita (L.142), contor "In asteptare (manual)"
- worker: expire_held (US-008, inchide gaura retentie PII), metrics held gauges
- ops: tools/carantina_held + runbook rollback (R4)
Nota review (/code-review high): re-snapshot held lipseste pe caile repune/corectie
(requeue_submission, post_corectie, bulk-fix) — de aliniat separat cu create_prezentari.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
348 lines
13 KiB
Python
348 lines
13 KiB
Python
"""Teste PRD 5.19 strat WEB: toggle "Trimite automat la RAR" (US-004), trimitere
|
|
manuala per rand + bulk (US-005), afisaj held (US-006 UI), banner coada tinuta
|
|
imbatranita + contor manual (US-007), audit app_events (US-009).
|
|
|
|
Model: TestClient + login web (ca tests/test_web_lifecycle.py). NU atinge conftest.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import tempfile
|
|
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(monkeypatch):
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "as.db"))
|
|
monkeypatch.setenv("AUTOPASS_LOG_DIR", os.path.join(tmp, "logs"))
|
|
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 _account_user(email, name="Service", password="parolasecreta10"):
|
|
from app.accounts import create_account
|
|
from app.users import create_user
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
aid = create_account(conn, name, active=True)
|
|
create_user(conn, aid, email, password)
|
|
return aid
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _login(client, email, password="parolasecreta10"):
|
|
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)
|
|
resp = client.post("/login", data={"email": email, "parola": password, "csrf_token": m.group(1)})
|
|
assert resp.status_code == 303, resp.text[:200]
|
|
return _csrf(client)
|
|
|
|
|
|
def _csrf(client):
|
|
resp = client.get("/?tab=acasa")
|
|
m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text)
|
|
assert m, "csrf_token negasit dupa login"
|
|
return m.group(1)
|
|
|
|
|
|
def _ins(account_id, status="queued", held=0, created_at=None):
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
content = {"vin": "WVWZZZ1KZAW000123", "nr_inmatriculare": "B999TST",
|
|
"data_prestatie": "2026-06-15", "odometru_final": "123456",
|
|
"prestatii": [{"cod_prestatie": "OE-1"}]}
|
|
if created_at is None:
|
|
cur = conn.execute(
|
|
"INSERT INTO submissions (idempotency_key, account_id, status, payload_json, held) "
|
|
"VALUES (?, ?, ?, ?, ?)",
|
|
(f"k-{os.urandom(6).hex()}", account_id, status, json.dumps(content), held),
|
|
)
|
|
else:
|
|
cur = conn.execute(
|
|
"INSERT INTO submissions (idempotency_key, account_id, status, payload_json, held, created_at) "
|
|
"VALUES (?, ?, ?, ?, ?, ?)",
|
|
(f"k-{os.urandom(6).hex()}", account_id, status, json.dumps(content), held, created_at),
|
|
)
|
|
conn.commit()
|
|
return int(cur.lastrowid)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _get_auto_send(account_id):
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
r = conn.execute("SELECT auto_send_enabled FROM accounts WHERE id=?", (account_id,)).fetchone()
|
|
return int(r["auto_send_enabled"])
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _held(sid):
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
return int(conn.execute("SELECT held FROM submissions WHERE id=?", (sid,)).fetchone()["held"])
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# US-004 — toggle Auto: persistenta + CSRF + scope #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_auto_send_persista(client):
|
|
aid = _account_user("a1@test.com")
|
|
csrf = _login(client, "a1@test.com")
|
|
assert _get_auto_send(aid) == 0
|
|
r = client.post("/auto-send", data={"enabled": "1", "csrf_token": csrf})
|
|
assert r.status_code == 200
|
|
assert _get_auto_send(aid) == 1
|
|
# Debifare
|
|
r = client.post("/auto-send", data={"csrf_token": csrf}) # enabled absent = OFF
|
|
assert r.status_code == 200
|
|
assert _get_auto_send(aid) == 0
|
|
|
|
|
|
def test_auto_send_csrf(client):
|
|
aid = _account_user("a2@test.com")
|
|
_login(client, "a2@test.com")
|
|
r = client.post("/auto-send", data={"enabled": "1", "csrf_token": "gresit"})
|
|
assert r.status_code == 403
|
|
assert _get_auto_send(aid) == 0
|
|
|
|
|
|
def test_auto_send_scoped_nu_atinge_alt_cont(client):
|
|
aid = _account_user("a3@test.com", name="A3")
|
|
other = _account_user("a3b@test.com", name="A3B")
|
|
csrf = _login(client, "a3@test.com")
|
|
r = client.post("/auto-send", data={"enabled": "1", "csrf_token": csrf})
|
|
assert r.status_code == 200
|
|
assert _get_auto_send(aid) == 1
|
|
# contul celalalt ramane OFF (account_id vine din sesiune, nu din formular)
|
|
assert _get_auto_send(other) == 0
|
|
|
|
|
|
def test_auto_send_scoped_ignora_account_id_din_formular(client):
|
|
aid = _account_user("a4@test.com", name="A4")
|
|
other = _account_user("a4b@test.com", name="A4B")
|
|
csrf = _login(client, "a4@test.com")
|
|
# incearca sa forteze alt cont prin formular -> trebuie ignorat
|
|
r = client.post("/auto-send", data={"enabled": "1", "account_id": str(other), "csrf_token": csrf})
|
|
assert r.status_code == 200
|
|
assert _get_auto_send(aid) == 1
|
|
assert _get_auto_send(other) == 0
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# US-004 — auto-release OFF->ON cu garda de confirmare #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_off_to_on_fara_confirmare_nu_elibereaza(client):
|
|
aid = _account_user("g1@test.com")
|
|
csrf = _login(client, "g1@test.com")
|
|
sid = _ins(aid, "queued", held=1)
|
|
r = client.post("/auto-send", data={"enabled": "1", "csrf_token": csrf})
|
|
assert r.status_code == 200
|
|
# fara confirmare: NU comita ON, NU elibereaza
|
|
assert _get_auto_send(aid) == 0
|
|
assert _held(sid) == 1
|
|
# raspunsul cere confirmare (modal)
|
|
assert "confirm" in r.text.lower() or "trimit" in r.text.lower()
|
|
|
|
|
|
def test_off_to_on_cu_confirmare_elibereaza_scoped(client):
|
|
aid = _account_user("g2@test.com", name="G2")
|
|
other = _account_user("g2b@test.com", name="G2B")
|
|
csrf = _login(client, "g2@test.com")
|
|
s1 = _ins(aid, "queued", held=1)
|
|
s2 = _ins(aid, "queued", held=1)
|
|
s_other = _ins(other, "queued", held=1)
|
|
r = client.post("/auto-send", data={"enabled": "1", "confirma": "1", "csrf_token": csrf})
|
|
assert r.status_code == 200
|
|
assert _get_auto_send(aid) == 1
|
|
assert _held(s1) == 0
|
|
assert _held(s2) == 0
|
|
# cross-account izolare: randul altui cont ramane tinut
|
|
assert _held(s_other) == 1
|
|
|
|
|
|
def test_off_to_on_fara_randuri_tinute_comite_direct(client):
|
|
aid = _account_user("g3@test.com")
|
|
csrf = _login(client, "g3@test.com")
|
|
# niciun rand tinut -> nu are nevoie de confirmare
|
|
r = client.post("/auto-send", data={"enabled": "1", "csrf_token": csrf})
|
|
assert r.status_code == 200
|
|
assert _get_auto_send(aid) == 1
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# US-005 — trimitere manuala per rand + bulk #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_trimite_acum_rand_tinut(client):
|
|
aid = _account_user("t1@test.com")
|
|
csrf = _login(client, "t1@test.com")
|
|
sid = _ins(aid, "queued", held=1)
|
|
r = client.post(f"/trimitere/{sid}/trimite-acum", data={"csrf_token": csrf})
|
|
assert r.status_code == 200
|
|
assert _held(sid) == 0
|
|
|
|
|
|
def test_trimite_acum_cross_account_404(client):
|
|
aid = _account_user("t2@test.com", name="T2")
|
|
other = _account_user("t2b@test.com", name="T2B")
|
|
csrf = _login(client, "t2@test.com")
|
|
sid_other = _ins(other, "queued", held=1)
|
|
r = client.post(f"/trimitere/{sid_other}/trimite-acum", data={"csrf_token": csrf})
|
|
assert r.status_code == 404
|
|
assert _held(sid_other) == 1
|
|
|
|
|
|
def test_trimite_acum_non_queued_noop(client):
|
|
aid = _account_user("t3@test.com")
|
|
csrf = _login(client, "t3@test.com")
|
|
sid = _ins(aid, "sent", held=0)
|
|
r = client.post(f"/trimitere/{sid}/trimite-acum", data={"csrf_token": csrf})
|
|
assert r.status_code == 200
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
assert conn.execute("SELECT status FROM submissions WHERE id=?", (sid,)).fetchone()["status"] == "sent"
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_trimite_acum_csrf(client):
|
|
aid = _account_user("t4@test.com")
|
|
_login(client, "t4@test.com")
|
|
sid = _ins(aid, "queued", held=1)
|
|
r = client.post(f"/trimitere/{sid}/trimite-acum", data={"csrf_token": "gresit"})
|
|
assert r.status_code == 403
|
|
assert _held(sid) == 1
|
|
|
|
|
|
def test_trimite_toate_scoped(client):
|
|
aid = _account_user("t5@test.com", name="T5")
|
|
other = _account_user("t5b@test.com", name="T5B")
|
|
csrf = _login(client, "t5@test.com")
|
|
s1 = _ins(aid, "queued", held=1)
|
|
s2 = _ins(aid, "queued", held=1)
|
|
s_other = _ins(other, "queued", held=1)
|
|
r = client.post("/trimite-toate", data={"csrf_token": csrf})
|
|
assert r.status_code == 200
|
|
assert _held(s1) == 0
|
|
assert _held(s2) == 0
|
|
assert _held(s_other) == 1
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# US-006 (UI) — eticheta held + buton Trimite in lista #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_rand_tinut_eticheta_si_buton(client):
|
|
aid = _account_user("u1@test.com")
|
|
_login(client, "u1@test.com")
|
|
sid_tinut = _ins(aid, "queued", held=1)
|
|
html = client.get("/_fragments/submissions").text
|
|
assert "In asteptare (manual)" in html or "Manual" in html
|
|
assert f"/trimitere/{sid_tinut}/trimite-acum" in html
|
|
|
|
|
|
def test_rand_ne_tinut_fara_buton(client):
|
|
aid = _account_user("u2@test.com")
|
|
_login(client, "u2@test.com")
|
|
sid = _ins(aid, "queued", held=0)
|
|
html = client.get("/_fragments/submissions").text
|
|
assert f"/trimitere/{sid}/trimite-acum" not in html
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# US-007 — banner coada tinuta imbatranita + contor manual #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_banner_aged_apare_pt_rand_vechi(client):
|
|
aid = _account_user("b1@test.com")
|
|
_login(client, "b1@test.com")
|
|
_ins(aid, "queued", held=1, created_at="2026-06-01 08:00:00") # >7 zile fata de 2026-07-05
|
|
html = client.get("/_fragments/status").text
|
|
assert "declarare obligatorie" in html.lower() or "tinute de" in html.lower()
|
|
|
|
|
|
def test_banner_aged_absent_pt_rand_recent(client):
|
|
aid = _account_user("b2@test.com")
|
|
_login(client, "b2@test.com")
|
|
from datetime import datetime, timezone
|
|
_ins(aid, "queued", held=1, created_at=datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"))
|
|
html = client.get("/_fragments/status").text
|
|
assert "declarare obligatorie" not in html.lower()
|
|
|
|
|
|
def test_contor_manual_apare(client):
|
|
aid = _account_user("b3@test.com")
|
|
_login(client, "b3@test.com")
|
|
_ins(aid, "queued", held=1)
|
|
_ins(aid, "queued", held=1)
|
|
html = client.get("/_fragments/status").text
|
|
assert "manual" in html.lower()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# US-009 — audit app_events #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def _events(account_id, tip):
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT * FROM app_events WHERE tip=? AND account_id=?", (tip, account_id)
|
|
).fetchall()
|
|
return rows
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_audit_auto_send_schimbat(client):
|
|
aid = _account_user("j1@test.com")
|
|
csrf = _login(client, "j1@test.com")
|
|
client.post("/auto-send", data={"enabled": "1", "csrf_token": csrf})
|
|
assert len(_events(aid, "auto_send_schimbat")) >= 1
|
|
|
|
|
|
def test_audit_held_eliberat_rand(client):
|
|
aid = _account_user("j2@test.com")
|
|
csrf = _login(client, "j2@test.com")
|
|
sid = _ins(aid, "queued", held=1)
|
|
client.post(f"/trimitere/{sid}/trimite-acum", data={"csrf_token": csrf})
|
|
assert len(_events(aid, "held_eliberat")) >= 1
|
|
|
|
|
|
def test_audit_held_eliberat_bulk(client):
|
|
aid = _account_user("j3@test.com")
|
|
csrf = _login(client, "j3@test.com")
|
|
_ins(aid, "queued", held=1)
|
|
_ins(aid, "queued", held=1)
|
|
client.post("/trimite-toate", data={"csrf_token": csrf})
|
|
evs = _events(aid, "held_eliberat")
|
|
assert len(evs) >= 1
|