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>
95 lines
2.8 KiB
Python
95 lines
2.8 KiB
Python
"""Teste retentie / purjare jurnal + RotatingFileHandler."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging.handlers
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture()
|
|
def env(monkeypatch):
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "jr.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 tmp
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def _conn():
|
|
from app.db import get_connection
|
|
return get_connection()
|
|
|
|
|
|
def test_app_events_primesc_purge_after(env):
|
|
from app import observ
|
|
observ.log_event("test", account_id=1, mesaj="x")
|
|
conn = _conn()
|
|
try:
|
|
row = conn.execute("SELECT purge_after FROM app_events ORDER BY id DESC LIMIT 1").fetchone()
|
|
assert row["purge_after"] is not None
|
|
is_future = conn.execute(
|
|
"SELECT purge_after > datetime('now') AS ok FROM app_events ORDER BY id DESC LIMIT 1"
|
|
).fetchone()["ok"]
|
|
assert is_future
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_retentie_configurabila(env, monkeypatch):
|
|
monkeypatch.setenv("AUTOPASS_LOG_RETENTION_DAYS", "10")
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
from app import observ
|
|
observ.log_event("test", mesaj="x")
|
|
conn = _conn()
|
|
try:
|
|
ok = conn.execute(
|
|
"SELECT purge_after < datetime('now','+11 days') AS ok FROM app_events ORDER BY id DESC LIMIT 1"
|
|
).fetchone()["ok"]
|
|
assert ok
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_purjare_sterge_evenimente_expirate(env):
|
|
from app import observ
|
|
import app.worker.__main__ as w
|
|
observ.log_event("vechi", mesaj="x")
|
|
conn = _conn()
|
|
try:
|
|
conn.execute("UPDATE app_events SET purge_after=datetime('now','-1 day')")
|
|
conn.commit()
|
|
stats = w.purge_expired(conn)
|
|
assert stats["events_purged"] >= 1
|
|
assert conn.execute("SELECT COUNT(*) AS n FROM app_events").fetchone()["n"] == 0
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_purjare_pastreaza_neexpirate(env):
|
|
from app import observ
|
|
import app.worker.__main__ as w
|
|
observ.log_event("nou", mesaj="x") # purge_after in viitor (90z)
|
|
conn = _conn()
|
|
try:
|
|
stats = w.purge_expired(conn)
|
|
assert stats["events_purged"] == 0
|
|
assert conn.execute("SELECT COUNT(*) AS n FROM app_events").fetchone()["n"] == 1
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_log_text_foloseste_rotating_file_handler(env):
|
|
from app import observ
|
|
observ.log_event("rotativ", mesaj="x")
|
|
lg = observ._text_logger("api")
|
|
assert any(isinstance(h, logging.handlers.RotatingFileHandler) for h in lg.handlers), \
|
|
"logul text trebuie sa foloseasca RotatingFileHandler (rotatie in aplicatie)"
|