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>
117 lines
4.1 KiB
Python
117 lines
4.1 KiB
Python
"""Helper sterge / re-pune in coada randuri blocate."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture()
|
|
def env(monkeypatch):
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "sa.db"))
|
|
monkeypatch.setenv("AUTOPASS_LOG_DIR", os.path.join(tmp, "logs"))
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
from app.db import init_db
|
|
init_db()
|
|
yield monkeypatch
|
|
get_settings.cache_clear()
|
|
|
|
|
|
@pytest.fixture()
|
|
def conn(env):
|
|
from app.db import get_connection
|
|
c = get_connection()
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
def _ins(conn, *, account_id=1, status="error", valid=True, key=None):
|
|
if valid:
|
|
content = {"vin": "WVWZZZ1KZAW000123", "nr_inmatriculare": "B999TST",
|
|
"data_prestatie": "2026-06-15", "odometru_final": "123456",
|
|
"prestatii": [{"cod_prestatie": "OE-1"}]}
|
|
else:
|
|
content = {"vin": "BAD", "nr_inmatriculare": "B1",
|
|
"data_prestatie": "2026-06-15", "odometru_final": "1",
|
|
"prestatii": [{"cod_prestatie": "OE-1"}]}
|
|
sfx = key or os.urandom(4).hex()
|
|
cur = conn.execute(
|
|
"INSERT INTO submissions (idempotency_key, account_id, status, payload_json) VALUES (?, ?, ?, ?)",
|
|
(f"k-{sfx}", account_id, status, json.dumps(content)),
|
|
)
|
|
return int(cur.lastrowid)
|
|
|
|
|
|
def test_sterge_rand_error_scoped(conn):
|
|
from app.submissions_admin import delete_submission
|
|
sid = _ins(conn, status="error")
|
|
res = delete_submission(conn, 1, sid)
|
|
assert res["status_anterior"] == "error"
|
|
assert conn.execute("SELECT 1 FROM submissions WHERE id=?", (sid,)).fetchone() is None
|
|
|
|
|
|
def test_nu_sterge_sent_sau_sending(conn):
|
|
from app.submissions_admin import delete_submission, SubmissionStateConflict
|
|
for st in ("sent", "sending"):
|
|
sid = _ins(conn, status=st)
|
|
with pytest.raises(SubmissionStateConflict):
|
|
delete_submission(conn, 1, sid)
|
|
assert conn.execute("SELECT 1 FROM submissions WHERE id=?", (sid,)).fetchone() is not None
|
|
|
|
|
|
def test_scope_cross_account_404(conn):
|
|
from app.accounts import create_account
|
|
from app.submissions_admin import delete_submission, requeue_submission, SubmissionNotFound
|
|
other = create_account(conn, "Alt cont", active=True)
|
|
sid = _ins(conn, account_id=other, status="error")
|
|
with pytest.raises(SubmissionNotFound):
|
|
delete_submission(conn, 1, sid)
|
|
with pytest.raises(SubmissionNotFound):
|
|
requeue_submission(conn, 1, sid)
|
|
# randul altui cont ramane intact
|
|
assert conn.execute("SELECT 1 FROM submissions WHERE id=?", (sid,)).fetchone() is not None
|
|
|
|
|
|
def test_repune_error_devine_queued_reset_retry(conn):
|
|
from app.submissions_admin import requeue_submission
|
|
sid = _ins(conn, status="error", valid=True)
|
|
conn.execute(
|
|
"UPDATE submissions SET retry_count=5, next_attempt_at=datetime('now','+1 day'), "
|
|
"purge_after=datetime('now','+30 days') WHERE id=?",
|
|
(sid,),
|
|
)
|
|
res = requeue_submission(conn, 1, sid)
|
|
assert res["status_nou"] == "queued"
|
|
row = conn.execute(
|
|
"SELECT status, retry_count, next_attempt_at, purge_after FROM submissions WHERE id=?",
|
|
(sid,),
|
|
).fetchone()
|
|
assert row["status"] == "queued"
|
|
assert row["retry_count"] == 0
|
|
assert row["next_attempt_at"] is None
|
|
assert row["purge_after"] is None
|
|
|
|
|
|
def test_repune_re_ruleaza_classify(conn):
|
|
"""Continut invalid la re-pune -> ramane needs_data (classify re-ruleaza)."""
|
|
from app.submissions_admin import requeue_submission
|
|
sid = _ins(conn, status="error", valid=False)
|
|
res = requeue_submission(conn, 1, sid)
|
|
assert res["status_nou"] == "needs_data"
|
|
|
|
|
|
def test_actiunile_emit_eveniment(conn):
|
|
from app.submissions_admin import delete_submission, requeue_submission
|
|
sid1 = _ins(conn, status="error", valid=True)
|
|
requeue_submission(conn, 1, sid1)
|
|
sid2 = _ins(conn, status="error")
|
|
delete_submission(conn, 1, sid2)
|
|
tipuri = {r["tip"] for r in conn.execute("SELECT tip FROM app_events").fetchall()}
|
|
assert "submission_repus" in tipuri
|
|
assert "submission_sters" in tipuri
|