rar-autopass/tests/test_cont_medii.py

"""Teste US-008 (PRD 5.20): configurare medii RAR per cont — Testare + Productie.

Ruta testata: POST /cont/rar-medii

Teste:
  test_activeaza_si_salveaza_creds_per_env  -- creds salvate criptat, mediu marcat disponibil
  test_default_doar_dintre_disponibile       -- mediu implicit validat contra disponibilelor
  test_activare_prod_cere_confirmare         -- prima activare prod cere checkbox L.142
  test_creds_criptate_fara_echo              -- parola niciodata in clar in DB sau HTML
"""

from __future__ import annotations

import os
import re
import tempfile

import pytest
from cryptography.fernet import Fernet
from starlette.testclient import TestClient


# ---------------------------------------------------------------------------
# Fixture
# ---------------------------------------------------------------------------


@pytest.fixture()
def client(monkeypatch):
    """Client izolat cu DB temporara + cheie Fernet pentru criptare creds."""
    tmp = tempfile.mkdtemp()
    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t_medii.db"))
    monkeypatch.setenv("AUTOPASS_CREDS_KEY", Fernet.generate_key().decode())
    from app.config import get_settings
    from app import crypto

    get_settings.cache_clear()
    crypto.reset_cache()
    from app.main import app

    with TestClient(app, follow_redirects=False) as c:
        yield c

    get_settings.cache_clear()
    crypto.reset_cache()


# ---------------------------------------------------------------------------
# Helpere
# ---------------------------------------------------------------------------


def _create_account_user(
    name: str = "Service Test SRL",
    email: str = "user@test.com",
    password: str = "parolasecreta10",
):
    """Creeaza cont + user. Returneaza (acct_id, user_id)."""
    from app.accounts import create_account
    from app.users import create_user
    from app.db import get_connection

    conn = get_connection()
    try:
        acct_id = create_account(conn, name, active=True)
        user_id = create_user(conn, acct_id, email, password)
        return acct_id, user_id
    finally:
        conn.close()


def _login(client, email: str, password: str) -> None:
    """Face login real prin HTTP si seteaza cookie-ul de sesiune pe client."""
    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,
    })
    assert resp.status_code == 303, f"Login esuat: {resp.status_code} {resp.text[:200]}"


def _get_csrf(client) -> str:
    """Obtine CSRF token din fragmentul /_fragments/cont."""
    resp = client.get("/_fragments/cont")
    assert resp.status_code == 200, f"/_fragments/cont a returnat {resp.status_code}"
    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[:400]}"
    return m.group(1)


def _mock_login_ok(monkeypatch) -> None:
    """Monkeypatch _valideaza_login_rar sa returneze (True, None) fara RAR live."""
    import app.web.routes as routes_mod
    monkeypatch.setattr(routes_mod, "_valideaza_login_rar", lambda *a, **kw: (True, None))


# ---------------------------------------------------------------------------
# Teste
# ---------------------------------------------------------------------------


def test_activeaza_si_salveaza_creds_per_env(client, monkeypatch):
    """Activez Testare cu creds valide (mock) -> DB: rar_test_enabled=1, rar_creds_test_enc non-null.
    medii_disponibile si test_disponibil reflecta starea noua.
    """
    _mock_login_ok(monkeypatch)
    acct_id, _ = _create_account_user("Firma T1", "t1@test.com")
    _login(client, "t1@test.com", "parolasecreta10")

    csrf = _get_csrf(client)
    resp = client.post("/cont/rar-medii", data={
        "csrf_token": csrf,
        "test_enabled": "1",
        "test_email": "rar_test@firma.ro",
        "test_parola": "parolaRARtest",
        # prod_enabled absent -> rar_prod_enabled setat la 0
    })
    assert resp.status_code == 200

    from app.db import get_connection
    conn = get_connection()
    try:
        row = conn.execute(
            "SELECT rar_test_enabled, rar_creds_test_enc, rar_prod_enabled FROM accounts WHERE id=?",
            (acct_id,),
        ).fetchone()
    finally:
        conn.close()

    assert row["rar_test_enabled"] == 1, "rar_test_enabled trebuia setat la 1"
    assert row["rar_creds_test_enc"] is not None, "rar_creds_test_enc trebuia salvat"
    # Indicator test_disponibil sau mesaj succes in HTML
    assert "configurat" in resp.text or "salvate si validate" in resp.text, \
        f"Indicator 'configurat' sau mesaj succes lipsa: {resp.text[:600]}"


def test_default_doar_dintre_disponibile(client, monkeypatch):
    """Incerc sa setez rar_env_default pe un mediu indisponibil -> valoarea veche ramane + eroare.
    Setarea pe mediu disponibil reuseste.
    """
    _mock_login_ok(monkeypatch)
    acct_id, _ = _create_account_user("Firma T2", "t2@test.com")
    _login(client, "t2@test.com", "parolasecreta10")

    # Pasul 1: activeaza Testare cu creds + seteaza default=test (test va fi singurul disponibil)
    csrf = _get_csrf(client)
    resp1 = client.post("/cont/rar-medii", data={
        "csrf_token": csrf,
        "test_enabled": "1",
        "test_email": "rar_test@firma.ro",
        "test_parola": "parolaRAR123",
        "rar_env_default": "test",
        # prod_enabled absent -> rar_prod_enabled=0 (prod indisponibil)
    })
    assert resp1.status_code == 200
    assert "actualizat" in resp1.text.lower(), \
        f"Mesaj 'actualizat' asteptat pentru setarea default=test: {resp1.text[:500]}"

    from app.db import get_connection
    conn = get_connection()
    try:
        row1 = conn.execute(
            "SELECT rar_env_default FROM accounts WHERE id=?", (acct_id,)
        ).fetchone()
    finally:
        conn.close()
    assert row1["rar_env_default"] == "test", "rar_env_default trebuia setat la 'test'"

    # Pasul 2: incerc sa setez default=prod (prod indisponibil: enabled=0, fara creds)
    csrf = _get_csrf(client)
    resp2 = client.post("/cont/rar-medii", data={
        "csrf_token": csrf,
        "test_enabled": "1",   # re-trimit enabled fara creds noi (creds existente raman)
        "rar_env_default": "prod",
    })
    assert resp2.status_code == 200
    assert "disponibil" in resp2.text.lower(), \
        f"Eroare 'nu e disponibil' asteptata in raspuns: {resp2.text[:500]}"

    conn = get_connection()
    try:
        row2 = conn.execute(
            "SELECT rar_env_default FROM accounts WHERE id=?", (acct_id,)
        ).fetchone()
    finally:
        conn.close()
    assert row2["rar_env_default"] == "test", \
        "rar_env_default NU trebuia schimbat la 'prod' (mediu indisponibil)"


def test_activare_prod_cere_confirmare(client, monkeypatch):
    """Prima activare Productie (de la dezactivat) fara prod_confirmare -> NU se activeaza.
    Cu prod_confirmare=1 -> rar_prod_enabled devine 1.
    """
    _mock_login_ok(monkeypatch)
    acct_id, _ = _create_account_user("Firma T3", "t3@test.com")
    _login(client, "t3@test.com", "parolasecreta10")

    # Pasul 0: dezactiveaza prod (schema default=1, trebuie adus la 0 pt a testa confirmarea)
    csrf = _get_csrf(client)
    client.post("/cont/rar-medii", data={
        "csrf_token": csrf,
        # prod_enabled absent -> rar_prod_enabled setat la 0
        # test_enabled absent -> rar_test_enabled setat la 0
    })

    from app.db import get_connection
    conn = get_connection()
    try:
        row0 = conn.execute(
            "SELECT rar_prod_enabled FROM accounts WHERE id=?", (acct_id,)
        ).fetchone()
    finally:
        conn.close()
    assert row0["rar_prod_enabled"] == 0, "Prod trebuia dezactivat in pasul 0"

    # Pasul 1: incerc sa activez prod FARA confirmare -> refuzat
    csrf = _get_csrf(client)
    resp1 = client.post("/cont/rar-medii", data={
        "csrf_token": csrf,
        "prod_enabled": "1",
        "prod_email": "rar_prod@firma.ro",
        "prod_parola": "parolaRARprod",
        # prod_confirmare absent
    })
    assert resp1.status_code == 200
    text1 = resp1.text.lower()
    assert "confirmare" in text1 or "l.142" in text1 or "inteleg" in text1, \
        f"Mesaj de confirmare asteptat in raspuns: {resp1.text[:600]}"

    conn = get_connection()
    try:
        row1 = conn.execute(
            "SELECT rar_prod_enabled FROM accounts WHERE id=?", (acct_id,)
        ).fetchone()
    finally:
        conn.close()
    assert row1["rar_prod_enabled"] == 0, "rar_prod_enabled NU trebuia activat fara confirmare"

    # Pasul 2: activeaza cu confirmare -> reuseste
    csrf = _get_csrf(client)
    resp2 = client.post("/cont/rar-medii", data={
        "csrf_token": csrf,
        "prod_enabled": "1",
        "prod_email": "rar_prod@firma.ro",
        "prod_parola": "parolaRARprod",
        "prod_confirmare": "1",
    })
    assert resp2.status_code == 200

    conn = get_connection()
    try:
        row2 = conn.execute(
            "SELECT rar_prod_enabled FROM accounts WHERE id=?", (acct_id,)
        ).fetchone()
    finally:
        conn.close()
    assert row2["rar_prod_enabled"] == 1, "rar_prod_enabled trebuia activat cu confirmare"


def test_creds_criptate_fara_echo(client, monkeypatch):
    """Dupa salvare, rar_creds_test_enc e criptat (nu parola in clar) si
    parola NU apare in HTML-ul raspuns.
    """
    _mock_login_ok(monkeypatch)
    acct_id, _ = _create_account_user("Firma T4", "t4@test.com")
    _login(client, "t4@test.com", "parolasecreta10")

    parola_test = "SECRETPAROLATEST999"
    csrf = _get_csrf(client)
    resp = client.post("/cont/rar-medii", data={
        "csrf_token": csrf,
        "test_enabled": "1",
        "test_email": "rar_test@firma.ro",
        "test_parola": parola_test,
    })
    assert resp.status_code == 200

    # Parola NU trebuie sa apara in HTML-ul raspuns
    assert parola_test not in resp.text, \
        f"Parola apare in raspunsul HTML (echo interzis): {resp.text[:500]}"

    # In DB: rar_creds_test_enc e criptat (nu contine parola in clar)
    from app.db import get_connection
    from app.crypto import decrypt_creds

    conn = get_connection()
    try:
        row = conn.execute(
            "SELECT rar_creds_test_enc FROM accounts WHERE id=?", (acct_id,)
        ).fetchone()
    finally:
        conn.close()

    enc = row["rar_creds_test_enc"]
    assert enc is not None, "rar_creds_test_enc trebuia salvat"
    assert parola_test not in enc, "Parola in clar gasita in rar_creds_test_enc (neacceptat)"

    # Decriptarea trebuie sa recupereze parola originala
    creds = decrypt_creds(enc)
    assert creds is not None, "Decriptarea a returnat None"
    assert creds.get("password") == parola_test, \
        f"Parola decriptata nu corespunde: {creds!r}"