devplacepy/tests/e2e/admin/auditlog.py

# retoor <retoor@molodetz.nl>

from uuid import uuid4
from datetime import datetime, timezone
from tests.conftest import BASE_URL
from devplacepy.database import get_table
from devplacepy.utils import make_combined_slug
def seed_admin_news(count=3):
    news_table = get_table("news")
    news_table.delete()
    for i in range(count):
        eid = f"admin_test_news_{i}"
        if news_table.find_one(external_id=eid):
            continue
        uid = str(uuid4())
        title = f"Admin News Article {i}"
        slug = make_combined_slug(title, uid)
        news_table.insert(
            {
                "deleted_at": None,
                "deleted_by": None,
                "uid": uid,
                "slug": slug,
                "title": title,
                "external_id": eid,
                "grade": 5 + i,
                "status": "draft" if i == 0 else "published",
                "show_on_landing": 1 if i == 1 else 0,
                "source_name": "AdminTest",
                "synced_at": datetime.now(timezone.utc).isoformat(),
                "description": f"Test article {i} for admin tests.",
            }
        )
def seed_extra_users(count=30):
    users = get_table("users")
    existing_count = len(list(users.all()))
    if existing_count > 5:
        return
    for i in range(count):
        uid = str(uuid4())
        users.insert(
            {
                "uid": uid,
                "username": f"pagu_{i:04d}",
                "terms_version": "1",
                "email": f"pagu{i:04d}@test.devplace",
                "password_hash": "x",
                "role": "Member",
                "is_active": True,
                "created_at": datetime.now(timezone.utc).isoformat(),
            }
        )
def _seed_target_user():
    uid = str(uuid4())
    get_table("users").insert(
        {
            "uid": uid,
            "username": f"target_{uid[:8]}",
            "terms_version": "1",
            "email": f"target_{uid[:8]}@test.devplace",
            "password_hash": "x",
            "role": "Member",
            "is_active": True,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    return uid
import time
import pytest
import requests
from devplacepy.database import get_table, refresh_snapshot, set_setting
from devplacepy.utils import generate_uid, make_combined_slug
JSON_audit_log = {"Accept": "application/json"}
_counter_audit_log = [0]
@pytest.fixture(scope="module", autouse=True)
def _audit_test_settings(app_server):
    # On a fresh DB init_db skips seeding the operational/upload settings (its
    # `tables` snapshot predates site_settings creation), so those rows are
    # absent and the admin settings form would INSERT them as "" - which both
    # closes registration and makes consumers that do int("") crash. Seed sane
    # values here so the form's empty submissions are skipped (existing key), and
    # lift the per-IP rate limit since this file fires many mutating requests.
    for key, value in {
        "rate_limit_per_minute": "1000000",
        "rate_limit_window_seconds": "60",
        "registration_open": "1",
        "maintenance_mode": "0",
        "max_upload_size_mb": "10",
        "allowed_file_types": "",
        "max_attachments_per_resource": "10",
        "session_max_age_days": "7",
        "session_remember_days": "30",
        "news_service_interval": "3600",
        "news_grade_threshold": "7",
    }.items():
        set_setting(key, value)
    yield
def _db_user(name):
    # the user is created by the server subprocess; refresh the test-process
    # SQLite snapshot before reading it back across the process boundary.
    refresh_snapshot()
    return get_table("users").find_one(username=name)
def _unique(prefix="au"):
    _counter_audit_log[0] += 1
    return f"{prefix}{int(time.time() * 1000)}{_counter_audit_log[0]}"
def _member():
    name = _unique("aumem")
    s = requests.Session()
    s.post(
        f"{BASE_URL}/auth/signup",
        data={
            "username": name,
            "email": f"{name}@t.dev",
            "password": "secret123",
            "confirm_password": "secret123",
            "birth_date": "1990-01-01",
            "accept_terms": "1",
        },
        allow_redirects=True,
    )
    return s, name
def _member_key():
    _, name = _member()
    return _db_user(name)["api_key"]
def _admin(seeded_db):
    # authenticate via the seeded admin's API key (header auth) rather than a
    # login POST - GET reads are exempt from the rate limiter, so reusing this
    # across the file's many tests never counts against the per-IP write budget.
    key = _db_user("alice_test")["api_key"]
    s = requests.Session()
    s.headers.update({"X-API-KEY": key})
    return s
def _audit(admin, **params):
    r = admin.get(f"{BASE_URL}/admin/audit-log", headers=JSON_audit_log, params=params)
    assert r.status_code == 200, r.text[:300]
    return r.json()
def _find(admin, event_key, predicate):
    data = _audit(admin, event_key=event_key)
    for entry in data["entries"]:
        if predicate(entry):
            return entry
    return None
def _new_post(session, body="audited post body here"):
    return session.post(
        f"{BASE_URL}/posts/create",
        headers=JSON_audit_log,
        data={"title": _unique("aup"), "content": body, "topic": "devlog"},
    ).json()["data"]
def _new_project(session):
    return session.post(
        f"{BASE_URL}/projects/create",
        headers=JSON_audit_log,
        data={
            "title": _unique("aupr"),
            "description": "audited project description text",
            "project_type": "software",
            "status": "In Development",
            "platforms": "",
        },
    ).json()["data"]
def _seed_news_audit_log():
    uid = generate_uid()
    title = _unique("aunews")
    get_table("news").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": uid,
            "slug": make_combined_slug(title, uid),
            "title": title,
            "external_id": uid,
            "status": "draft",
            "featured": 0,
            "show_on_landing": 0,
            "grade": 5,
            "source_name": "AuditTest",
            "synced_at": datetime.now(timezone.utc).isoformat(),
            "description": "audited news article",
        }
    )
    refresh_snapshot()
    return uid


def test_admin_audit_log_page_loads(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/admin/audit-log", wait_until="domcontentloaded")
    assert page.is_visible("text=Audit Log") or page.is_visible("text=Audit")


def test_admin_audit_log_event_listing(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/admin/audit-log", wait_until="domcontentloaded")
    table = page.locator("table")
    assert table.is_visible() or page.is_visible("text=No audit events")


def test_audit_log_page_loads(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/admin/audit-log", wait_until="domcontentloaded")
    assert page.is_visible("h2:has-text('Audit Log')")
    assert page.locator("table.admin-table").count() == 1
    active = page.locator(
        ".sidebar-link.active[href='/admin/audit-log']"
    )
    assert active.count() == 1


def test_audit_log_filter_bar_renders(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/admin/audit-log", wait_until="domcontentloaded")
    for name in ("category", "event_key", "actor_role", "origin", "result"):
        assert page.locator(f"select[name='{name}']").count() == 1
    assert page.locator("input[name='q']").count() == 1
    assert page.locator("input[name='date_from']").count() == 1
    assert page.locator(".audit-filter-bar button:has-text('Filter')").count() == 1
    assert page.locator(".audit-filter-bar a:has-text('Clear')").count() == 1


def test_audit_log_filter_apply_search(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/admin/audit-log", wait_until="domcontentloaded")
    page.fill("input[name='q']", "auth")
    page.click(".audit-filter-bar button:has-text('Filter')")
    page.wait_for_url("**/admin/audit-log?**q=auth**", wait_until="domcontentloaded")
    assert "q=auth" in page.url
    assert page.locator("input[name='q']").input_value() == "auth"


def test_audit_log_filter_by_category_select(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/admin/audit-log", wait_until="domcontentloaded")
    page.select_option("select[name='category']", "auth")
    page.click(".audit-filter-bar button:has-text('Filter')")
    page.wait_for_url(
        "**/admin/audit-log?**category=auth**", wait_until="domcontentloaded"
    )
    rows = page.locator("table.admin-table tbody tr")
    assert rows.count() > 0
    # every visible event badge in the auth category begins with a known auth prefix
    badges = page.locator("table.admin-table tbody tr td .audit-badge").all_inner_texts()
    assert any(b.startswith("auth.") for b in badges)


def test_audit_log_clear_filters(alice):
    page, _ = alice
    page.goto(
        f"{BASE_URL}/admin/audit-log?category=auth&q=login",
        wait_until="domcontentloaded",
    )
    page.click(".audit-filter-bar a:has-text('Clear')")
    page.wait_for_url(f"{BASE_URL}/admin/audit-log", wait_until="domcontentloaded")
    assert page.url.rstrip("/") == f"{BASE_URL}/admin/audit-log"


def test_audit_log_view_detail(alice):
    page, _ = alice
    page.goto(f"{BASE_URL}/admin/audit-log", wait_until="domcontentloaded")
    page.locator("a:has-text('View')").first.click()
    page.wait_for_url("**/admin/audit-log/**", wait_until="domcontentloaded")
    assert page.is_visible("text=Related objects")
    assert page.locator(".audit-detail-grid").count() == 1
    assert page.locator(".audit-badge").first.is_visible()


def test_audit_log_detail_unknown_404(alice):
    page, _ = alice
    resp = page.goto(
        f"{BASE_URL}/admin/audit-log/nonexistent", wait_until="domcontentloaded"
    )
    assert resp.status == 404


def test_audit_log_guest_redirects_to_login(page, app_server):
    page.goto(f"{BASE_URL}/admin/audit-log", wait_until="domcontentloaded")
    assert page.url.startswith(f"{BASE_URL}/auth/login")


def test_audit_log_member_redirects_to_feed(bob):
    page, _ = bob
    page.goto(f"{BASE_URL}/admin/audit-log", wait_until="domcontentloaded")
    assert page.url.rstrip("/") == f"{BASE_URL}/feed"