romfast/rar-autopass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(web): self-service cheie/creds + admin web + email signup (PRD 3.3b)

Browse Source 
US-007: rute web proprii /cont/roteste-cheie + /cont/rar-creds scoped pe
sesiune (C13), sectiune "Contul meu" cu cheie afisata o data.
US-010: rol admin (users.is_admin) + require_admin->403 + CLI set-admin +
bootstrap primul cont=admin (count_admins in BEGIN IMMEDIATE, anti-race).
US-011: panou /admin (activare/dezactivare conturi, CSRF + PRG), link admin
+ logout pe dashboard.
US-012: app/email.py notify_signup best-effort degradat fara SMTP + config smtp_*.
Fix: migrare defensiva users.is_admin/email_verified in _migrate.

VERIFY x2 context curat (PASS) + /code-review high. 393 teste pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Agent
2026-06-18 17:19:06 +00:00
parent 504b490d3b
commit b92055eb01
21 changed files with 1766 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

213
tests/test_admin_panel.py Normal file
View File

@@ -0,0 +1,213 @@
"""Teste US-011 (PRD 3.3b): panou admin web /admin — conturi in asteptare + activare.
TDD strict: testele se scriu INAINTE de implementare; la inceput pica (RED),
dupa implementare trec (GREEN).
Fisiere testate: app/web/admin_routes.py, app/web/templates/admin.html.
"""
from __future__ import annotations
import os
import re
import tempfile
import pytest
from starlette.testclient import TestClient
# ---------------------------------------------------------------------------
# Fixture client (web_auth_required=true -> CSRF enforce)
# ---------------------------------------------------------------------------
@pytest.fixture()
def client(monkeypatch):
"""TestClient pe aplicatia completa, cu DB izolata si web_auth_required=true."""
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "test_admin_panel.db"))
monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true")
# Ridica limita rate-limit pentru signup ca testele nu se blocheze intre ele
monkeypatch.setenv("AUTOPASS_SIGNUP_RATE_MAX", "100")
from app.config import get_settings
get_settings.cache_clear()
# Curata hits-urile rate-limit intre teste
from app.web import ratelimit
ratelimit._hits.clear()
from app.main import app
with TestClient(app, follow_redirects=False) as c:
yield c
get_settings.cache_clear()
# ---------------------------------------------------------------------------
# Helper-e
# ---------------------------------------------------------------------------
def _get_csrf(client: TestClient, url: str) -> str:
"""Extrage csrf_token din pagina HTML."""
resp = client.get(url, follow_redirects=True)
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, f"csrf_token negasit in {url}: {resp.text[:500]}"
return m.group(1)
def _signup(client: TestClient, name: str, email: str, password: str = "parola_test_001") -> int:
"""Creeaza cont via POST /signup si intoarce account_id."""
token = _get_csrf(client, "/signup")
resp = client.post("/signup", data={
"name": name,
"email": email,
"parola": password,
"csrf_token": token,
}, follow_redirects=True)
assert resp.status_code == 200, f"signup esuat: {resp.text[:300]}"
# Extrage account_id din raspuns (pagina afiseaza cheia rfak_ + account_id)
m = re.search(r"cont=(\d+)", resp.text)
if not m:
# fallback: citeste din DB
from app.db import get_connection
conn = get_connection()
row = conn.execute(
"SELECT account_id FROM users WHERE email=? COLLATE NOCASE", (email,)
).fetchone()
conn.close()
assert row, f"userul {email} nu a fost creat"
return int(row["account_id"])
return int(m.group(1))
def _login(client: TestClient, email: str, password: str = "parola_test_001") -> None:
"""Autentifica userul (seteaza sesiunea cookie)."""
token = _get_csrf(client, "/login")
resp = client.post("/login", data={
"email": email,
"parola": password,
"csrf_token": token,
}, follow_redirects=False)
assert resp.status_code == 303, f"login esuat cu {email}: {resp.status_code} {resp.text[:200]}"
def _make_admin(account_id: int) -> None:
"""Marcheaza contul ca admin direct in DB."""
from app.db import get_connection
from app.users import set_admin
conn = get_connection()
try:
set_admin(conn, account_id, is_admin=True)
conn.commit()
finally:
conn.close()
def _get_account_active(account_id: int) -> bool:
"""Citeste accounts.active din DB."""
from app.db import get_connection
conn = get_connection()
try:
row = conn.execute("SELECT active FROM accounts WHERE id=?", (account_id,)).fetchone()
return bool(row["active"]) if row else False
finally:
conn.close()
# ---------------------------------------------------------------------------
# Cazuri de test
# ---------------------------------------------------------------------------
def test_admin_vede_conturi_pending(client):
"""Admin logat -> GET /admin contine numele contului pending (active=0)."""
# Creeaza un cont pending (active=0 implicit la signup)
_signup(client, "Service Pending SRL", "pending@test.ro")
# Creeaza contul admin
admin_id = _signup(client, "Admin Corp SA", "admin@test.ro")
_make_admin(admin_id)
# Login ca admin
_login(client, "admin@test.ro")
resp = client.get("/admin")
assert resp.status_code == 200, f"GET /admin a returnat {resp.status_code}"
assert "Service Pending SRL" in resp.text, (
f"Contul pending nu apare in /admin. Raspuns: {resp.text[:600]}"
)
def test_activare_din_admin(client):
"""POST /admin/activate cu CSRF -> accounts.active=1; redirect 303."""
# Cont pending
pending_id = _signup(client, "Firma De Activat SRL", "firma@test.ro")
assert not _get_account_active(pending_id), "contul trebuie sa fie inactiv initial"
# Cont admin
admin_id = _signup(client, "Admin Activator SA", "activator@test.ro")
_make_admin(admin_id)
_login(client, "activator@test.ro")
# Obtine CSRF din pagina /admin
resp = client.get("/admin")
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 in /admin"
csrf = m.group(1)
resp2 = client.post("/admin/activate", data={
"account_id": str(pending_id),
"csrf_token": csrf,
})
assert resp2.status_code == 303, (
f"POST /admin/activate trebuia redirect 303, got {resp2.status_code}: {resp2.text[:300]}"
)
assert _get_account_active(pending_id), "contul trebuia sa fie activat dupa POST /admin/activate"
def test_non_admin_403(client):
"""User logat NON-admin -> GET /admin -> 403.
Creeaza intai un cont admin (bootstrap: primul user devine admin),
apoi un al doilea cont (non-admin) si verifica ca al doilea primeste 403.
"""
# Primul signup devine automat admin (bootstrap US-010)
_signup(client, "Admin Bootstrap SA", "bootstrap@test.ro")
# Al doilea user NU e admin
_signup(client, "User Simplu SRL", "user@test.ro")
_login(client, "user@test.ro")
resp = client.get("/admin")
assert resp.status_code == 403, (
f"User non-admin trebuia 403 pe /admin, got {resp.status_code}"
)
def test_admin_nelogat_redirect(client):
"""Fara sesiune -> GET /admin -> 303 redirect la /login."""
resp = client.get("/admin")
assert resp.status_code == 303, (
f"Nelogat pe /admin trebuia 303, got {resp.status_code}"
)
loc = resp.headers.get("location", "")
assert "/login" in loc, f"Redirect gresit: {loc}"
def test_activate_fara_csrf_403(client):
"""Admin logat, POST /admin/activate fara token CSRF -> 403."""
pending_id = _signup(client, "Firma Fara CSRF SRL", "nocsrf@test.ro")
admin_id = _signup(client, "Admin CSRF Test SA", "csrfadmin@test.ro")
_make_admin(admin_id)
_login(client, "csrfadmin@test.ro")
# POST fara token (sau token gol)
resp = client.post("/admin/activate", data={
"account_id": str(pending_id),
"csrf_token": "",
})
assert resp.status_code == 403, (
f"POST fara CSRF trebuia 403, got {resp.status_code}"
)

277
tests/test_admin_role.py Normal file
View File

@@ -0,0 +1,277 @@
"""Teste US-010 (PRD 3.3b): rol admin + bootstrap + guard require_admin.
Fisiere testate: app/users.py (helper-e admin), app/web/session.py (require_admin),
tools/account.py (subcomanda set-admin).
"""
from __future__ import annotations
import os
import tempfile
import pytest
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from starlette.middleware.sessions import SessionMiddleware
from starlette.testclient import TestClient
# ---------------------------------------------------------------------------
# Fixture DB
# ---------------------------------------------------------------------------
@pytest.fixture()
def conn(monkeypatch):
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "test_admin_role.db"))
from app.config import get_settings
get_settings.cache_clear()
from app.db import get_connection, init_db
init_db()
c = get_connection()
yield c
c.close()
get_settings.cache_clear()
@pytest.fixture()
def account_id(conn):
"""Cont de test (nu default id=1)."""
from app.accounts import create_account
return create_account(conn, "Service Test Admin")
@pytest.fixture()
def user_id(conn, account_id):
"""User de test pe contul de test."""
from app.users import create_user
return create_user(conn, account_id, "admin_test@exemplu.ro", "parola_sigura_123")
@pytest.fixture()
def env_cli(monkeypatch):
"""Fixture pentru CLI: DB separata + clear settings cache."""
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "test_admin_cli.db"))
from app.config import get_settings
get_settings.cache_clear()
from app.db import init_db
init_db()
yield
get_settings.cache_clear()
# ---------------------------------------------------------------------------
# Helper app minimal pentru require_admin
# ---------------------------------------------------------------------------
def _make_app_admin() -> FastAPI:
"""App FastAPI minimal cu ruta protejata de require_admin."""
mini = FastAPI()
mini.add_middleware(
SessionMiddleware,
secret_key="test-secret-admin",
session_cookie="autopass_session",
https_only=False,
same_site="strict",
)
from app.web.session import LoginRequired, AdminRequired
@mini.exception_handler(LoginRequired)
async def login_required_handler(request: Request, exc: LoginRequired):
return JSONResponse(status_code=401, content={"detail": "neautentificat"})
@mini.exception_handler(AdminRequired)
async def admin_required_handler(request: Request, exc: AdminRequired):
return JSONResponse(status_code=403, content={"detail": "acces interzis (necesita admin)"})
@mini.get("/set-session")
def set_sess(request: Request, account_id: int = 1, user_id: int = 1):
from app.web.session import set_session
set_session(request, account_id, user_id)
return {"ok": True}
@mini.get("/admin-only")
def admin_only(request: Request):
from app.web.session import require_admin
aid = require_admin(request)
return {"account_id": aid}
return mini
# ---------------------------------------------------------------------------
# Teste helper-e app/users.py
# ---------------------------------------------------------------------------
def test_count_admins_initial_zero(conn):
"""Fara niciun user cu is_admin=1, count_admins returneaza 0."""
from app.users import count_admins
assert count_admins(conn) == 0
def test_set_admin_marcheaza_userii_contului(conn, account_id, user_id):
"""set_admin(conn, account_id) seteaza is_admin=1 pe toti userii contului."""
from app.users import set_admin, count_admins
assert count_admins(conn) == 0
set_admin(conn, account_id, is_admin=True)
assert count_admins(conn) == 1
def test_create_user_is_admin_flag(conn, account_id):
"""create_user cu is_admin=True seteaza coloana la 1."""
from app.users import create_user, count_admins
create_user(conn, account_id, "newadmin@exemplu.ro", "parola_sigura_456", is_admin=True)
assert count_admins(conn) == 1
def test_is_account_admin(conn, account_id, user_id):
"""is_account_admin returneaza False inainte si True dupa set_admin."""
from app.users import is_account_admin, set_admin
assert is_account_admin(conn, account_id) is False
set_admin(conn, account_id, is_admin=True)
assert is_account_admin(conn, account_id) is True
def test_list_admin_emails(conn, account_id):
"""list_admin_emails returneaza emailurile userilor cu is_admin=1."""
from app.users import create_user, set_admin, list_admin_emails
uid = create_user(conn, account_id, "admin1@exemplu.ro", "parola_sigura_789")
# Inainte de set_admin -> lista goala
assert list_admin_emails(conn) == []
set_admin(conn, account_id, is_admin=True)
emails = list_admin_emails(conn)
assert "admin1@exemplu.ro" in emails
def test_set_admin_cont_inexistent_valueerror(conn):
"""set_admin pe cont care nu exista ridica ValueError."""
from app.users import set_admin
with pytest.raises(ValueError, match="cont inexistent"):
set_admin(conn, 9999, is_admin=True)
def test_set_admin_cont_fara_users_silentios(conn):
"""set_admin pe cont existent fara useri e no-op silentios (nu ridica exceptie)."""
from app.accounts import create_account
from app.users import set_admin, count_admins
acct_fara_user = create_account(conn, "Cont Fara User")
# Nu trebuie sa ridice
set_admin(conn, acct_fara_user, is_admin=True)
assert count_admins(conn) == 0 # niciun user modificat
# ---------------------------------------------------------------------------
# Teste require_admin (app/web/session.py) prin TestClient
# ---------------------------------------------------------------------------
@pytest.fixture()
def client_admin(monkeypatch):
"""TestClient pe app cu require_admin; DB cu cont admin + cont non-admin."""
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "test_require_admin.db"))
monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true")
from app.config import get_settings
get_settings.cache_clear()
from app.db import get_connection, init_db
init_db()
conn = get_connection()
from app.accounts import create_account
from app.users import create_user, set_admin
# Cont admin (id=2)
acct_admin = create_account(conn, "Admin Corp")
uid_admin = create_user(conn, acct_admin, "admin@corp.ro", "parola_admin_001")
set_admin(conn, acct_admin, is_admin=True)
# Cont non-admin (id=3)
acct_user = create_account(conn, "User Corp")
uid_user = create_user(conn, acct_user, "user@corp.ro", "parola_user_001")
conn.close()
app = _make_app_admin()
with TestClient(app, follow_redirects=False) as c:
yield c, acct_admin, uid_admin, acct_user, uid_user
get_settings.cache_clear()
def test_require_admin_blocheaza_non_admin(client_admin):
"""User logat NON-admin pe ruta admin -> 403."""
client, acct_admin, uid_admin, acct_user, uid_user = client_admin
client.get(f"/set-session?account_id={acct_user}&user_id={uid_user}")
resp = client.get("/admin-only")
assert resp.status_code == 403
assert "admin" in resp.json()["detail"]
def test_require_admin_lasa_admin(client_admin):
"""User logat ADMIN pe ruta admin -> 200 cu account_id."""
client, acct_admin, uid_admin, acct_user, uid_user = client_admin
client.get(f"/set-session?account_id={acct_admin}&user_id={uid_admin}")
resp = client.get("/admin-only")
assert resp.status_code == 200
assert resp.json()["account_id"] == acct_admin
def test_require_admin_nelogat_ridica_login_required(client_admin):
"""Fara sesiune, require_admin ridica LoginRequired (-> 401 in app-ul nostru de test)."""
client, *_ = client_admin
resp = client.get("/admin-only")
# In app-ul de test, LoginRequired -> 401
assert resp.status_code == 401
# ---------------------------------------------------------------------------
# Teste CLI tools/account.py set-admin
# ---------------------------------------------------------------------------
def _run_account(argv):
from tools.account import main
return main(argv)
def test_cli_set_admin_marcheaza_contul(env_cli, capsys):
"""CLI set-admin --account N seteaza is_admin=1 pe userii contului."""
# Creeaza cont cu user
from app.db import get_connection
from app.accounts import create_account
from app.users import create_user, is_account_admin
conn = get_connection()
acct_id = create_account(conn, "Service CLI Admin")
create_user(conn, acct_id, "cli_admin@corp.ro", "parola_cli_admin_1")
conn.close()
rc = _run_account(["set-admin", "--account", str(acct_id)])
out = capsys.readouterr().out
assert rc == 0
assert str(acct_id) in out or "admin" in out.lower()
conn2 = get_connection()
assert is_account_admin(conn2, acct_id) is True
conn2.close()
def test_cli_set_admin_remove(env_cli, capsys):
"""CLI set-admin --account N --remove scoate adminul."""
from app.db import get_connection
from app.accounts import create_account
from app.users import create_user, set_admin, is_account_admin
conn = get_connection()
acct_id = create_account(conn, "Service CLI Remove")
create_user(conn, acct_id, "remove@corp.ro", "parola_remove_001")
set_admin(conn, acct_id, is_admin=True)
conn.close()
rc = _run_account(["set-admin", "--account", str(acct_id), "--remove"])
assert rc == 0
conn2 = get_connection()
assert is_account_admin(conn2, acct_id) is False
conn2.close()
def test_cli_set_admin_cont_inexistent_exit_2(env_cli, capsys):
"""CLI set-admin pe cont inexistent -> exit code 2 + mesaj pe stderr."""
rc = _run_account(["set-admin", "--account", "9999"])
err = capsys.readouterr().err
assert rc == 2
assert "inexistent" in err or "eroare" in err.lower()

55
tests/test_dashboard_admin_link.py Normal file
View File

@@ -0,0 +1,55 @@
"""Test US-011 (discoverability): linkul 'Panou admin' apare pe dashboard doar pentru admini.
Completeaza intentia US-011 — adminul trebuie sa poata descoperi /admin din UI, nu doar
prin URL direct.
"""
from __future__ import annotations
import os
import tempfile
import pytest
from fastapi.testclient import TestClient
@pytest.fixture()
def env(monkeypatch):
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.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()
yield c, conn
conn.close()
get_settings.cache_clear()
def _account_with_user(conn, name, *, is_admin):
from app.accounts import create_account
from app.users import create_user
acct = create_account(conn, name)
email = f"{name.replace(' ', '').lower()}@test.ro"
create_user(conn, acct, email, "parolaSuperSecreta", is_admin=is_admin)
return acct
def test_admin_vede_link_panou_admin(env, monkeypatch):
client, conn = env
acct = _account_with_user(conn, "Admin Co", is_admin=True)
monkeypatch.setattr("app.web.routes.require_login", lambda r: acct)
r = client.get("/")
assert r.status_code == 200
assert 'href="/admin"' in r.text
def test_non_admin_nu_vede_link(env, monkeypatch):
client, conn = env
acct = _account_with_user(conn, "Service Normal", is_admin=False)
monkeypatch.setattr("app.web.routes.require_login", lambda r: acct)
r = client.get("/")
assert r.status_code == 200
assert 'href="/admin"' not in r.text

124
tests/test_migrate_users.py Normal file
View File

@@ -0,0 +1,124 @@
"""Teste migrare defensiva coloane users (is_admin, email_verified).
TDD: RED -> implementare in _migrate -> GREEN.
"""
from __future__ import annotations
import sqlite3
import tempfile
from pathlib import Path
import pytest
from app.db import _migrate, init_db, get_connection
from app.config import get_settings
# ---------------------------------------------------------------------------
# Test 1: _migrate adauga is_admin si email_verified pe o tabela users minima
# ---------------------------------------------------------------------------
def test_migrate_adauga_is_admin_pe_users_veche() -> None:
"""Pe un DB cu tabela users creata fara is_admin/email_verified,
_migrate trebuie sa adauge ambele coloane."""
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
db_path = Path(f.name)
conn = sqlite3.connect(str(db_path), isolation_level=None)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode = WAL")
conn.execute("PRAGMA foreign_keys = OFF") # evitam FK checks pe schema minima
# Tabela accounts minima (necesara pentru FK la users).
# Trebuie sa aiba cui + rar_creds_enc + active ca _migrate sa nu crape
# pe ALTER/CREATE INDEX care le refera inainte de blocul users.
conn.execute("""
CREATE TABLE accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
cui TEXT,
active INTEGER NOT NULL DEFAULT 1,
rar_creds_enc TEXT
)
""")
conn.execute("INSERT INTO accounts (id, name) VALUES (1, 'default')")
# Tabela submissions minima (necesara pentru _migrate -- coloane submissions)
conn.execute("""
CREATE TABLE submissions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
idempotency_key TEXT NOT NULL UNIQUE,
account_id INTEGER,
status TEXT NOT NULL DEFAULT 'queued',
payload_json TEXT NOT NULL,
rar_creds_enc TEXT,
rar_status_code INTEGER,
rar_error TEXT,
id_prezentare INTEGER,
retry_count INTEGER NOT NULL DEFAULT 0,
next_attempt_at TEXT,
sending_since TEXT,
purge_after TEXT,
batch_id INTEGER,
row_index INTEGER,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
# Tabela users MINIMA: fara is_admin, fara email_verified
conn.execute("""
CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
account_id INTEGER NOT NULL,
email TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
salt TEXT NOT NULL,
scrypt_params TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
try:
# Rulam migrarea
_migrate(conn)
# Verificam ca ambele coloane au fost adaugate
cols = {r["name"] for r in conn.execute("PRAGMA table_info(users)").fetchall()}
assert "is_admin" in cols, f"is_admin lipseste dupa migrare; coloane prezente: {cols}"
assert "email_verified" in cols, f"email_verified lipseste dupa migrare; coloane prezente: {cols}"
finally:
conn.close()
db_path.unlink(missing_ok=True)
# ---------------------------------------------------------------------------
# Test 2: _migrate este idempotent pe un DB initializat normal cu init_db()
# ---------------------------------------------------------------------------
def test_migrate_idempotent_pe_users_curenta(tmp_path: pytest.TempPathFactory, monkeypatch: pytest.MonkeyPatch) -> None:
"""Pe un DB initializat normal, re-apelarea _migrate nu ridica exceptie
si coloanele is_admin/email_verified raman prezente."""
db_file = tmp_path / "test_idem.db"
monkeypatch.setenv("AUTOPASS_DB_PATH", str(db_file))
# Resetam settings-ul cached ca sa preia noul AUTOPASS_DB_PATH
get_settings.cache_clear() # type: ignore[attr-defined]
try:
# Initializare normala
init_db()
# Re-apelam _migrate direct (idempotenta)
conn = get_connection()
try:
_migrate(conn) # nu trebuie sa ridice
cols = {r["name"] for r in conn.execute("PRAGMA table_info(users)").fetchall()}
assert "is_admin" in cols
assert "email_verified" in cols
finally:
conn.close()
finally:
get_settings.cache_clear() # type: ignore[attr-defined]

181
tests/test_signup_notify.py Normal file
View File

@@ -0,0 +1,181 @@
"""Teste US-012 (PRD 3.3b): notificare email admin la signup (degradat) + bootstrap admin.
TDD: testele se scriu INAINTE de implementare; la inceput pica (RED),
dupa implementare trec (GREEN).
"""
from __future__ import annotations
import os
import tempfile
import pytest
from starlette.testclient import TestClient
# ---------------------------------------------------------------------------
# Fixture client (pattern identic cu test_web_signup.py)
# ---------------------------------------------------------------------------
@pytest.fixture()
def client(monkeypatch):
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
monkeypatch.setenv("AUTOPASS_SIGNUP_RATE_MAX", "100")
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) as c:
yield c
get_settings.cache_clear()
def _csrf(c: TestClient) -> str:
"""Obtine un token CSRF proaspat de la GET /signup."""
import re
resp = c.get("/signup")
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 in HTML"
return m.group(1)
def _do_signup(c: TestClient, name: str, email: str, parola: str = "parolasecreta") -> object:
token = _csrf(c)
return c.post("/signup", data={
"name": name,
"email": email,
"parola": parola,
"csrf_token": token,
})
# ---------------------------------------------------------------------------
# Teste notify_signup (unitare, fara TestClient)
# ---------------------------------------------------------------------------
def test_notify_noop_fara_smtp(monkeypatch):
"""smtp_host None -> notify_signup nu ridica si nu incearca sa trimita."""
import smtplib
# Daca smtplib.SMTP e apelat, testul pica
def fake_smtp(*a, **kw):
raise AssertionError("smtplib.SMTP nu trebuia apelat fara smtp_host configurat")
monkeypatch.setattr(smtplib, "SMTP", fake_smtp)
# Asigura smtp_host = None
from app.config import get_settings
get_settings.cache_clear()
monkeypatch.setenv("AUTOPASS_SMTP_HOST", "") # string gol -> None in Settings
get_settings.cache_clear()
from app.email import notify_signup
# Nu trebuie sa ridice
notify_signup(["admin@test.com"], account_id=1, email="nou@test.com")
get_settings.cache_clear()
def test_notify_nu_blocheaza_la_eroare(monkeypatch):
"""smtp_host setat, SMTP ridica exceptie -> notify_signup returneaza normal (best-effort)."""
import smtplib
class FakeSMTP:
def __init__(self, *a, **kw):
raise ConnectionRefusedError("simulam eroare retea")
monkeypatch.setattr(smtplib, "SMTP", FakeSMTP)
monkeypatch.setenv("AUTOPASS_SMTP_HOST", "smtp.test.local")
monkeypatch.setenv("AUTOPASS_SMTP_PORT", "587")
monkeypatch.setenv("AUTOPASS_SMTP_FROM", "autopass@test.local")
from app.config import get_settings
get_settings.cache_clear()
from app.email import notify_signup
# Nu trebuie sa ridice, chiar daca SMTP esueaza
notify_signup(["admin@test.com"], account_id=5, email="nou@test.com")
get_settings.cache_clear()
# ---------------------------------------------------------------------------
# Teste bootstrap admin (prin TestClient)
# ---------------------------------------------------------------------------
def test_primul_signup_devine_admin(client):
"""Primul signup -> userul are is_admin=1 (count_admins==1).
Al doilea signup cu alt email -> is_admin=0 (count_admins ramane 1)."""
# Primul signup
resp = _do_signup(client, "Primul Service", "primul@test.com")
assert resp.status_code == 200
assert "rfak_" in resp.text
from app.db import get_connection
from app.users import count_admins
conn = get_connection()
try:
n_admins = count_admins(conn)
assert n_admins == 1, f"Dupa primul signup, count_admins trebuie sa fie 1, nu {n_admins}"
user = conn.execute(
"SELECT is_admin FROM users WHERE email='primul@test.com'"
).fetchone()
assert user is not None
assert user["is_admin"] == 1, "Primul user trebuie sa fie admin"
finally:
conn.close()
# Al doilea signup
resp2 = _do_signup(client, "Al Doilea Service", "aldoilea@test.com")
assert resp2.status_code == 200
assert "rfak_" in resp2.text
conn = get_connection()
try:
n_admins = count_admins(conn)
assert n_admins == 1, f"Dupa al doilea signup, count_admins trebuie sa ramana 1, nu {n_admins}"
user2 = conn.execute(
"SELECT is_admin FROM users WHERE email='aldoilea@test.com'"
).fetchone()
assert user2 is not None
assert user2["is_admin"] == 0, "Al doilea user NU trebuie sa fie admin"
finally:
conn.close()
# ---------------------------------------------------------------------------
# Teste C16 (log SIGNUP pastrat) si best-effort E2E
# ---------------------------------------------------------------------------
def test_signup_inca_logheaza_si_notifica(client, capsys):
"""Signup reusit -> stdout contine 'SIGNUP cont=' (C16 pastrat)."""
resp = _do_signup(client, "Service Log Test", "log@test.com")
assert resp.status_code == 200
assert "rfak_" in resp.text
captured = capsys.readouterr()
assert "SIGNUP cont=" in captured.out, (
f"Linia de log C16 'SIGNUP cont=' lipseste din stdout. Capturat: {captured.out!r}"
)
def test_signup_neblocat_de_notify(monkeypatch, client):
"""notify_signup ridica -> signup returneaza totusi 200 cu cheia (best-effort E2E)."""
# Monkeypatch notify_signup sa ridice
def notify_always_raises(*a, **kw):
raise RuntimeError("simulam eroare fatala in notify")
# Importam modulul inainte de monkeypatch
import app.email as email_mod
monkeypatch.setattr(email_mod, "notify_signup", notify_always_raises)
resp = _do_signup(client, "Service Robust", "robust@test.com")
assert resp.status_code == 200, (
f"Signup trebuia sa reuseasca indiferent de eroarea din notify. status={resp.status_code}"
)
assert "rfak_" in resp.text, "Cheia API trebuia afisata chiar daca notify a esuat"

232
tests/test_web_cont.py Normal file
View File

@@ -0,0 +1,232 @@
"""Teste US-007 (PRD 3.3b): sectiunea 'Contul meu' — rotire cheie API + creds RAR din UI.
TDD: testele se scriu INAINTE de implementare; la inceput pica (RED),
dupa implementare trec (GREEN).
Rute testate:
- GET /_fragments/cont -> card "Contul meu"
- POST /cont/roteste-cheie -> cheie noua afisata o singura data
- POST /cont/rar-creds -> seteaza rar_creds_enc per cont din sesiune
"""
from __future__ import annotations
import os
import re
import tempfile
import pytest
from starlette.testclient import TestClient
@pytest.fixture()
def client(monkeypatch):
"""Client fara web_auth_required (dev mode) — sesiunea se seteaza manual."""
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
from app.config import get_settings
get_settings.cache_clear()
from app.main import app
with TestClient(app, follow_redirects=False) as c:
yield c
get_settings.cache_clear()
@pytest.fixture()
def client_prod(monkeypatch):
"""Client cu web_auth_required=True (mod prod) — CSRF enforce."""
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true")
from app.config import get_settings
get_settings.cache_clear()
from app.main import app
with TestClient(app, follow_redirects=False) as c:
yield c
get_settings.cache_clear()
def _create_account_user(email: str = "user@test.com", password: str = "parolasecreta10"):
"""Creeaza cont + user + cheie API initiala. Intoarce (acct_id, user_id, api_key)."""
from app.accounts import create_account
from app.users import create_user
from app.auth import create_api_key
from app.db import get_connection
conn = get_connection()
try:
acct_id = create_account(conn, "Service Test", active=True)
user_id = create_user(conn, acct_id, email, password)
api_key = create_api_key(conn, acct_id)
return acct_id, user_id, api_key
finally:
conn.close()
def _login(client, email: str, password: str) -> None:
"""Face login real prin HTTP si seteaza cookie-ul de sesiune pe client."""
# Obtine CSRF token de pe pagina de login
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,
})
# 303 redirect la / inseamna login reusit
assert resp.status_code == 303, f"Login esuat: {resp.status_code} {resp.text[:200]}"
def _get_csrf_from_fragment(client) -> str:
"""Obtine CSRF token din fragmentul /_fragments/cont."""
resp = client.get("/_fragments/cont")
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, f"csrf_token negasit in /_fragments/cont: {resp.text[:500]}"
return m.group(1)
# ============================================================
# test_roteste_cheie_afisata_o_data
# ============================================================
def test_roteste_cheie_afisata_o_data(client):
"""User logat roteste cheia: raspunsul contine 'rfak_'; cheia veche revocata."""
acct_id, user_id, api_key_initiala = _create_account_user("roteste@test.com")
_login(client, "roteste@test.com", "parolasecreta10")
csrf = _get_csrf_from_fragment(client)
resp = client.post("/cont/roteste-cheie", data={"csrf_token": csrf})
assert resp.status_code == 200
assert "rfak_" in resp.text, f"Cheia noua nu apare in raspuns: {resp.text[:500]}"
# Verifica in DB: cheia veche revocata, una noua activa
from app.db import get_connection
conn = get_connection()
try:
rows = conn.execute(
"SELECT id, active FROM api_keys WHERE account_id=? ORDER BY id",
(acct_id,),
).fetchall()
# Trebuie sa avem minim 2 chei: cea initiala (active=0) si cea noua (active=1)
active_keys = [r for r in rows if r["active"] == 1]
inactive_keys = [r for r in rows if r["active"] == 0]
assert len(active_keys) == 1, f"Trebuia exact 1 cheie activa, gasit: {len(active_keys)}"
assert len(inactive_keys) >= 1, "Cheia veche trebuia revocata (active=0)"
finally:
conn.close()
# ============================================================
# test_set_creds_rar_din_sesiune
# ============================================================
def test_set_creds_rar_din_sesiune(client):
"""User logat seteaza creds RAR: accounts.rar_creds_enc != NULL, decriptabil."""
acct_id, user_id, _ = _create_account_user("creds@test.com")
_login(client, "creds@test.com", "parolasecreta10")
csrf = _get_csrf_from_fragment(client)
resp = client.post("/cont/rar-creds", data={
"csrf_token": csrf,
"rar_email": "user@rar.ro",
"rar_parola": "parolaRAR123",
})
assert resp.status_code == 200
# Mesaj de succes in raspuns
assert "succes" in resp.text.lower() or "salvat" in resp.text.lower() or "configurat" in resp.text.lower(), \
f"Mesaj de succes lipsa: {resp.text[:500]}"
# Verifica in DB: rar_creds_enc setat si decriptabil
from app.db import get_connection
from app.crypto import decrypt_creds
conn = get_connection()
try:
row = conn.execute(
"SELECT rar_creds_enc FROM accounts WHERE id=?", (acct_id,)
).fetchone()
assert row is not None
assert row["rar_creds_enc"] is not None, "rar_creds_enc trebuia setat"
creds = decrypt_creds(row["rar_creds_enc"])
assert creds is not None
assert creds.get("email") == "user@rar.ro"
assert creds.get("password") == "parolaRAR123"
finally:
conn.close()
# ============================================================
# test_creds_alt_cont_neafectat
# ============================================================
def test_creds_alt_cont_neafectat(client):
"""User A seteaza creds -> contul B ramane cu rar_creds_enc NULL."""
acct_a, user_a, _ = _create_account_user("userA@test.com")
acct_b, user_b, _ = _create_account_user("userB@test.com")
# Logam user A si setam creds
_login(client, "userA@test.com", "parolasecreta10")
csrf = _get_csrf_from_fragment(client)
resp = client.post("/cont/rar-creds", data={
"csrf_token": csrf,
"rar_email": "a@rar.ro",
"rar_parola": "parolaA123",
})
assert resp.status_code == 200
# Verifica: contul A are creds, contul B ramane NULL
from app.db import get_connection
conn = get_connection()
try:
row_a = conn.execute("SELECT rar_creds_enc FROM accounts WHERE id=?", (acct_a,)).fetchone()
row_b = conn.execute("SELECT rar_creds_enc FROM accounts WHERE id=?", (acct_b,)).fetchone()
assert row_a["rar_creds_enc"] is not None, "Contul A trebuia sa aiba creds"
assert row_b["rar_creds_enc"] is None, "Contul B nu trebuia atins"
finally:
conn.close()
# ============================================================
# test_roteste_fara_csrf_403_in_prod
# ============================================================
def test_roteste_fara_csrf_403_in_prod(client_prod):
"""Prod + sesiune autentificata + CSRF lipsa -> 403."""
# Cream cont + user
acct_id, user_id, _ = _create_account_user("csrf_test@test.com")
# Login real
_login(client_prod, "csrf_test@test.com", "parolasecreta10")
# POST fara csrf_token (sau cu token gresit)
resp = client_prod.post("/cont/roteste-cheie", data={"csrf_token": "token_gresit"})
assert resp.status_code == 403, f"Trebuia 403, got {resp.status_code}"
# ============================================================
# test_fragment_cont_nelogat_redirect
# ============================================================
def test_fragment_cont_nelogat_redirect(monkeypatch):
"""Fara sesiune + web_auth_required=True -> 303 redirect /login."""
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t_nl.db"))
monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true")
from app.config import get_settings
get_settings.cache_clear()
from app.main import app
with TestClient(app, follow_redirects=False) as c:
resp = c.get("/_fragments/cont")
assert resp.status_code == 303
assert "/login" in resp.headers.get("location", "")
get_settings.cache_clear()