rar-autopass/tests/test_admin_lifecycle.py

"""Rute admin pentru ciclul de viata al conturilor —
block/archive/delete + bulk pe lista account_id, require_admin + CSRF + PRG, dev id=1 protejat.
"""

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, "test_admin_lifecycle.db"))
    monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true")
    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, follow_redirects=False) as c:
        yield c
    get_settings.cache_clear()


def _csrf(client, url="/admin"):
    resp = client.get(url, follow_redirects=True)
    m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text)
    assert m, f"csrf negasit in {url}"
    return m.group(1)


def _signup(client, name, email, password="parola_test_001"):
    from tests.conftest import make_test_cui
    tok = _csrf(client, "/signup")
    resp = client.post("/signup", data={"name": name, "cui": make_test_cui(email),
                                        "email": email, "parola": password,
                                        "consent": "1", "csrf_token": tok},
                       follow_redirects=True)
    assert resp.status_code == 200
    from app.db import get_connection
    conn = get_connection()
    try:
        row = conn.execute("SELECT account_id FROM users WHERE email=? COLLATE NOCASE",
                           (email,)).fetchone()
        return int(row["account_id"])
    finally:
        conn.close()


def _make_admin(account_id):
    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 _login(client, email, password="parola_test_001"):
    tok = _csrf(client, "/login")
    resp = client.post("/login", data={"email": email, "parola": password, "csrf_token": tok})
    assert resp.status_code == 303


def _status(account_id):
    from app.db import get_connection
    conn = get_connection()
    try:
        row = conn.execute("SELECT status FROM accounts WHERE id=?", (account_id,)).fetchone()
        return row["status"] if row else None
    finally:
        conn.close()


def _admin_login(client):
    admin_id = _signup(client, "Admin SA", "admin@test.ro")
    _make_admin(admin_id)
    _login(client, "admin@test.ro")
    return admin_id


@pytest.mark.parametrize("verb,expected", [
    ("block", "blocked"),
    ("archive", "archived"),
    ("delete", "deleted"),
])
def test_lifecycle_single(client, verb, expected):
    target = _signup(client, "Tinta SRL", "tinta@test.ro")
    _admin_login(client)
    tok = _csrf(client)
    resp = client.post(f"/admin/{verb}", data={"account_id": target, "csrf_token": tok})
    assert resp.status_code == 303
    assert _status(target) == expected


def test_bulk_pe_lista_account_id(client):
    a = _signup(client, "A SRL", "a@test.ro")
    b = _signup(client, "B SRL", "b@test.ro")
    _admin_login(client)
    tok = _csrf(client)
    resp = client.post("/admin/block", data={"account_id": [a, b], "csrf_token": tok})
    assert resp.status_code == 303
    assert _status(a) == "blocked" and _status(b) == "blocked"


def test_bulk_sare_contul_dev(client):
    target = _signup(client, "Tinta SRL", "t2@test.ro")
    _admin_login(client)
    tok = _csrf(client)
    # include id=1 (cont de sistem) in selectie -> sarit, fara eroare; tinta procesata
    resp = client.post("/admin/archive", data={"account_id": [1, target], "csrf_token": tok})
    assert resp.status_code == 303
    assert _status(1) == "active", "contul dev id=1 trebuie sa ramana neatins"
    assert _status(target) == "archived"


def test_non_admin_403(client):
    target = _signup(client, "Tinta SRL", "t3@test.ro")
    _signup(client, "Neadmin SRL", "plain@test.ro")
    _login(client, "plain@test.ro")
    # csrf de pe o pagina accesibila non-admin
    tok = _csrf(client, "/")
    resp = client.post("/admin/block", data={"account_id": target, "csrf_token": tok})
    assert resp.status_code == 403


def test_csrf_obligatoriu(client):
    target = _signup(client, "Tinta SRL", "t4@test.ro")
    _admin_login(client)
    resp = client.post("/admin/delete", data={"account_id": target})  # fara csrf_token
    assert resp.status_code != 303
    assert _status(target) != "deleted"