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>
104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
"""Teste retentie / purjare randuri ne-sent 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, "pb.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, status="queued"):
|
|
content = {"vin": "WVWZZZ1KZAW000123", "nr_inmatriculare": "B1",
|
|
"data_prestatie": "2026-06-15", "odometru_final": "1",
|
|
"prestatii": [{"cod_prestatie": "OE-1"}]}
|
|
cur = conn.execute(
|
|
"INSERT INTO submissions (idempotency_key, account_id, status, payload_json) VALUES (?, ?, ?, ?)",
|
|
(f"k-{os.urandom(4).hex()}", 1, status, json.dumps(content)),
|
|
)
|
|
return int(cur.lastrowid)
|
|
|
|
|
|
def test_error_primeste_purge_after(conn):
|
|
import app.worker.__main__ as w
|
|
sid = _ins(conn)
|
|
w.mark(conn, sid, "error", rar_error="creds gresite")
|
|
row = conn.execute("SELECT purge_after FROM submissions WHERE id=?", (sid,)).fetchone()
|
|
assert row["purge_after"] is not None
|
|
# In viitor, dar mai aproape de 30z decat de 90z
|
|
is_future = conn.execute(
|
|
"SELECT purge_after > datetime('now') AS ok FROM submissions WHERE id=?", (sid,)
|
|
).fetchone()["ok"]
|
|
assert is_future
|
|
|
|
|
|
def test_needs_data_si_needs_mapping_primesc_purge_after(conn):
|
|
import app.worker.__main__ as w
|
|
for st in ("needs_data", "needs_mapping"):
|
|
sid = _ins(conn)
|
|
w.mark(conn, sid, st)
|
|
row = conn.execute("SELECT purge_after FROM submissions WHERE id=?", (sid,)).fetchone()
|
|
assert row["purge_after"] is not None, f"{st} trebuie sa primeasca purge_after"
|
|
|
|
|
|
def test_purjare_sterge_error_expirat(conn):
|
|
import app.worker.__main__ as w
|
|
sid = _ins(conn)
|
|
conn.execute(
|
|
"UPDATE submissions SET status='error', purge_after=datetime('now','-1 day') WHERE id=?",
|
|
(sid,),
|
|
)
|
|
stats = w.purge_expired(conn)
|
|
assert stats["submissions_purged"] == 1
|
|
assert conn.execute("SELECT 1 FROM submissions WHERE id=?", (sid,)).fetchone() is None
|
|
|
|
|
|
def test_purjare_nu_atinge_queued_sau_sending(conn):
|
|
import app.worker.__main__ as w
|
|
# chiar daca au un purge_after rezidual in trecut, queued/sending NU se purjeaza
|
|
for st in ("queued", "sending"):
|
|
sid = _ins(conn)
|
|
conn.execute(
|
|
"UPDATE submissions SET status=?, purge_after=datetime('now','-1 day') WHERE id=?",
|
|
(st, sid),
|
|
)
|
|
stats = w.purge_expired(conn)
|
|
assert stats["submissions_purged"] == 0
|
|
|
|
|
|
def test_retentie_blocate_configurabila(conn, monkeypatch):
|
|
monkeypatch.setenv("AUTOPASS_BLOCKED_RETENTION_DAYS", "7")
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
import app.worker.__main__ as w
|
|
sid = _ins(conn)
|
|
w.mark(conn, sid, "error")
|
|
# purge_after ~ now + 7 zile: e inainte de now + 8 zile
|
|
row = conn.execute(
|
|
"SELECT purge_after < datetime('now','+8 days') AS ok FROM submissions WHERE id=?",
|
|
(sid,),
|
|
).fetchone()
|
|
assert row["ok"]
|