44f261e269
Curatare globala a comentariilor si docstring-urilor (app, tools, teste, scripturi): eliminate referintele la PRD-uri, US-xxx, task-uri istorice si review-uri; pastrata doar informatia functionala, formulata scurt. Regula adaugata in CLAUDE.md (sectiunea Stil). Fara modificari de cod sau comportament. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
122 lines
3.8 KiB
Python
122 lines
3.8 KiB
Python
"""Teste API DELETE + re-pune in coada, scoped + oracol scope/stare."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(monkeypatch):
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "life.db"))
|
|
monkeypatch.setenv("AUTOPASS_LOG_DIR", os.path.join(tmp, "logs"))
|
|
monkeypatch.setenv("AUTOPASS_REQUIRE_API_KEY", "true")
|
|
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 _account_with_key(name="Cont"):
|
|
from app.accounts import create_account
|
|
from app.auth import create_api_key
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
aid = create_account(conn, name, active=True)
|
|
key = create_api_key(conn, aid)
|
|
conn.commit()
|
|
return aid, key
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _body(**over):
|
|
prez = {"vin": "WVWZZZ1KZAW000123", "nr_inmatriculare": "B999TST",
|
|
"data_prestatie": "2026-06-15", "odometru_final": "123456",
|
|
"prestatii": [{"cod_prestatie": "OE-1"}]}
|
|
prez.update(over)
|
|
return {"rar_credentials": {"email": "x@y.ro", "password": "s"}, "prezentari": [prez]}
|
|
|
|
|
|
def _enqueue(client, key, **over):
|
|
r = client.post("/v1/prezentari", json=_body(**over), headers={"X-API-Key": key})
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["results"][0]["submission_id"]
|
|
|
|
|
|
def _force_status(sid, status):
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
conn.execute("UPDATE submissions SET status=? WHERE id=?", (status, sid))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_delete_scoped_pe_cheie(client):
|
|
_aid, key = _account_with_key()
|
|
sid = _enqueue(client, key)
|
|
_force_status(sid, "error")
|
|
r = client.request("DELETE", f"/v1/prezentari/{sid}", headers={"X-API-Key": key})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert body["submission_id"] == sid
|
|
assert body["status_anterior"] == "error"
|
|
|
|
|
|
def test_delete_sent_409(client):
|
|
_aid, key = _account_with_key()
|
|
sid = _enqueue(client, key)
|
|
_force_status(sid, "sent")
|
|
r = client.request("DELETE", f"/v1/prezentari/{sid}", headers={"X-API-Key": key})
|
|
assert r.status_code == 409
|
|
|
|
|
|
def test_delete_inexistent_404(client):
|
|
_aid, key = _account_with_key()
|
|
r = client.request("DELETE", "/v1/prezentari/99999", headers={"X-API-Key": key})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_delete_cross_account_sent_404(client):
|
|
"""Scope INAINTEA starii: randul `sent` al ALTUI cont -> 404 (nu 409, nu confirmam existenta)."""
|
|
aid_a, key_a = _account_with_key("A")
|
|
aid_b, key_b = _account_with_key("B")
|
|
sid = _enqueue(client, key_b)
|
|
_force_status(sid, "sent")
|
|
# contul A incearca sa stearga randul lui B (sent) -> 404, NU 409
|
|
r = client.request("DELETE", f"/v1/prezentari/{sid}", headers={"X-API-Key": key_a})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_repune_error_queued(client):
|
|
_aid, key = _account_with_key()
|
|
sid = _enqueue(client, key)
|
|
_force_status(sid, "error")
|
|
r = client.post(f"/v1/prezentari/{sid}/repune", headers={"X-API-Key": key})
|
|
assert r.status_code == 200
|
|
assert r.json()["status_nou"] == "queued"
|
|
|
|
|
|
def test_repune_inexistent_404(client):
|
|
_aid, key = _account_with_key()
|
|
r = client.post("/v1/prezentari/99999/repune", headers={"X-API-Key": key})
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_repune_sending_409(client):
|
|
_aid, key = _account_with_key()
|
|
sid = _enqueue(client, key)
|
|
_force_status(sid, "sending")
|
|
r = client.post(f"/v1/prezentari/{sid}/repune", headers={"X-API-Key": key})
|
|
assert r.status_code == 409
|