rar-autopass/tests/test_web_signup.py

"""Teste US-003 (PRD 3.3): GET/POST /signup.

TDD: testele se scriu INAINTE de implementarea auth_routes.py; la inceput pica (RED),
dupa implementare trec (GREEN).
"""

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, "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(html: str) -> str:
    m = re.search(r'name="csrf_token"\s+value="([^"]+)"', html)
    if not m:
        m = re.search(r'value="([^"]+)"\s+name="csrf_token"', html)
    assert m, "csrf_token negasit in HTML"
    return m.group(1)


def test_signup_creeaza_cont_user_si_cheie(client):
    """POST /signup valid -> cont active=0, user, api_key create in DB; cheie rfak_ in raspuns."""
    resp = client.get("/signup")
    assert resp.status_code == 200
    token = _csrf(resp.text)

    resp = client.post("/signup", data={
        "name": "Service Auto Test",
        "cui": "RO12345678",
        "email": "test@example.com",
        "parola": "parolasecreta",
        "consent": "1",
        "csrf_token": token,
    })
    assert resp.status_code == 200
    assert "rfak_" in resp.text

    from app.db import get_connection
    conn = get_connection()
    try:
        acct = conn.execute(
            "SELECT * FROM accounts WHERE name='Service Auto Test'"
        ).fetchone()
        assert acct is not None
        assert acct["active"] == 0, "Contul trebuie creat inactive (in asteptare)"

        user = conn.execute(
            "SELECT * FROM users WHERE email='test@example.com'"
        ).fetchone()
        assert user is not None
        assert user["account_id"] == acct["id"]

        key = conn.execute(
            "SELECT * FROM api_keys WHERE account_id=?", (acct["id"],)
        ).fetchone()
        assert key is not None
        assert key["active"] == 1
    finally:
        conn.close()


def test_signup_email_duplicat_eroare(client):
    """Email duplicat -> ROLLBACK; COUNT(accounts) neschimbat (fara cont orfan)."""
    from tests.conftest import make_test_cui
    resp = client.get("/signup")
    token = _csrf(resp.text)
    client.post("/signup", data={
        "name": "Service A",
        "cui": make_test_cui("dup@example.com"),
        "email": "dup@example.com",
        "parola": "parolasecreta",
        "consent": "1",
        "csrf_token": token,
    })

    from app.db import get_connection
    conn = get_connection()
    count_before = conn.execute("SELECT COUNT(*) AS n FROM accounts").fetchone()["n"]
    conn.close()

    resp = client.get("/signup")
    token = _csrf(resp.text)
    resp2 = client.post("/signup", data={
        "name": "Service B",
        "cui": make_test_cui("dup-b@example.com"),
        "email": "dup@example.com",
        "parola": "altaparola123",
        "consent": "1",
        "csrf_token": token,
    })
    assert resp2.status_code in (200, 422)
    assert "rfak_" not in resp2.text

    conn = get_connection()
    count_after = conn.execute("SELECT COUNT(*) AS n FROM accounts").fetchone()["n"]
    conn.close()

    assert count_after == count_before, "Cont orfan creat la email duplicat (ROLLBACK a esuat)"


def test_signup_parola_scurta_eroare(client):
    """Parola sub 10 caractere -> eroare, fara creare cont/user."""
    resp = client.get("/signup")
    token = _csrf(resp.text)

    resp = client.post("/signup", data={
        "name": "Service Test",
        "email": "scurta@test.com",
        "parola": "scurt",
        "csrf_token": token,
    })
    assert resp.status_code in (200, 422)
    assert "rfak_" not in resp.text

    from app.db import get_connection
    conn = get_connection()
    try:
        acct = conn.execute(
            "SELECT * FROM accounts WHERE name='Service Test'"
        ).fetchone()
        assert acct is None, "Cont creat desi parola era prea scurta"
    finally:
        conn.close()


def test_signup_fara_consent_eroare(client):
    """Consimtamant GDPR lipsa -> 422, fara creare cont; mesaj despre Termeni/GDPR.

    Checkbox-ul de consimtamant trebuie validat server-side (functional, nu doar client-side):
    fara el contul nu se creeaza si planul/datele introduse se pastreaza in re-render.
    """
    from tests.conftest import make_test_cui
    resp = client.get("/signup")
    token = _csrf(resp.text)

    resp = client.post("/signup", data={
        "name": "Service Fara Consent",
        "cui": make_test_cui("fara-consent@test.com"),
        "email": "fara-consent@test.com",
        "parola": "parolasecreta123",
        # fara "consent"
        "csrf_token": token,
    })
    assert resp.status_code == 422
    assert "rfak_" not in resp.text
    assert "GDPR" in resp.text or "Termeni" in resp.text

    from app.db import get_connection
    conn = get_connection()
    try:
        acct = conn.execute(
            "SELECT * FROM accounts WHERE name='Service Fara Consent'"
        ).fetchone()
        assert acct is None, "Cont creat desi consimtamantul lipsea"
    finally:
        conn.close()


def test_signup_salveaza_requested_plan_si_consent(client):
    """POST /signup cu plan ales -> accounts.requested_plan = codul ales, consent_at setat,
    iar tier RAMANE 'free' (planul cerut NU acorda drepturi)."""
    from tests.conftest import make_test_cui
    resp = client.get("/signup")
    token = _csrf(resp.text)

    resp = client.post("/signup", data={
        "name": "Service Plan Pro",
        "cui": make_test_cui("plan-pro@test.com"),
        "email": "plan-pro@test.com",
        "parola": "parolasecreta123",
        "plan": "pro",
        "consent": "1",
        "csrf_token": token,
    })
    assert resp.status_code == 200
    assert "rfak_" in resp.text

    from app.db import get_connection
    conn = get_connection()
    try:
        acct = conn.execute(
            "SELECT * FROM accounts WHERE name='Service Plan Pro'"
        ).fetchone()
        assert acct is not None
        assert acct["requested_plan"] == "pro", "Planul cerut nu a fost salvat"
        assert acct["tier"] == "free", "tier NU trebuie urcat din planul cerut (doar dupa plata)"
        assert acct["consent_at"], "consent_at trebuie setat la signup cu consimtamant"
    finally:
        conn.close()


def test_cheie_afisata_o_data(client):
    """Cheia rfak_ apare in raspunsul POST /signup; GET /signup nu o contine."""
    from tests.conftest import make_test_cui
    resp = client.get("/signup")
    token = _csrf(resp.text)

    resp_post = client.post("/signup", data={
        "name": "Service Cheie",
        "cui": make_test_cui("cheie@test.com"),
        "email": "cheie@test.com",
        "parola": "parolasecreta",
        "consent": "1",
        "csrf_token": token,
    })
    assert resp_post.status_code == 200
    assert "rfak_" in resp_post.text, "Cheia trebuia afisata in raspunsul POST /signup"

    resp_get = client.get("/signup")
    assert "rfak_" not in resp_get.text, "GET /signup nu trebuie sa contina cheia (afisata o singura data)"