102 lines
3.2 KiB
Python
102 lines
3.2 KiB
Python
"""P1-3 · 审计保留回归：超窗删/窗内留/dry-run/yes 门禁/分批/API 正常。"""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from django.core.management import call_command
|
|
from django.utils import timezone
|
|
from datetime import timedelta
|
|
|
|
|
|
def _make_log(tenant, days_ago, action="view"):
|
|
from apps.core.models import AuditLog
|
|
|
|
log = AuditLog.objects.create(
|
|
tenant=tenant, action=action, target_type="Product", target_id="1",
|
|
)
|
|
AuditLog.objects.filter(pk=log.pk).update(
|
|
created_at=timezone.now() - timedelta(days=days_ago)
|
|
)
|
|
return log
|
|
|
|
|
|
def test_dry_run_deletes_nothing(db, tenant):
|
|
_make_log(tenant, 400)
|
|
before = __import__("apps.core.models", fromlist=["AuditLog"]).AuditLog.objects.count()
|
|
call_command("purge_audit_logs", days=180, dry_run=True)
|
|
from apps.core.models import AuditLog
|
|
|
|
assert AuditLog.objects.count() == before
|
|
|
|
|
|
def test_without_yes_deletes_nothing(db, tenant):
|
|
from apps.core.models import AuditLog
|
|
|
|
_make_log(tenant, 400)
|
|
before = AuditLog.objects.count()
|
|
call_command("purge_audit_logs", days=180)
|
|
assert AuditLog.objects.count() == before
|
|
|
|
|
|
def test_window_boundary(db, tenant):
|
|
from apps.core.models import AuditLog
|
|
|
|
old = _make_log(tenant, 400)
|
|
fresh = _make_log(tenant, 10)
|
|
call_command("purge_audit_logs", days=180, yes=True)
|
|
ids = set(AuditLog.objects.values_list("id", flat=True))
|
|
assert old.id not in ids
|
|
assert fresh.id in ids
|
|
|
|
|
|
def test_batch_and_export(db, tenant, tmp_path):
|
|
from apps.core.models import AuditLog
|
|
|
|
for _ in range(5):
|
|
_make_log(tenant, 400)
|
|
out = tmp_path / "audit.jsonl"
|
|
call_command("purge_audit_logs", days=180, yes=True, batch=2,
|
|
export=str(out))
|
|
assert AuditLog.objects.filter(
|
|
created_at__lt=timezone.now() - timedelta(days=180)).count() == 0
|
|
lines = out.read_text(encoding="utf-8").strip().split("\n")
|
|
assert len(lines) == 5
|
|
row = json.loads(lines[0])
|
|
assert row["action"] == "view" and "created_at" in row
|
|
|
|
|
|
def test_purge_actor_registered_and_runs(db, tenant):
|
|
"""P1-3 定时接线：core.tasks actor 可直接调用，超窗日志被删。"""
|
|
from apps.core import tasks as core_tasks
|
|
|
|
_make_log(tenant, 400)
|
|
fresh = _make_log(tenant, 5)
|
|
core_tasks.purge_expired_audit_logs()
|
|
from apps.core.models import AuditLog
|
|
|
|
ids = set(AuditLog.objects.values_list("id", flat=True))
|
|
assert fresh.id in ids
|
|
assert AuditLog.objects.filter(
|
|
created_at__lt=timezone.now() - timedelta(days=180)).count() == 0
|
|
|
|
|
|
def test_audit_api_still_paginates_after_purge(db, tenant, user):
|
|
from apps.core.models import AuditLog
|
|
from rest_framework.test import APIClient
|
|
from rest_framework_simplejwt.tokens import RefreshToken
|
|
|
|
for _ in range(3):
|
|
_make_log(tenant, 400)
|
|
_make_log(tenant, 5)
|
|
call_command("purge_audit_logs", days=180, yes=True)
|
|
|
|
c = APIClient()
|
|
c.credentials(
|
|
HTTP_AUTHORIZATION=f"Bearer {RefreshToken.for_user(user).access_token}",
|
|
HTTP_X_TENANT_ID=tenant.code,
|
|
)
|
|
r = c.get("/api/v1/core/audit-logs/")
|
|
assert r.status_code == 200, r.content
|
|
body = r.json()
|
|
assert body["count"] == AuditLog.objects.filter(tenant=tenant).count() == 1
|