rar-autopass/tests/test_web_lifecycle.py

"""Teste US-011 (PRD 5.6): butoane web sterge / re-pune in coada + bulk, scoped + CSRF."""

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, "wl.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]
    # set_session goleste sesiunea la login -> token CSRF nou, obtinut DUPA login.
    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="error"):
    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"}]}
        cur = conn.execute(
            "INSERT INTO submissions (idempotency_key, account_id, status, payload_json) VALUES (?, ?, ?, ?)",
            (f"k-{os.urandom(5).hex()}", account_id, status, json.dumps(content)),
        )
        conn.commit()
        return int(cur.lastrowid)
    finally:
        conn.close()


def test_buton_sterge_doar_pe_blocate(client):
    aid = _account_user("b@test.com")
    _login(client, "b@test.com")
    sid_err = _ins(aid, "error")
    sid_sent = _ins(aid, "sent")

    html_err = client.get(f"/_fragments/trimitere/{sid_err}").text
    assert "Re-pune in coada" in html_err
    assert "/trimitere/%d/sterge" % sid_err in html_err or f"/trimitere/{sid_err}/sterge" in html_err

    html_sent = client.get(f"/_fragments/trimitere/{sid_sent}").text
    assert "Re-pune in coada" not in html_sent
    assert f"/trimitere/{sid_sent}/sterge" not in html_sent


def test_repune_din_ui_scoped_sesiune(client):
    aid = _account_user("r@test.com")
    csrf = _login(client, "r@test.com")
    sid = _ins(aid, "error")
    r = client.post(f"/trimitere/{sid}/repune", data={"csrf_token": csrf})
    assert r.status_code == 200
    from app.db import get_connection
    conn = get_connection()
    try:
        st = conn.execute("SELECT status FROM submissions WHERE id=?", (sid,)).fetchone()["status"]
    finally:
        conn.close()
    assert st == "queued"


def test_sterge_din_ui(client):
    aid = _account_user("s@test.com")
    csrf = _login(client, "s@test.com")
    sid = _ins(aid, "error")
    r = client.post(f"/trimitere/{sid}/sterge", data={"csrf_token": csrf})
    assert r.status_code == 200
    from app.db import get_connection
    conn = get_connection()
    try:
        assert conn.execute("SELECT 1 FROM submissions WHERE id=?", (sid,)).fetchone() is None
    finally:
        conn.close()


def test_sterge_sent_409(client):
    aid = _account_user("se@test.com")
    csrf = _login(client, "se@test.com")
    sid = _ins(aid, "sent")
    r = client.post(f"/trimitere/{sid}/sterge", data={"csrf_token": csrf})
    assert r.status_code == 409


def test_csrf_enforce(client):
    aid = _account_user("c@test.com")
    _login(client, "c@test.com")
    sid = _ins(aid, "error")
    r = client.post(f"/trimitere/{sid}/sterge", data={"csrf_token": "gresit"})
    assert r.status_code == 403
    # randul ramane
    from app.db import get_connection
    conn = get_connection()
    try:
        assert conn.execute("SELECT 1 FROM submissions WHERE id=?", (sid,)).fetchone() is not None
    finally:
        conn.close()


def test_bulk_sterge_doar_blocate_scoped(client):
    aid = _account_user("bk@test.com")
    csrf = _login(client, "bk@test.com")
    sid_err = _ins(aid, "error")
    sid_nd = _ins(aid, "needs_data")
    sid_sent = _ins(aid, "sent")
    r = client.post(
        "/trimiteri/sterge-bulk",
        data={"submission_id": [str(sid_err), str(sid_nd), str(sid_sent)], "csrf_token": csrf},
    )
    assert r.status_code == 200
    from app.db import get_connection
    conn = get_connection()
    try:
        assert conn.execute("SELECT 1 FROM submissions WHERE id=?", (sid_err,)).fetchone() is None
        assert conn.execute("SELECT 1 FROM submissions WHERE id=?", (sid_nd,)).fetchone() is None
        # sent NU se sterge
        assert conn.execute("SELECT 1 FROM submissions WHERE id=?", (sid_sent,)).fetchone() is not None
    finally:
        conn.close()


def test_repune_cross_account_404(client):
    aid = _account_user("x1@test.com", name="X1")
    other = _account_user("x2@test.com", name="X2")
    csrf = _login(client, "x1@test.com")
    sid_other = _ins(other, "error")
    r = client.post(f"/trimitere/{sid_other}/repune", data={"csrf_token": csrf})
    assert r.status_code == 404