feat(web): hub integrare /integrare — exemple cod + retetar VFP + ping + export (PRD 5.1)

Pagina /integrare (tab autentificat, scoped pe cont): exemple cod multi-limbaj
(curl/Python/PHP/C#/Node) + retetar Visual FoxPro (MSXML2 + WinHttp) pe ambele
canale (prezentari JSON + import fisier), export Postman/OpenAPI/Swagger si buton
"Testeaza conexiunea".

- US-001: GET /v1/ping (readiness: account_id/mediu/autentificat_cu_cheie/
  are_creds_rar/ts) + GET /v1/integrare/postman.json (v2.1.0, allowlist 3 rute)
- US-002: app/web/integrare_examples.py pur (7 limbaje x 2 canale, drift-test
  is_required(), JSON compact pentru C#/VFP)
- US-003: tab "Integrare" IA pe 2 niveluri (limbaj->canal, VFP cu dialecte),
  copy din <pre><code>, empty-state CTA, export .cardlink, script scoped
- US-004: POST /integrare/test-cheie (account_for_key direct, scoped sesiune,
  no-echo cheie)

Backend trimitere (worker/masina stari/idempotenta/mapping) si schema neatinse.
568 teste pass. VERIFY context curat + E2E browser (Playwright) + code-review high.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Agent
2026-06-22 12:16:41 +00:00
parent be36c2c53b
commit f0786051f5
13 changed files with 2263 additions and 62 deletions

215
tests/test_integrare_api.py Normal file
View File

@@ -0,0 +1,215 @@
"""Teste US-001: endpoint-uri de integrare (GET /v1/ping + export Postman).
TDD — toate testele RED inainte de implementare.
"""
from __future__ import annotations
import json
import os
import tempfile
import pytest
from fastapi.testclient import TestClient
# --------------------------------------------------------------------------- #
# Fixture client izolat #
# --------------------------------------------------------------------------- #
@pytest.fixture()
def client(monkeypatch):
"""Client FastAPI cu DB temporara izolata, require_api_key=False (dev)."""
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "false")
monkeypatch.setenv("AUTOPASS_RAR_ENV", "test")
from app.config import get_settings
get_settings.cache_clear()
from app.main import app
with TestClient(app) as c:
yield c
get_settings.cache_clear()
@pytest.fixture()
def client_prod(monkeypatch):
"""Client cu require_api_key=True (mod prod)."""
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "true")
monkeypatch.setenv("AUTOPASS_RAR_ENV", "prod")
from app.config import get_settings
get_settings.cache_clear()
from app.main import app
with TestClient(app) as c:
yield c
get_settings.cache_clear()
def _creeaza_cheie(monkeypatch) -> str:
"""Seed cont id=1 + creeaza cheie API; intoarce cheia in clar."""
from app.db import get_connection
from app.auth import create_api_key
conn = get_connection()
try:
cheie = create_api_key(conn, 1)
conn.commit()
finally:
conn.close()
return cheie
def _seteaza_rar_creds(monkeypatch=None) -> None:
"""Seteaza rar_creds_enc pe contul id=1."""
from app.db import get_connection
from app.crypto import encrypt_creds
conn = get_connection()
try:
enc = encrypt_creds({"email": "test@rar.ro", "password": "secret"})
conn.execute("UPDATE accounts SET rar_creds_enc=? WHERE id=1", (enc,))
conn.commit()
finally:
conn.close()
# --------------------------------------------------------------------------- #
# Teste ping #
# --------------------------------------------------------------------------- #
def test_ping_cu_cheie_valida_200(client, monkeypatch):
"""GET /v1/ping cu X-API-Key valida -> 200 cu campurile cerute."""
cheie = _creeaza_cheie(monkeypatch)
r = client.get("/v1/ping", headers={"X-API-Key": cheie})
assert r.status_code == 200, r.text
body = r.json()
assert "account_id" in body
assert "mediu" in body
assert "autentificat_cu_cheie" in body
assert "are_creds_rar" in body
assert "ts" in body
assert body["autentificat_cu_cheie"] is True
assert body["mediu"] == "test"
def test_ping_cu_bearer_valid_200(client, monkeypatch):
"""GET /v1/ping cu Authorization: Bearer valida -> 200, autentificat_cu_cheie=True."""
cheie = _creeaza_cheie(monkeypatch)
r = client.get("/v1/ping", headers={"Authorization": f"Bearer {cheie}"})
assert r.status_code == 200, r.text
body = r.json()
assert body["autentificat_cu_cheie"] is True
assert body["account_id"] == 1
def test_ping_fara_cheie_dev_cont_implicit(client, monkeypatch):
"""Fara cheie, require_api_key=False -> cont 1, autentificat_cu_cheie=False."""
r = client.get("/v1/ping")
assert r.status_code == 200, r.text
body = r.json()
assert body["account_id"] == 1
assert body["autentificat_cu_cheie"] is False
def test_ping_x_api_key_gol_in_dev_cont_implicit(client, monkeypatch):
"""X-API-Key cu doar spatii = lipsa cheie -> cont 1, autentificat_cu_cheie=False in dev."""
r = client.get("/v1/ping", headers={"X-API-Key": " "})
assert r.status_code == 200, r.text
body = r.json()
assert body["account_id"] == 1
assert body["autentificat_cu_cheie"] is False
def test_ping_are_creds_rar_reflecta_contul(client, monkeypatch):
"""Cont fara creds -> are_creds_rar=False; dupa setare -> True."""
# Fara creds
r1 = client.get("/v1/ping")
assert r1.status_code == 200
assert r1.json()["are_creds_rar"] is False
# Seteaza creds pe cont 1
_seteaza_rar_creds()
r2 = client.get("/v1/ping")
assert r2.status_code == 200
assert r2.json()["are_creds_rar"] is True
def test_ping_cheie_invalida_401(client, monkeypatch):
"""Cheie invalida -> 401, indiferent de require_api_key."""
r = client.get("/v1/ping", headers={"X-API-Key": "rfak_cheie_falsa_xxxxxxxx"})
assert r.status_code == 401
def test_ping_prod_fara_cheie_401(client_prod, monkeypatch):
"""require_api_key=True, fara cheie -> 401."""
r = client_prod.get("/v1/ping")
assert r.status_code == 401
def test_ruta_ping_inregistrata_o_singura_data(client, monkeypatch):
"""Ruta /v1/ping trebuie sa apara exact o data in app.routes."""
from app.main import app
rute_ping = [r for r in app.routes if hasattr(r, "path") and r.path == "/v1/ping"]
assert len(rute_ping) == 1, f"Asteptat 1 ruta /v1/ping, gasit: {len(rute_ping)}"
# --------------------------------------------------------------------------- #
# Teste export Postman #
# --------------------------------------------------------------------------- #
def test_postman_export_json_valid(client, monkeypatch):
"""GET /v1/integrare/postman.json -> 200, Content-Type JSON, structura Postman v2.1.0."""
r = client.get("/v1/integrare/postman.json")
assert r.status_code == 200, r.text
assert "application/json" in r.headers.get("content-type", "")
body = r.json()
assert "info" in body
assert "v2.1.0" in body["info"].get("schema", ""), f"Schema gresita: {body['info'].get('schema')}"
def test_postman_contine_exact_trei_requesturi(client, monkeypatch):
"""Colectia Postman trebuie sa contina exact 3 requesturi cu headers si url corecte."""
r = client.get("/v1/integrare/postman.json")
assert r.status_code == 200
body = r.json()
items = body.get("item", [])
assert len(items) == 3, f"Asteptat 3 requesturi, gasit: {len(items)}"
for item in items:
req = item.get("request", {})
# Fiecare trebuie sa aiba header X-API-Key cu valoarea {{api_key}}
headers = req.get("header", [])
cheie_header = [h for h in headers if h.get("key") == "X-API-Key"]
assert len(cheie_header) == 1, f"Header X-API-Key lipsa in request '{item.get('name')}'"
assert cheie_header[0].get("value") == "{{api_key}}", \
f"Valoare header gresita: {cheie_header[0].get('value')}"
# URL-ul trebuie sa contina {{base_url}}
url = req.get("url", {})
url_raw = url.get("raw", "") if isinstance(url, dict) else str(url)
assert "{{base_url}}" in url_raw, \
f"{{{{base_url}}}} lipsa in url pentru '{item.get('name')}': {url_raw}"
def test_postman_nu_deriva_din_app_routes(client, monkeypatch):
"""Colectia Postman contine EXACT cele 3 rute allowlist, nu mai mult."""
r = client.get("/v1/integrare/postman.json")
assert r.status_code == 200
body = r.json()
items = body.get("item", [])
# Extrage URL-urile / path-urile din colectie
cai_expuse = set()
for item in items:
req = item.get("request", {})
url = req.get("url", {})
url_raw = url.get("raw", "") if isinstance(url, dict) else str(url)
# Normalizeaza: scoate {{base_url}} si parametrii query
cale = url_raw.replace("{{base_url}}", "").split("?")[0]
cai_expuse.add(cale)
# Allowlist: exact acestea 3
allowlist = {"/v1/prezentari", "/v1/import", "/v1/ping"}
assert cai_expuse == allowlist, \
f"Colectia expune cai neasteptate: {cai_expuse - allowlist} sau lipsesc: {allowlist - cai_expuse}"

View File

@@ -0,0 +1,233 @@
"""Teste pentru app.web.integrare_examples — modul PUR, fara I/O.
Ordinea: RED (fara implementare) -> GREEN (dupa implementare).
"""
import json
import pytest
from app.models import PrezentareIn
from app.web.integrare_examples import exemple
BASE_URL = "https://autopass.example.com"
ACCOUNT_ID = 7
RESULT = exemple(BASE_URL, ACCOUNT_ID)
LIMBAJE_OBLIGATORII = ["curl", "python", "php", "csharp", "node", "vfp_msxml", "vfp_winhttp"]
# ---------------------------------------------------------------------------
# Structura de baza
# ---------------------------------------------------------------------------
def test_toate_limbajele_prezente():
"""Toate limbajele obligatorii trebuie sa fie chei in dictionar."""
for limbaj in LIMBAJE_OBLIGATORII:
assert limbaj in RESULT, f"lipseste limbajul: {limbaj}"
def test_ambele_canale_per_limbaj():
"""Fiecare limbaj are atat 'prezentari' cat si 'import'."""
for limbaj in LIMBAJE_OBLIGATORII:
assert "prezentari" in RESULT[limbaj], f"{limbaj} lipseste canalul 'prezentari'"
assert "import" in RESULT[limbaj], f"{limbaj} lipseste canalul 'import'"
assert isinstance(RESULT[limbaj]["prezentari"], str), f"{limbaj}.prezentari nu e string"
assert isinstance(RESULT[limbaj]["import"], str), f"{limbaj}.import nu e string"
# ---------------------------------------------------------------------------
# curl — prezentari
# ---------------------------------------------------------------------------
def test_snippet_curl_prezentari_contine_endpoint_si_header():
"""Snippetul curl prezentari contine URL-ul corect si header-ul X-API-Key."""
snippet = RESULT["curl"]["prezentari"]
assert f"{BASE_URL}/v1/prezentari" in snippet, "lipseste endpoint-ul /v1/prezentari"
assert "X-API-Key" in snippet, "lipseste header-ul X-API-Key"
assert "rfak_..." in snippet, "placeholder-ul cheii trebuie sa fie rfak_..."
# ---------------------------------------------------------------------------
# python — import (multipart)
# ---------------------------------------------------------------------------
def test_snippet_python_import_upload_multipart():
"""Snippetul python import foloseste multipart (files=) pe /v1/import."""
snippet = RESULT["python"]["import"]
assert f"{BASE_URL}/v1/import" in snippet, "lipseste endpoint-ul /v1/import"
assert "files=" in snippet or 'files =' in snippet, "lipseste multipart files="
assert "rfak_..." in snippet, "placeholder-ul cheii trebuie sa fie rfak_..."
# ---------------------------------------------------------------------------
# VFP — doua dialecte distincte
# ---------------------------------------------------------------------------
def test_vfp_msxml_si_winhttp_distincte():
"""vfp_msxml foloseste MSXML2.ServerXMLHTTP.6.0; vfp_winhttp foloseste WinHttp.WinHttpRequest.5.1."""
msxml_snippet = RESULT["vfp_msxml"]["prezentari"]
winhttp_snippet = RESULT["vfp_winhttp"]["prezentari"]
assert "MSXML2.ServerXMLHTTP.6.0" in msxml_snippet, (
"vfp_msxml trebuie sa foloseasca MSXML2.ServerXMLHTTP.6.0"
)
assert "WinHttp.WinHttpRequest.5.1" in winhttp_snippet, (
"vfp_winhttp trebuie sa foloseasca WinHttp.WinHttpRequest.5.1"
)
# Distincte — nu acelasi obiect COM in ambele
assert "MSXML2.ServerXMLHTTP.6.0" not in winhttp_snippet, (
"vfp_winhttp nu trebuie sa contina MSXML2 (dialect gresit)"
)
assert "WinHttp.WinHttpRequest.5.1" not in msxml_snippet, (
"vfp_msxml nu trebuie sa contina WinHttp (dialect gresit)"
)
# ---------------------------------------------------------------------------
# Drift-test schema — campuri obligatorii din model
# ---------------------------------------------------------------------------
def test_payload_acopera_campurile_obligatorii_din_model():
"""Snippetul curl prezentari contine toate campurile obligatorii din PrezentareIn.
Deriva lista din model_fields pentru a fi rezistenta la schimbari de schema.
Campurile cu default (odometru_initial, obs, b64_image, sistem_reparat) nu sunt
obligatorii si nu trebuie sa cauzeze drift fals.
"""
obligatorii = {
camp
for camp, field in PrezentareIn.model_fields.items()
if field.is_required()
}
# Obligatorii asteptate conform spec: vin, nr_inmatriculare, data_prestatie,
# odometru_final, prestatii
snippet = RESULT["curl"]["prezentari"]
for camp in obligatorii:
assert camp in snippet, f"camp obligatoriu absent din snippet: {camp}"
# Credentiale RAR (email + password)
assert "email" in snippet, "camp 'email' absent din snippet (RarCredentials)"
assert "password" in snippet, "camp 'password' absent din snippet (RarCredentials)"
def test_prestatii_in_snippet_are_cod():
"""Snippetul prezentari contine cod_prestatie sau cod_op_service in payload."""
for limbaj in LIMBAJE_OBLIGATORII:
snippet = RESULT[limbaj]["prezentari"]
are_cod = "cod_prestatie" in snippet or "cod_op_service" in snippet
assert are_cod, f"{limbaj}.prezentari lipseste cod_prestatie / cod_op_service"
# ---------------------------------------------------------------------------
# Placeholder cheie
# ---------------------------------------------------------------------------
def test_placeholder_cheie_nu_e_valoare_reala():
"""Toate snippet-urile cu autentificare contin literal 'rfak_', nu o cheie reala."""
for limbaj in LIMBAJE_OBLIGATORII:
for canal in ["prezentari", "import"]:
snippet = RESULT[limbaj][canal]
assert "rfak_" in snippet, (
f"{limbaj}.{canal} lipseste placeholder-ul rfak_ (cheie API)"
)
# ---------------------------------------------------------------------------
# FIX-1 — C# si VFP: JSON compact (fara newline in string literal)
# ---------------------------------------------------------------------------
def test_csharp_payload_pe_o_singura_linie():
"""Snippetul C# prezentari: JSON-ul din var json = "..." e pe o singura linie.
Un JSON multi-linie intr-un string literal C# NU compileaza.
Verificam ca helper-ul compact produce un JSON fara newline
si ca acel JSON compact apare in snippetul C#.
"""
from app.web.integrare_examples import _payload_json_compact, _snippet_csharp_prezentari
compact = _payload_json_compact(ACCOUNT_ID)
assert "\n" not in compact, (
"_payload_json_compact produce newline — nu e compact"
)
snippet = _snippet_csharp_prezentari(BASE_URL, ACCOUNT_ID)
# JSON-ul compact (cu ghilimele escape-uite) trebuie sa fie substring in snippet
escaped = compact.replace('"', '\\"')
assert escaped in snippet, (
"JSON-ul compact (escaped) nu apare in snippetul C#; "
"probabil _snippet_csharp_prezentari inca foloseste _payload_json_str"
)
# Verificare directa: linia var json = "..." nu contine newline in interiorul ei
# In C# linia se termina cu "; (ghilimea de inchidere + punct-virgula)
for line in snippet.splitlines():
if 'var json = "' in line:
stripped = line.rstrip()
# Linia trebuie sa se inchida pe acelasi rand (cu "; sau ")
assert stripped.endswith('";') or stripped.endswith('"'), (
f"Linia 'var json = \"...' nu se termina pe acelasi rand: {line!r}"
)
break
else:
assert False, "Linia 'var json = \"...' nu a fost gasita in snippetul C#"
def test_vfp_payload_pe_o_singura_linie():
"""Snippetul VFP prezentari (ambele dialecte): cPayload = "..." e pe o linie.
Un string literal VFP multi-linie NU e valid.
Verificam ambele dialecte: vfp_msxml si vfp_winhttp.
"""
from app.web.integrare_examples import _payload_json_compact
compact = _payload_json_compact(ACCOUNT_ID)
# In VFP ghilimele se dubleaza
compact_vfp = compact.replace('"', '""')
for dialect in ("vfp_msxml", "vfp_winhttp"):
snippet = RESULT[dialect]["prezentari"]
# Payload-ul VFP (cu doubling) trebuie sa fie substring in snippet
assert compact_vfp in snippet, (
f"{dialect}: JSON-ul compact (cu \"\" doubling) nu apare in snippet; "
"probabil inca se foloseste _payload_json_str cu indent"
)
# Linia cPayload = "..." nu trebuie sa contina newline in interiorul valorii
for line in snippet.splitlines():
if 'cPayload = "' in line:
assert line.rstrip().endswith('"'), (
f"{dialect}: linia 'cPayload = \"...' nu se termina pe acelasi rand: {line!r}"
)
break
else:
assert False, f"{dialect}: linia 'cPayload = \"...' nu a fost gasita in snippet"
# ---------------------------------------------------------------------------
# FIX-2 — Node import: FormData/Blob globale, fara import din node:buffer
# ---------------------------------------------------------------------------
def test_node_import_nu_foloseste_node_buffer():
"""Snippetul Node import nu importa din 'node:buffer' si foloseste FormData globala.
'node:buffer' nu exporta FormData — `new FormData()` ar arunca TypeError.
In Node 18+ FormData si Blob sunt globale.
"""
snippet = RESULT["node"]["import"]
assert "node:buffer" not in snippet, (
"Snippetul Node import contine 'node:buffer' — invalid, FormData nu e acolo"
)
assert "new FormData()" in snippet, (
"Snippetul Node import nu contine 'new FormData()' — FormData trebuie folosita global"
)

View File

@@ -0,0 +1,265 @@
"""Teste US-004: POST /integrare/test-cheie — verifica cheia API lipita de utilizator.
TDD: testele se scriu INAINTE de implementare; la inceput pica (RED),
dupa implementare trec (GREEN).
Ruta testata:
- POST /integrare/test-cheie -> fragment HTML cu rezultat validare
"""
from __future__ import annotations
import os
import re
import tempfile
import pytest
from starlette.testclient import TestClient
# ---------------------------------------------------------------------------
# Fixture-uri
# ---------------------------------------------------------------------------
@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()
# ---------------------------------------------------------------------------
# Helper-e
# ---------------------------------------------------------------------------
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."""
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_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)
# ---------------------------------------------------------------------------
# Teste
# ---------------------------------------------------------------------------
def test_cheie_valida_a_contului_curent_ok(client):
"""Cheie activa a contului sesiunii -> raspuns de succes cu mention cont."""
acct_id, user_id, api_key = _create_account_user("cheie_ok@test.com")
_login(client, "cheie_ok@test.com", "parolasecreta10")
csrf = _get_csrf_from_fragment(client)
resp = client.post("/integrare/test-cheie", data={
"csrf_token": csrf,
"api_key": api_key,
})
assert resp.status_code == 200
body = resp.text.lower()
# Trebuie sa contina un mesaj de succes (clasa flash sau text "valida")
assert "valida" in body or "succes" in body, f"Mesaj de succes lipsa: {resp.text[:500]}"
# Trebuie sa mentioneze contul (account_id)
assert str(acct_id) in resp.text, f"account_id {acct_id} absent din raspuns: {resp.text[:500]}"
def test_cheie_a_altui_cont_respinsa(client):
"""Cheia unui alt cont -> mesaj neutru 'nu apartine contului tau', fara sa spuna care cont."""
acct_a, user_a, key_a = _create_account_user("altcont_a@test.com")
acct_b, user_b, key_b = _create_account_user("altcont_b@test.com")
# Logam user A; testam cu cheia lui B
_login(client, "altcont_a@test.com", "parolasecreta10")
csrf = _get_csrf_from_fragment(client)
resp = client.post("/integrare/test-cheie", data={
"csrf_token": csrf,
"api_key": key_b,
})
assert resp.status_code == 200
body = resp.text.lower()
# Trebuie sa respinga - mesaj ca nu apartine contului
assert "nu apartine" in body or "alt cont" in body or "nu este" in body, \
f"Mesaj de respingere lipsa: {resp.text[:500]}"
# NU trebuie sa dezvaluie ca e cheia contului B (nu mentionam alt account_id)
assert str(acct_b) not in resp.text, \
f"account_id-ul contului {acct_b} (terta) nu trebuia dezvaluit: {resp.text[:500]}"
def test_cheie_invalida_mesaj_clar(client):
"""Cheie inexistenta -> mesaj clar de eroare (nu valida)."""
acct_id, user_id, _ = _create_account_user("invalida@test.com")
_login(client, "invalida@test.com", "parolasecreta10")
csrf = _get_csrf_from_fragment(client)
resp = client.post("/integrare/test-cheie", data={
"csrf_token": csrf,
"api_key": "rfak_cheie_inexistenta_total_falsa",
})
assert resp.status_code == 200
body = resp.text.lower()
# Trebuie sa contina un mesaj de eroare (nu succes)
assert "invalida" in body or "inexistenta" in body or "negasita" in body or "eroare" in body, \
f"Mesaj de eroare lipsa: {resp.text[:500]}"
assert "valida" not in body or "invalida" in body, \
f"Raspuns da fals pozitiv 'valida': {resp.text[:500]}"
def test_cheie_revocata_dupa_rotire_respinsa(client):
"""Cheia veche (revocata dupa rotire) -> tratata ca invalida."""
acct_id, user_id, cheie_veche = _create_account_user("rotire@test.com")
_login(client, "rotire@test.com", "parolasecreta10")
# Rotim cheia (cheia_veche devine revocata)
from app.auth import rotate_api_key
from app.db import get_connection
conn = get_connection()
try:
cheie_noua = rotate_api_key(conn, acct_id)
finally:
conn.close()
csrf = _get_csrf_from_fragment(client)
resp = client.post("/integrare/test-cheie", data={
"csrf_token": csrf,
"api_key": cheie_veche,
})
assert resp.status_code == 200
body = resp.text.lower()
# Cheia veche revocata trebuie respinsa (invalida sau revocata)
assert "invalida" in body or "revocata" in body or "negasita" in body or "eroare" in body, \
f"Cheia revocata nu a fost respinsa: {resp.text[:500]}"
assert "valida — cont" not in body.lower(), \
f"Cheia revocata a primit fals pozitiv: {resp.text[:500]}"
def test_cheie_goala_nu_da_fals_pozitiv_in_dev(client):
"""Cheie goala sau whitespace -> eroare clara, NU 'valida cont 1' (fals pozitiv dev)."""
acct_id, user_id, _ = _create_account_user("goala@test.com")
_login(client, "goala@test.com", "parolasecreta10")
csrf = _get_csrf_from_fragment(client)
for cheie_goala in ["", " ", "\t"]:
resp = client.post("/integrare/test-cheie", data={
"csrf_token": csrf,
"api_key": cheie_goala,
})
assert resp.status_code == 200
body = resp.text.lower()
assert "valida — cont" not in body, \
f"Cheie goala '{repr(cheie_goala)}' da fals pozitiv: {resp.text[:500]}"
assert "goala" in body or "lipsa" in body or "invalida" in body or "eroare" in body, \
f"Cheie goala '{repr(cheie_goala)}' nu da mesaj de eroare: {resp.text[:500]}"
def test_fara_login_redirect_sau_401(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.post("/integrare/test-cheie", data={
"api_key": "rfak_orice",
})
assert resp.status_code in (303, 401), \
f"Trebuia redirect sau 401 fara login, got: {resp.status_code}"
if resp.status_code == 303:
assert "/login" in resp.headers.get("location", "")
get_settings.cache_clear()
def test_csrf_lipsa_respinsa(client):
"""Post cu CSRF invalid/lipsa si sesiune activa -> 403."""
acct_id, user_id, api_key = _create_account_user("csrf_test@test.com")
_login(client, "csrf_test@test.com", "parolasecreta10")
resp = client.post("/integrare/test-cheie", data={
"csrf_token": "token_gresit_total",
"api_key": api_key,
})
assert resp.status_code == 403, f"Trebuia 403 la CSRF invalid, got: {resp.status_code}"
def test_cheia_nu_apare_in_raspuns_sau_log(client):
"""Cheia lipita nu trebuie sa apara in body-ul raspunsului (no-echo)."""
acct_id, user_id, api_key = _create_account_user("noecho@test.com")
_login(client, "noecho@test.com", "parolasecreta10")
csrf = _get_csrf_from_fragment(client)
resp = client.post("/integrare/test-cheie", data={
"csrf_token": csrf,
"api_key": api_key,
})
assert resp.status_code == 200
# Cheia completa nu trebuie sa apara in raspuns (nici macar partial daca e unica)
assert api_key not in resp.text, \
f"Cheia API a aparut in body raspuns (no-echo violation): {resp.text[:500]}"

438
tests/test_web_integrare.py Normal file
View File

@@ -0,0 +1,438 @@
"""Teste US-003 — tab "Integrare" (hub documentatie + exemple cod).
TDD: testele se scriu INAINTE de implementare; la inceput pica (RED),
dupa implementare trec (GREEN).
Rute testate:
- GET / -> tab-bar contine tab "Integrare"
- GET /?tab=integrare -> panou randat server-side
- GET /_fragments/integrare -> fragment HTMX cu require_login
"""
from __future__ import annotations
import os
import re
import tempfile
import pytest
from starlette.testclient import TestClient
# =========================================================================== #
# Helpers comune #
# =========================================================================== #
def _create_account_user(email: str, password: str = "parolasecreta10"):
"""Creeaza cont + user. Intoarce (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, "Service Test Integrare", 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 = "parolasecreta10") -> None:
"""Autentificare HTTP; seteaza cookie de sesiune pe client."""
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)
assert m, "csrf_token negasit pe /login"
resp = client.post("/login", data={
"email": email,
"parola": password,
"csrf_token": m.group(1),
})
assert resp.status_code == 303, f"Login esuat: {resp.status_code} {resp.text[:200]}"
def _add_creds(acct_id: int) -> None:
"""Adauga credentiale RAR criptate pe cont (pentru test are_creds=True)."""
from app.db import get_connection
from app.crypto import encrypt_creds
conn = get_connection()
try:
creds_enc = encrypt_creds({"email": "test@rar.ro", "password": "secret"})
conn.execute(
"UPDATE accounts SET rar_creds_enc=? WHERE id=?",
(creds_enc, acct_id),
)
conn.commit()
finally:
conn.close()
def _add_api_key(acct_id: int) -> str:
"""Adauga o cheie API activa pe cont. Intoarce cheia bruta."""
from app.db import get_connection
from app.auth import create_api_key
conn = get_connection()
try:
key = create_api_key(conn, acct_id)
return key
finally:
conn.close()
@pytest.fixture()
def client(monkeypatch):
"""Client cu BD izolata si autentificare web activata."""
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "integrare_test.db"))
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()
# =========================================================================== #
# test_tab_integrare_in_nav #
# =========================================================================== #
def test_tab_integrare_in_nav(client):
"""Tab-bar-ul principal contine link-ul catre tab-ul Integrare."""
_create_account_user("nav_integrare@test.com")
_login(client, "nav_integrare@test.com")
resp = client.get("/")
assert resp.status_code == 200
html = resp.text
# Trebuie sa existe un link sau buton cu ?tab=integrare
assert "tab=integrare" in html, "tab-ul 'integrare' lipseste din tab-bar"
# Textul "Integrare" vizibil
assert "Integrare" in html
# =========================================================================== #
# test_deeplink_tab_integrare_randeaza_panou_server_side #
# =========================================================================== #
def test_deeplink_tab_integrare_randeaza_panou_server_side(client):
"""GET /?tab=integrare randeaza panoul server-side (nu redirectioneaza)."""
_create_account_user("deeplink_integrare@test.com")
_login(client, "deeplink_integrare@test.com")
resp = client.get("/?tab=integrare")
assert resp.status_code == 200
html = resp.text
# Panoul trebuie sa contina continut specific integrare
# (nu sa cada pe Acasa din cauza unui tab invalid)
assert "integrare" in html.lower()
# tab-ul activ trebuie sa fie "integrare"
assert 'aria-selected="true"' in html or "tab-activ" in html
# =========================================================================== #
# test_fragment_integrare_necesita_login #
# =========================================================================== #
def test_fragment_integrare_necesita_login(client):
"""GET /_fragments/integrare fara sesiune redirectioneaza la /login (require_login)."""
# Fara login -> trebuie 303 catre /login
resp = client.get("/_fragments/integrare")
assert resp.status_code == 303
assert "/login" in resp.headers.get("location", "")
# =========================================================================== #
# test_pagina_contine_account_id_si_endpoint_real #
# =========================================================================== #
def test_pagina_contine_account_id_si_endpoint_real(client):
"""Panoul Integrare afiseaza account_id-ul contului si URL-ul real al endpoint-ului."""
acct_id, _ = _create_account_user("endpoint_real@test.com")
_login(client, "endpoint_real@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
# account_id trebuie sa apara in pagina
assert str(acct_id) in html
# endpoint-ul /v1/prezentari trebuie sa apara
assert "/v1/prezentari" in html
# =========================================================================== #
# test_pagina_are_tab_limbaje_si_vfp_cu_dialecte #
# =========================================================================== #
def test_pagina_are_tab_limbaje_si_vfp_cu_dialecte(client):
"""Panoul Integrare are tab-uri pentru limbaje si VFP are sub-tab-uri (dialecte)."""
_create_account_user("limbaje@test.com")
_login(client, "limbaje@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
# Limbajele principale trebuie sa fie prezente
for limbaj in ("curl", "Python", "PHP"):
assert limbaj in html, f"Limbajul '{limbaj}' lipseste din panoul Integrare"
# VFP (Visual FoxPro) trebuie sa fie prezent
assert "VFP" in html or "FoxPro" in html or "Visual Fox" in html
# Dialectele VFP
assert "MSXML2" in html or "MSXML" in html
assert "WinHttp" in html
# =========================================================================== #
# test_canal_secundar_prezentari_si_import #
# =========================================================================== #
def test_canal_secundar_prezentari_si_import(client):
"""Fiecare limbaj are sectiuni pentru canalele Prezentari JSON si Import fisier."""
_create_account_user("canal_sec@test.com")
_login(client, "canal_sec@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
# Ambele canale trebuie sa fie vizibile in pagina
assert "prezentari" in html.lower() or "/v1/prezentari" in html
assert "import" in html.lower() or "/v1/import" in html
# =========================================================================== #
# test_export_card_openapi_postman_swagger #
# =========================================================================== #
def test_export_card_openapi_postman_swagger(client):
"""Panoul Integrare are card cu linkuri: /docs, /openapi.json, postman.json."""
_create_account_user("export_card@test.com")
_login(client, "export_card@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
# Linkuri de referinta
assert "/docs" in html
assert "/openapi.json" in html
assert "postman" in html.lower()
# =========================================================================== #
# test_buton_copiaza_citeste_din_pre_code #
# =========================================================================== #
def test_buton_copiaza_citeste_din_pre_code(client):
"""Snippet-urile au buton 'Copiaza' care citeste din <pre><code> (nu din data-*)."""
_create_account_user("copiaza@test.com")
_login(client, "copiaza@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
# Trebuie sa existe un buton de copiere
assert "Copiaza" in html or "copiaza" in html.lower()
# Snippet-urile trebuie sa fie in <pre><code> (nu text ascuns in data-*)
assert "<pre>" in html and "<code>" in html
# Butonul NU trebuie sa aiba data-cod sau data-snippet cu continutul
# (copiaza din DOM, nu din attribut)
assert 'data-cod="' not in html
assert 'data-snippet="' not in html
# =========================================================================== #
# test_empty_state_cta_cont_cand_fara_cheie_sau_creds #
# =========================================================================== #
def test_empty_state_cta_cont_cand_fara_cheie_sau_creds(client):
"""Fara cheie API sau credentiale RAR, panoul afiseaza CTA catre tab Cont."""
# Cont nou fara cheie si fara credentiale
_create_account_user("empty_state@test.com")
_login(client, "empty_state@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
# Trebuie sa existe un mesaj de empty-state cu link catre tab cont
assert "/?tab=cont" in html or "tab=cont" in html
# Mesajul trebuie sa atraga atentia ca lipsesc ceva
# (cheie sau credentiale)
lower = html.lower()
assert "cheie" in lower or "credentiale" in lower or "cont" in lower
# =========================================================================== #
# test_fara_culori_hardcodate_doar_tokens #
# =========================================================================== #
def test_fara_culori_hardcodate_doar_tokens(client):
"""Panoul Integrare nu contine culori hex hardcodate (#RRGGBB) — doar var(--...) tokens."""
_create_account_user("tokens_css@test.com")
_login(client, "tokens_css@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
# Cauta culori hex in stil inline (style="...#...") sau in tag style
# Pattern: # urmat de 3 sau 6 hex digits, in context CSS
import re as _re
# Cautam doar in atributele style="" inline si in taguri <style>
style_attrs = _re.findall(r'style="([^"]*)"', html)
style_tags = _re.findall(r'<style[^>]*>(.*?)</style>', html, _re.DOTALL)
all_css = " ".join(style_attrs) + " ".join(style_tags)
# Culori hex in CSS: #rgb sau #rrggbb (precedate de spatiu, :, sau ;)
hex_colors = _re.findall(r'(?<=[: ])#[0-9a-fA-F]{3,6}\b', all_css)
assert not hex_colors, (
f"Culori hex hardcodate gasite in _integrare.html: {hex_colors}. "
"Foloseste var(--...) tokens CSS."
)
# =========================================================================== #
# test_export_postman_are_atribut_download [FIX-2] #
# =========================================================================== #
def test_export_postman_are_atribut_download(client):
"""Linkul Postman (.json) contine atributul download (PRD US-003)."""
_create_account_user("postman_download@test.com")
_login(client, "postman_download@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
# Trebuie sa existe linkul postman.json cu atribut download
assert "postman.json" in html, "Linkul postman.json lipseste din pagina"
import re as _re
# Cautam <a cu href postman.json care contine si atribut download
postman_links = _re.findall(r'<a[^>]*postman\.json[^>]*>', html)
assert postman_links, "Tag-ul <a> cu postman.json nu a fost gasit"
assert any("download" in lnk for lnk in postman_links), (
f"Linkul Postman nu are atribut 'download'. Tag gasit: {postman_links}"
)
# =========================================================================== #
# test_export_card_foloseste_cardlink [FIX-1] #
# =========================================================================== #
def test_export_card_foloseste_cardlink(client):
"""Cardul Export & referinta foloseste componenta .cardlink (PRD US-003)."""
_create_account_user("export_cardlink@test.com")
_login(client, "export_cardlink@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
# Cardul export trebuie sa contina clasa cardlink
assert "cardlink" in html, (
"Clasa 'cardlink' lipseste din panoul Integrare. "
"Cardul Export & referinta trebuie sa foloseasca componenta .cardlink."
)
# Linkurile de export trebuie sa foloseasca clasa cardlink
import re as _re
cardlink_anchors = _re.findall(r'<a[^>]*cardlink[^>]*>', html)
assert cardlink_anchors, "Nu exista niciun <a class=\"cardlink\"> in pagina"
# =========================================================================== #
# test_microcopy_anticonfuzie_la_test_cheie [FIX-3] #
# =========================================================================== #
def test_microcopy_anticonfuzie_la_test_cheie(client):
"""Formularul 'Testeaza conexiunea' contine microcopy anti-confuzie (PRD US-004)."""
_create_account_user("microcopy@test.com")
_login(client, "microcopy@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
# Textul specific din PRD US-004
assert "Nu o salvam" in html, (
"Microcopy anti-confuzie lipseste din formularul 'Testeaza conexiunea'. "
"Trebuie sa contina: 'Nu o salvam si nu o memoram'."
)
# =========================================================================== #
# test_buton_copiaza_schimba_label_in_copiat [FIX-4] #
# =========================================================================== #
def test_buton_copiaza_schimba_label_in_copiat(client):
"""Scriptul JS schimba label-ul butonului in 'Copiat' la copiere (PRD US-003)."""
_create_account_user("copiat_label@test.com")
_login(client, "copiat_label@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
# Scriptul trebuie sa contina logica de schimbare a textContent pe buton
assert "textContent" in html, (
"Scriptul JS nu contine 'textContent' — schimbarea label-ului butonului lipseste."
)
assert "Copiat" in html, (
"Textul 'Copiat' lipseste din script — butonul nu isi schimba label-ul."
)
import re as _re
script_blocks = _re.findall(r'<script[^>]*>(.*?)</script>', html, _re.DOTALL)
script_text = " ".join(script_blocks)
# Verificam ca butonul (btn) isi schimba textContent (nu doar feedback div)
assert "btn.textContent" in script_text, (
"Scriptul JS nu contine 'btn.textContent' — label-ul butonului nu se schimba."
)
# Verificam valoarea setata pe buton
assert "'Copiat'" in script_text or '"Copiat"' in script_text, (
"Scriptul JS nu seteaza valoarea 'Copiat' pe buton."
)
# Verificam revenirea la 'Copiaza' dupa setTimeout
assert "setTimeout" in script_text, (
"Scriptul JS nu contine setTimeout — revenirea la 'Copiaza' dupa 2s lipseste."
)
assert "'Copiaza'" in script_text or '"Copiaza"' in script_text, (
"Scriptul JS nu seteaza revenirea la 'Copiaza' dupa timeout."
)
# =========================================================================== #
# test_script_integrare_scoped_pe_container [FIX-3] #
# =========================================================================== #
def test_script_integrare_scoped_pe_container(client):
"""Scriptul JS din _integrare.html este scoped pe #integrare-section.
Un querySelectorAll global ar ataca si tablist-ul principal din dashboard.html,
acumuland handlere si provocand dubla-legare pe fiecare swap HTMX.
Verificam ca scriptul porneste de la getElementById('integrare-section')
si NU face document.querySelectorAll('[role="tablist"]') global.
"""
_create_account_user("scoped_script@test.com")
_login(client, "scoped_script@test.com")
resp = client.get("/_fragments/integrare")
assert resp.status_code == 200
html = resp.text
import re as _re
script_blocks = _re.findall(r'<script[^>]*>(.*?)</script>', html, _re.DOTALL)
script_text = " ".join(script_blocks)
# Trebuie sa existe getElementById('integrare-section') ca root
assert "getElementById('integrare-section')" in script_text or \
'getElementById("integrare-section")' in script_text, (
"Scriptul JS nu contine getElementById('integrare-section') — "
"scoping-ul pe container lipseste"
)
# NU trebuie sa existe document.querySelectorAll cu '[role="tablist"]' global
# (adica fara a folosi root-ul)
has_global_tablist = bool(
_re.search(r'document\.querySelectorAll\([\'"][^"\']*\[role=["\']tablist["\'][^"\']*[\'"]\)', script_text)
)
assert not has_global_tablist, (
"Scriptul JS contine document.querySelectorAll('[role=\"tablist\"]') global — "
"trebuie sa foloseasca root.querySelectorAll dupa getElementById('integrare-section')"
)