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>
106 lines
3.0 KiB
Python
106 lines
3.0 KiB
Python
"""Logger structurat central observ.log_event.
|
|
|
|
Scrie in app_events SI in log text, redacteaza creds/PII, filtreaza pe nivel.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture()
|
|
def env(monkeypatch):
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "observ.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 _events(tip=None):
|
|
from app.db import get_connection
|
|
conn = get_connection()
|
|
try:
|
|
if tip:
|
|
return conn.execute("SELECT * FROM app_events WHERE tip=?", (tip,)).fetchall()
|
|
return conn.execute("SELECT * FROM app_events").fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_log_event_scrie_in_db_si_fisier(env):
|
|
from app import observ
|
|
observ.log_event("test_eveniment", nivel="INFO", account_id=2, cod="X", mesaj="salut")
|
|
|
|
rows = _events("test_eveniment")
|
|
assert len(rows) == 1
|
|
assert rows[0]["account_id"] == 2
|
|
assert rows[0]["cod"] == "X"
|
|
assert rows[0]["mesaj"] == "salut"
|
|
assert rows[0]["sursa"] == "api"
|
|
|
|
# Log text scris
|
|
log_path = os.path.join(env, "logs", "app-api.log")
|
|
assert os.path.exists(log_path)
|
|
with open(log_path, encoding="utf-8") as f:
|
|
content = f.read()
|
|
assert "test_eveniment" in content
|
|
|
|
|
|
def test_log_event_redacteaza_pii_si_creds(env):
|
|
from app import observ
|
|
observ.log_event(
|
|
"cu_secrete",
|
|
context={
|
|
"password": "parolaSuperSecreta",
|
|
"rar_credentials": {"email": "a@b.ro", "password": "x"},
|
|
"vin": "WVWZZZ1KZAW000123",
|
|
"token": "eyJhbGciOi",
|
|
"count": 3,
|
|
},
|
|
)
|
|
rows = _events("cu_secrete")
|
|
assert len(rows) == 1
|
|
ctx = rows[0]["context_json"]
|
|
assert "parolaSuperSecreta" not in ctx
|
|
assert "***REDACTED***" in ctx
|
|
# VIN doar partial
|
|
assert "WVWZZZ1KZAW000123" not in ctx
|
|
assert "0123" in ctx
|
|
# campuri benigne raman
|
|
assert "count" in ctx
|
|
# fisierul text nu contine parola
|
|
log_path = os.path.join(env, "logs", "app-api.log")
|
|
with open(log_path, encoding="utf-8") as f:
|
|
content = f.read()
|
|
assert "parolaSuperSecreta" not in content
|
|
|
|
|
|
def test_nivel_filtrat_din_env(env, monkeypatch):
|
|
monkeypatch.setenv("AUTOPASS_LOG_LEVEL", "WARNING")
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
from app import observ
|
|
observ.log_event("sub_nivel", nivel="INFO")
|
|
observ.log_event("peste_nivel", nivel="ERROR")
|
|
assert len(_events("sub_nivel")) == 0
|
|
assert len(_events("peste_nivel")) == 1
|
|
|
|
|
|
def test_log_event_best_effort_nu_propaga(env, monkeypatch):
|
|
"""O eroare interna de scriere nu propaga (jurnal best-effort)."""
|
|
from app import observ
|
|
|
|
class Boom:
|
|
def execute(self, *a, **k):
|
|
raise RuntimeError("boom")
|
|
|
|
observ.log_event("nepasator", conn=Boom())
|