c842e3352a
Implementeaza PRD 5.6 complet (14 stories, TDD). Doua axe:
Lifecycle trimiteri blocate (Val A):
- submissions_admin.py: sterge/repune scoped (404 cross-account inaintea lui 409 stare)
- reactivare dedup peste `error` cu CAS (WHERE id=? AND status='error'), creds noi in
submissions + accounts.rar_creds_enc; worker invalideaza sesiunea RAR la creds proaspete
(JWT 30h vechi nu mai trimite cu parola gresita); camp aditiv `reactivated:true`
- retentie randuri blocate 30z; purge_expired exclude queued/sending; purge_after curatat
la reactivare/requeue
- API DELETE /v1/prezentari/{id} + /repune (200+JSON); UI butoane + bulk + banner actionabil
Observabilitate:
- app/observ.py log_event: dublu canal app_events (DB) + RotatingFileHandler per-proces,
redactare creds/PII la scriere (redact_pii/vin_partial)
- request_id middleware + X-Request-ID pe toate raspunsurile
- handler global excepții -> 500 envelope 6-chei + request_id (traceback doar in jurnal)
- audit cerere API (api_prezentari/api_auth_esuat) + audit worker (rar_login/tranzitii)
- tab "Jurnal" filtrabil scoped (non-admin doar contul sau); retentie jurnal 90z
- rar_error expus in GET /v1/prezentari/{id} (recovery observabil)
pytest -q: 741 passed, 0 failed. Docs: PRD raport VERIFY, contract endpointuri noi, ROADMAP.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
113 lines
3.5 KiB
Python
113 lines
3.5 KiB
Python
"""Teste US-006 (PRD 5.6): tab Jurnal in dashboard (scoped + filtre)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
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, "jrnl.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", admin=False):
|
|
from app.accounts import create_account
|
|
from app.users import create_user, set_admin
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
aid = create_account(conn, name, active=True)
|
|
create_user(conn, aid, email, "parolasecreta10")
|
|
if admin:
|
|
set_admin(conn, aid, True)
|
|
conn.commit()
|
|
return aid
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _login(client, email):
|
|
resp = client.get("/login")
|
|
m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text)
|
|
resp = client.post("/login", data={"email": email, "parola": "parolasecreta10", "csrf_token": m.group(1)})
|
|
assert resp.status_code == 303
|
|
|
|
|
|
def _event(account_id, tip, nivel="INFO", mesaj="x"):
|
|
from app import observ
|
|
observ.log_event(tip, nivel=nivel, account_id=account_id, mesaj=mesaj)
|
|
|
|
|
|
def test_non_admin_vede_doar_evenimentele_contului_sau(client):
|
|
aid = _account_user("u@test.com")
|
|
other = _account_user("o@test.com", name="Alt")
|
|
_event(aid, "api_prezentari", mesaj="al meu MARKER_A")
|
|
_event(other, "api_prezentari", mesaj="al altuia MARKER_B")
|
|
_login(client, "u@test.com")
|
|
|
|
html = client.get("/_fragments/jurnal").text
|
|
assert "MARKER_A" in html
|
|
assert "MARKER_B" not in html
|
|
|
|
|
|
def test_admin_vede_toate_si_filtru_cont(client):
|
|
admin = _account_user("admin@test.com", name="Admin", admin=True)
|
|
other = _account_user("client@test.com", name="Client")
|
|
_event(admin, "rar_login", mesaj="eveniment ADMINEV")
|
|
_event(other, "api_prezentari", mesaj="eveniment CLIENTEV")
|
|
_login(client, "admin@test.com")
|
|
|
|
# admin vede tot
|
|
html = client.get("/_fragments/jurnal").text
|
|
assert "ADMINEV" in html
|
|
assert "CLIENTEV" in html
|
|
# filtru pe cont
|
|
html2 = client.get(f"/_fragments/jurnal?cont={other}").text
|
|
assert "CLIENTEV" in html2
|
|
assert "ADMINEV" not in html2
|
|
|
|
|
|
def test_filtru_pe_tip_si_nivel(client):
|
|
aid = _account_user("f@test.com")
|
|
_event(aid, "api_prezentari", nivel="INFO", mesaj="EV_INFO")
|
|
_event(aid, "submission_error", nivel="ERROR", mesaj="EV_ERR")
|
|
_login(client, "f@test.com")
|
|
|
|
html = client.get("/_fragments/jurnal?tip=submission_error").text
|
|
assert "EV_ERR" in html
|
|
assert "EV_INFO" not in html
|
|
|
|
html2 = client.get("/_fragments/jurnal?nivel=INFO").text
|
|
assert "EV_INFO" in html2
|
|
assert "EV_ERR" not in html2
|
|
|
|
|
|
def test_jurnal_necesita_login(client):
|
|
r = client.get("/_fragments/jurnal")
|
|
assert r.status_code in (303, 401)
|
|
|
|
|
|
def test_deep_link_tab_jurnal(client):
|
|
_account_user("d@test.com")
|
|
_login(client, "d@test.com")
|
|
r = client.get("/?tab=jurnal")
|
|
assert r.status_code == 200
|
|
assert "Jurnal de aplicatie" in r.text
|