feat: add audit log tables, indexes, and CLI/content recording hooks
This commit is contained in:
@@ -0,0 +1,40 @@
|
||||
from pathlib import Path
|
||||
from tests.conftest import BASE_URL
|
||||
|
||||
|
||||
def test_attachment_upload_endpoint(alice):
|
||||
page, user = alice
|
||||
page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
|
||||
page.locator(".feed-fab").first.wait_for(state="visible", timeout=10000)
|
||||
page.locator(".feed-fab").click()
|
||||
page.locator("#create-post-modal").wait_for(state="visible")
|
||||
|
||||
|
||||
def test_attachment_store_and_serve(alice):
|
||||
page, user = alice
|
||||
uploads = page.context.request.post(f"{BASE_URL}/attachments/upload")
|
||||
|
||||
|
||||
def test_post_attachments_display(alice):
|
||||
page, user = alice
|
||||
page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
|
||||
|
||||
|
||||
def test_project_attachments_display(alice):
|
||||
page, user = alice
|
||||
page.goto(f"{BASE_URL}/projects", wait_until="domcontentloaded")
|
||||
|
||||
|
||||
def test_bug_attachments_display(alice):
|
||||
page, user = alice
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
|
||||
|
||||
def test_attachment_uploader_component_present(alice):
|
||||
page, user = alice
|
||||
page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
|
||||
page.locator(".feed-fab").wait_for(state="visible", timeout=10000)
|
||||
page.locator(".feed-fab").click()
|
||||
page.locator("#create-post-modal").wait_for(state="visible")
|
||||
page.locator(".attachment-upload-area").wait_for(state="visible", timeout=5000)
|
||||
page.locator(".attachment-upload-btn").wait_for(state="visible", timeout=5000)
|
||||
@@ -0,0 +1,929 @@
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import get_table, refresh_snapshot, set_setting
|
||||
from devplacepy.utils import generate_uid, make_combined_slug
|
||||
|
||||
JSON = {"Accept": "application/json"}
|
||||
_counter = [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[0] += 1
|
||||
return f"{prefix}{int(time.time() * 1000)}{_counter[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",
|
||||
},
|
||||
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, 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
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# security and access control #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_audit_log_requires_admin(seeded_db):
|
||||
# guest: JSON -> 401, browser -> redirect to login
|
||||
assert (
|
||||
requests.get(
|
||||
f"{BASE_URL}/admin/audit-log", headers=JSON, allow_redirects=False
|
||||
).status_code
|
||||
== 401
|
||||
)
|
||||
assert (
|
||||
requests.get(
|
||||
f"{BASE_URL}/admin/audit-log", allow_redirects=False
|
||||
).status_code
|
||||
== 303
|
||||
)
|
||||
# authenticated member -> 403 for JSON
|
||||
key = _member_key()
|
||||
assert (
|
||||
requests.get(
|
||||
f"{BASE_URL}/admin/audit-log",
|
||||
headers={**JSON, "X-API-KEY": key},
|
||||
allow_redirects=False,
|
||||
).status_code
|
||||
== 403
|
||||
)
|
||||
# admin -> 200
|
||||
admin = _admin(seeded_db)
|
||||
assert admin.get(f"{BASE_URL}/admin/audit-log", headers=JSON).status_code == 200
|
||||
|
||||
|
||||
def test_audit_event_detail_requires_admin(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
uid = _audit(admin)["entries"][0]["uid"]
|
||||
assert (
|
||||
requests.get(
|
||||
f"{BASE_URL}/admin/audit-log/{uid}", headers=JSON, allow_redirects=False
|
||||
).status_code
|
||||
== 401
|
||||
)
|
||||
key = _member_key()
|
||||
assert (
|
||||
requests.get(
|
||||
f"{BASE_URL}/admin/audit-log/{uid}",
|
||||
headers={**JSON, "X-API-KEY": key},
|
||||
allow_redirects=False,
|
||||
).status_code
|
||||
== 403
|
||||
)
|
||||
assert admin.get(f"{BASE_URL}/admin/audit-log/{uid}", headers=JSON).status_code == 200
|
||||
|
||||
|
||||
def test_audit_event_detail_404(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
r = admin.get(f"{BASE_URL}/admin/audit-log/does-not-exist-uid", headers=JSON)
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# JSON shape #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_audit_log_json_shape(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
data = _audit(admin)
|
||||
assert set(["entries", "pagination", "filters", "options"]).issubset(data.keys())
|
||||
assert isinstance(data["entries"], list)
|
||||
assert data["pagination"]["per_page"] == 25
|
||||
entry = data["entries"][0]
|
||||
for field in ("uid", "event_key", "category", "actor_kind", "actor_role", "result"):
|
||||
assert field in entry
|
||||
for opt in ("event_key", "category", "actor_role", "origin", "result"):
|
||||
assert opt in data["options"]
|
||||
|
||||
|
||||
def test_audit_event_detail_json_shape(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
uid = _audit(admin)["entries"][0]["uid"]
|
||||
detail = admin.get(f"{BASE_URL}/admin/audit-log/{uid}", headers=JSON).json()
|
||||
assert "event" in detail and "links" in detail
|
||||
assert detail["event"]["uid"] == uid
|
||||
assert isinstance(detail["links"], list)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# emission - the audit log actually records actions #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_signup_is_recorded(seeded_db):
|
||||
_, name = _member()
|
||||
admin = _admin(seeded_db)
|
||||
event = _find(admin, "auth.signup", lambda e: e.get("target_label") == name)
|
||||
assert event is not None
|
||||
assert event["category"] == "auth"
|
||||
assert event["result"] == "success"
|
||||
|
||||
|
||||
def test_login_failure_is_recorded(seeded_db):
|
||||
email = f"{_unique('aufail')}@t.dev"
|
||||
requests.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
data={"email": email, "password": "wrongpassword"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
admin = _admin(seeded_db)
|
||||
event = _find(
|
||||
admin,
|
||||
"auth.login.failure",
|
||||
lambda e: email in (e.get("summary") or ""),
|
||||
)
|
||||
assert event is not None
|
||||
assert event["result"] == "failure"
|
||||
assert event["actor_kind"] == "guest"
|
||||
|
||||
|
||||
def test_login_logout_recorded(seeded_db):
|
||||
name = _unique("aulog")
|
||||
email = f"{name}@t.dev"
|
||||
s = requests.Session()
|
||||
s.post(
|
||||
f"{BASE_URL}/auth/signup",
|
||||
data={
|
||||
"username": name,
|
||||
"email": email,
|
||||
"password": "secret123",
|
||||
"confirm_password": "secret123",
|
||||
},
|
||||
allow_redirects=True,
|
||||
)
|
||||
# explicit login generates auth.login.success (signup alone does not)
|
||||
s.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
data={"email": email, "password": "secret123"},
|
||||
allow_redirects=True,
|
||||
)
|
||||
s.get(f"{BASE_URL}/auth/logout", allow_redirects=False)
|
||||
admin = _admin(seeded_db)
|
||||
success = _find(
|
||||
admin, "auth.login.success", lambda e: e.get("target_label") == name
|
||||
)
|
||||
logout = _find(admin, "auth.logout", lambda e: e.get("target_label") == name)
|
||||
assert success is not None
|
||||
assert logout is not None
|
||||
|
||||
|
||||
def test_post_create_and_delete_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
create = s.post(
|
||||
f"{BASE_URL}/posts/create",
|
||||
headers=JSON,
|
||||
data={"title": _unique("aupost"), "content": "audit body", "topic": "devlog"},
|
||||
).json()
|
||||
uid = create["data"]["uid"]
|
||||
slug = create["data"]["slug"]
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(admin, "post.create", lambda e: e.get("target_uid") == uid) is not None
|
||||
s.post(f"{BASE_URL}/posts/delete/{slug}", headers=JSON, allow_redirects=False)
|
||||
assert _find(admin, "post.delete", lambda e: e.get("target_uid") == uid) is not None
|
||||
|
||||
|
||||
def test_follow_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
s.post(f"{BASE_URL}/follow/bob_test", headers=JSON, allow_redirects=False)
|
||||
admin = _admin(seeded_db)
|
||||
event = _find(
|
||||
admin, "follow.follow", lambda e: e.get("target_label") == "bob_test"
|
||||
)
|
||||
assert event is not None
|
||||
|
||||
|
||||
def test_vote_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
create = s.post(
|
||||
f"{BASE_URL}/posts/create",
|
||||
headers=JSON,
|
||||
data={"title": _unique("auvote"), "content": "vote target", "topic": "devlog"},
|
||||
).json()
|
||||
uid = create["data"]["uid"]
|
||||
s.post(f"{BASE_URL}/votes/post/{uid}", data={"value": "1"}, allow_redirects=False)
|
||||
admin = _admin(seeded_db)
|
||||
event = _find(admin, "vote.post.up", lambda e: e.get("target_uid") == uid)
|
||||
assert event is not None
|
||||
assert event["new_value"] == "1"
|
||||
|
||||
|
||||
def test_profile_update_recorded(seeded_db):
|
||||
s, name = _member()
|
||||
s.post(
|
||||
f"{BASE_URL}/profile/update",
|
||||
headers=JSON,
|
||||
data={"bio": "audited bio", "location": "", "git_link": "", "website": ""},
|
||||
allow_redirects=False,
|
||||
)
|
||||
admin = _admin(seeded_db)
|
||||
event = _find(admin, "profile.update", lambda e: e.get("target_label") == name)
|
||||
assert event is not None
|
||||
|
||||
|
||||
def test_admin_setting_update_recorded(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
settings = get_table("site_settings")
|
||||
original = (settings.find_one(key="site_tagline") or {}).get("value", "")
|
||||
newval = _unique("autag")
|
||||
# keep the booleans pinned: the settings form inserts every missing key, so
|
||||
# an unset registration_open/maintenance_mode would otherwise default to ""
|
||||
# and silently close registration for the rest of the suite.
|
||||
keep = {"registration_open": "1", "maintenance_mode": "0"}
|
||||
try:
|
||||
admin.post(
|
||||
f"{BASE_URL}/admin/settings",
|
||||
headers=JSON,
|
||||
data={"site_tagline": newval, **keep},
|
||||
allow_redirects=False,
|
||||
)
|
||||
event = _find(
|
||||
admin, "admin.setting.update", lambda e: e.get("new_value") == newval
|
||||
)
|
||||
assert event is not None
|
||||
assert event["target_type"] == "setting"
|
||||
assert event["target_uid"] == "site_tagline"
|
||||
finally:
|
||||
admin.post(
|
||||
f"{BASE_URL}/admin/settings",
|
||||
data={"site_tagline": original, **keep},
|
||||
allow_redirects=False,
|
||||
)
|
||||
|
||||
|
||||
def test_admin_role_change_and_self_denied(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
_, target = _member()
|
||||
target_uid = _db_user(target)["uid"]
|
||||
admin.post(
|
||||
f"{BASE_URL}/admin/users/{target_uid}/role",
|
||||
headers=JSON,
|
||||
data={"role": "admin"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
changed = _find(
|
||||
admin,
|
||||
"admin.user.role.change",
|
||||
lambda e: e.get("target_uid") == target_uid and e.get("result") == "success",
|
||||
)
|
||||
assert changed is not None
|
||||
assert changed["new_value"] == "Admin"
|
||||
# self role change is denied and recorded
|
||||
alice_uid = _db_user("alice_test")["uid"]
|
||||
admin.post(
|
||||
f"{BASE_URL}/admin/users/{alice_uid}/role",
|
||||
headers=JSON,
|
||||
data={"role": "member"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
denied = _find(
|
||||
admin,
|
||||
"admin.user.role.change",
|
||||
lambda e: e.get("target_uid") == alice_uid and e.get("result") == "denied",
|
||||
)
|
||||
assert denied is not None
|
||||
|
||||
|
||||
def test_security_authz_denied_recorded(seeded_db):
|
||||
requests.get(f"{BASE_URL}/messages", allow_redirects=False)
|
||||
admin = _admin(seeded_db)
|
||||
event = _find(
|
||||
admin,
|
||||
"security.authz.denied",
|
||||
lambda e: e.get("request_path") == "/messages",
|
||||
)
|
||||
assert event is not None
|
||||
assert event["result"] == "denied"
|
||||
|
||||
|
||||
def test_devii_origin_header_attribution(seeded_db):
|
||||
s, _ = _member()
|
||||
create = s.post(
|
||||
f"{BASE_URL}/posts/create",
|
||||
headers={**JSON, "X-Devii-Agent": "1"},
|
||||
data={"title": _unique("audevii"), "content": "audited via the agent", "topic": "devlog"},
|
||||
).json()
|
||||
uid = create["data"]["uid"]
|
||||
admin = _admin(seeded_db)
|
||||
event = _find(admin, "post.create", lambda e: e.get("target_uid") == uid)
|
||||
assert event is not None
|
||||
assert event["origin"] == "devii"
|
||||
assert event["via_agent"] == 1
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# filtering and pagination (JSON) #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_filter_by_event_key(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
data = _audit(admin, event_key="auth.signup")
|
||||
assert data["pagination"]["total"] >= 1
|
||||
assert all(e["event_key"] == "auth.signup" for e in data["entries"])
|
||||
assert data["filters"]["event_key"] == "auth.signup"
|
||||
|
||||
|
||||
def test_filter_by_result(seeded_db):
|
||||
# ensure at least one failure exists
|
||||
requests.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
data={"email": f"{_unique('aurf')}@t.dev", "password": "nope"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
admin = _admin(seeded_db)
|
||||
data = _audit(admin, result="failure")
|
||||
assert data["pagination"]["total"] >= 1
|
||||
assert all(e["result"] == "failure" for e in data["entries"])
|
||||
|
||||
|
||||
def test_filter_by_category(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
data = _audit(admin, category="auth")
|
||||
assert data["pagination"]["total"] >= 1
|
||||
assert all(e["category"] == "auth" for e in data["entries"])
|
||||
|
||||
|
||||
def test_free_text_search(seeded_db):
|
||||
s, name = _member()
|
||||
s.post(f"{BASE_URL}/follow/bob_test", headers=JSON, allow_redirects=False)
|
||||
admin = _admin(seeded_db)
|
||||
data = _audit(admin, q=name)
|
||||
assert data["pagination"]["total"] >= 1
|
||||
assert all(name in (e["summary"] or "") for e in data["entries"])
|
||||
|
||||
|
||||
def test_pagination_spans_pages(seeded_db):
|
||||
for _ in range(30):
|
||||
requests.get(f"{BASE_URL}/notifications", allow_redirects=False)
|
||||
admin = _admin(seeded_db)
|
||||
data = _audit(admin, event_key="security.authz.denied")
|
||||
assert data["pagination"]["total"] >= 30
|
||||
assert data["pagination"]["total_pages"] >= 2
|
||||
page2 = _audit(admin, event_key="security.authz.denied", page=2)
|
||||
assert len(page2["entries"]) > 0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# admin UI (Playwright) #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
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_sidebar_link_present(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/admin/users", wait_until="domcontentloaded")
|
||||
link = page.locator(".sidebar-card a.sidebar-link[href='/admin/audit-log']")
|
||||
assert link.count() == 1
|
||||
assert "Audit Log" in link.inner_text()
|
||||
|
||||
|
||||
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_pagination_preserves_filters(alice):
|
||||
page, _ = alice
|
||||
for _ in range(30):
|
||||
requests.get(f"{BASE_URL}/bookmarks/saved", allow_redirects=False)
|
||||
page.goto(
|
||||
f"{BASE_URL}/admin/audit-log?event_key=security.authz.denied",
|
||||
wait_until="domcontentloaded",
|
||||
)
|
||||
next_link = page.locator("a.pagination-btn:has-text('Next')")
|
||||
assert next_link.count() > 0
|
||||
href = next_link.first.get_attribute("href")
|
||||
assert "event_key=security.authz.denied" in href
|
||||
assert "page=2" in href
|
||||
next_link.first.click()
|
||||
page.wait_for_url("**page=2**", wait_until="domcontentloaded")
|
||||
assert "event_key=security.authz.denied" in page.url
|
||||
# the filter is still applied on page 2: every visible event badge matches
|
||||
badges = page.locator(
|
||||
"table.admin-table tbody tr td .audit-badge"
|
||||
).all_inner_texts()
|
||||
event_badges = [b for b in badges if "." in b and b.islower()]
|
||||
assert event_badges
|
||||
assert all(b == "security.authz.denied" for b in event_badges)
|
||||
|
||||
|
||||
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"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# emission - content, engagement, projects, files (every mechanism) #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _new_post(session, body="audited post body here"):
|
||||
return session.post(
|
||||
f"{BASE_URL}/posts/create",
|
||||
headers=JSON,
|
||||
data={"title": _unique("aup"), "content": body, "topic": "devlog"},
|
||||
).json()["data"]
|
||||
|
||||
|
||||
def _new_project(session):
|
||||
return session.post(
|
||||
f"{BASE_URL}/projects/create",
|
||||
headers=JSON,
|
||||
data={
|
||||
"title": _unique("aupr"),
|
||||
"description": "audited project description text",
|
||||
"project_type": "software",
|
||||
"status": "In Development",
|
||||
"platforms": "",
|
||||
},
|
||||
).json()["data"]
|
||||
|
||||
|
||||
def test_post_edit_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
post = _new_post(s)
|
||||
s.post(
|
||||
f"{BASE_URL}/posts/edit/{post['slug']}",
|
||||
headers=JSON,
|
||||
data={"title": "edited audited title", "content": "edited body content", "topic": "devlog"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(admin, "post.edit", lambda e: e.get("target_uid") == post["uid"]) is not None
|
||||
|
||||
|
||||
def test_content_delete_denied_recorded(seeded_db):
|
||||
owner, _ = _member()
|
||||
post = _new_post(owner)
|
||||
other, _ = _member()
|
||||
other.post(
|
||||
f"{BASE_URL}/posts/delete/{post['slug']}", headers=JSON, allow_redirects=False
|
||||
)
|
||||
admin = _admin(seeded_db)
|
||||
event = _find(
|
||||
admin,
|
||||
"post.delete",
|
||||
lambda e: e.get("target_uid") == post["uid"] and e.get("result") == "denied",
|
||||
)
|
||||
assert event is not None
|
||||
|
||||
|
||||
def test_comment_create_and_delete_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
post = _new_post(s)
|
||||
comment = s.post(
|
||||
f"{BASE_URL}/comments/create",
|
||||
headers=JSON,
|
||||
data={"content": "audited comment body", "target_type": "post", "post_uid": post["uid"]},
|
||||
).json()["data"]
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(admin, "comment.create.post", lambda e: e.get("target_uid") == comment["uid"]) is not None
|
||||
s.post(f"{BASE_URL}/comments/delete/{comment['uid']}", headers=JSON, allow_redirects=False)
|
||||
assert _find(admin, "comment.delete", lambda e: e.get("target_uid") == comment["uid"]) is not None
|
||||
|
||||
|
||||
def test_reaction_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
post = _new_post(s)
|
||||
s.post(f"{BASE_URL}/reactions/post/{post['uid']}", data={"emoji": "\U0001F44D"}, allow_redirects=False)
|
||||
admin = _admin(seeded_db)
|
||||
event = _find(admin, "reaction.add", lambda e: e.get("target_uid") == post["uid"])
|
||||
assert event is not None
|
||||
assert (event.get("metadata") or {}).get("emoji") == "\U0001F44D"
|
||||
|
||||
|
||||
def test_bookmark_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
post = _new_post(s)
|
||||
s.post(f"{BASE_URL}/bookmarks/post/{post['uid']}", allow_redirects=False)
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(admin, "bookmark.add", lambda e: e.get("target_uid") == post["uid"]) is not None
|
||||
|
||||
|
||||
def test_poll_create_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
question = _unique("auq") + "?"
|
||||
s.post(
|
||||
f"{BASE_URL}/posts/create",
|
||||
headers=JSON,
|
||||
data={
|
||||
"title": _unique("aupoll"),
|
||||
"content": "post with an audited poll",
|
||||
"topic": "devlog",
|
||||
"poll_question": question,
|
||||
"poll_options": ["Alpha", "Beta"],
|
||||
},
|
||||
)
|
||||
admin = _admin(seeded_db)
|
||||
event = _find(
|
||||
admin, "poll.create", lambda e: (e.get("metadata") or {}).get("question") == question
|
||||
)
|
||||
assert event is not None
|
||||
|
||||
|
||||
def test_gist_create_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
gist = s.post(
|
||||
f"{BASE_URL}/gists/create",
|
||||
headers=JSON,
|
||||
data={"title": _unique("aug"), "description": "audited gist", "source_code": "print(1)", "language": "python"},
|
||||
).json()["data"]
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(admin, "gist.create", lambda e: e.get("target_uid") == gist["uid"]) is not None
|
||||
|
||||
|
||||
def test_bug_create_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
bug = s.post(
|
||||
f"{BASE_URL}/bugs/create",
|
||||
headers=JSON,
|
||||
data={"title": _unique("aubug"), "description": "audited bug description text"},
|
||||
).json()["data"]
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(admin, "bug.create", lambda e: e.get("target_uid") == bug["uid"]) is not None
|
||||
|
||||
|
||||
def test_message_send_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
receiver = _db_user("bob_test")["uid"]
|
||||
msg = s.post(
|
||||
f"{BASE_URL}/messages/send",
|
||||
headers=JSON,
|
||||
data={"content": "audited direct message", "receiver_uid": receiver},
|
||||
).json()["data"]
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(admin, "message.send", lambda e: e.get("target_uid") == msg["uid"]) is not None
|
||||
|
||||
|
||||
def test_attachment_upload_and_delete_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
up = s.post(
|
||||
f"{BASE_URL}/uploads/upload",
|
||||
files={"file": ("audit.txt", b"audited file contents here")},
|
||||
).json()
|
||||
auid = up.get("uid")
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(admin, "attachment.upload", lambda e: e.get("target_uid") == auid) is not None
|
||||
s.delete(f"{BASE_URL}/uploads/delete/{auid}")
|
||||
assert _find(admin, "attachment.delete", lambda e: e.get("target_uid") == auid) is not None
|
||||
|
||||
|
||||
def test_project_create_visibility_readonly_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
proj = _new_project(s)
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(admin, "project.create", lambda e: e.get("target_uid") == proj["uid"]) is not None
|
||||
s.post(f"{BASE_URL}/projects/{proj['slug']}/private", headers=JSON, data={"value": "true"}, allow_redirects=False)
|
||||
assert _find(admin, "project.visibility.private", lambda e: e.get("target_uid") == proj["uid"]) is not None
|
||||
s.post(f"{BASE_URL}/projects/{proj['slug']}/readonly", headers=JSON, data={"value": "true"}, allow_redirects=False)
|
||||
assert _find(admin, "project.readonly.enable", lambda e: e.get("target_uid") == proj["uid"]) is not None
|
||||
|
||||
|
||||
def test_project_fork_and_zip_request_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
proj = _new_project(s)
|
||||
s.post(f"{BASE_URL}/projects/{proj['slug']}/fork", headers=JSON, data={"title": _unique("aufork")}, allow_redirects=False)
|
||||
s.post(f"{BASE_URL}/projects/{proj['slug']}/zip", headers=JSON, allow_redirects=False)
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(admin, "project.fork.request", lambda e: e.get("target_uid") == proj["uid"]) is not None
|
||||
assert _find(admin, "project.zip.request", lambda e: e.get("target_uid") == proj["uid"]) is not None
|
||||
|
||||
|
||||
def test_project_file_write_and_readonly_denied_recorded(seeded_db):
|
||||
s, _ = _member()
|
||||
proj = _new_project(s)
|
||||
s.post(
|
||||
f"{BASE_URL}/projects/{proj['slug']}/files/write",
|
||||
headers=JSON,
|
||||
data={"path": "audit_a.txt", "content": "hello world"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(
|
||||
admin, "file.write.create", lambda e: e.get("target_uid") == f"{proj['uid']}:audit_a.txt"
|
||||
) is not None
|
||||
# turn the project read-only, then a write must be recorded as denied
|
||||
s.post(f"{BASE_URL}/projects/{proj['slug']}/readonly", headers=JSON, data={"value": "true"}, allow_redirects=False)
|
||||
s.post(
|
||||
f"{BASE_URL}/projects/{proj['slug']}/files/write",
|
||||
headers=JSON,
|
||||
data={"path": "audit_b.txt", "content": "blocked"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
denied = _find(
|
||||
admin,
|
||||
"file.write.create",
|
||||
lambda e: e.get("target_uid") == f"{proj['uid']}:audit_b.txt" and e.get("result") == "denied",
|
||||
)
|
||||
assert denied is not None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# emission - admin governance and services #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_admin_password_reset_recorded(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
_, target = _member()
|
||||
uid = _db_user(target)["uid"]
|
||||
admin.post(f"{BASE_URL}/admin/users/{uid}/password", headers=JSON, data={"password": "newsecret123"}, allow_redirects=False)
|
||||
event = _find(admin, "admin.user.password.reset", lambda e: e.get("target_uid") == uid)
|
||||
assert event is not None
|
||||
# the password value is never recorded
|
||||
assert "newsecret123" not in (event.get("summary") or "")
|
||||
import json as _json
|
||||
|
||||
assert "newsecret123" not in _json.dumps(event.get("metadata") or {})
|
||||
|
||||
|
||||
def test_admin_user_toggle_and_self_disable_denied(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
_, target = _member()
|
||||
uid = _db_user(target)["uid"]
|
||||
admin.post(f"{BASE_URL}/admin/users/{uid}/toggle", headers=JSON, allow_redirects=False)
|
||||
assert _find(admin, "admin.user.active.disable", lambda e: e.get("target_uid") == uid) is not None
|
||||
alice_uid = _db_user("alice_test")["uid"]
|
||||
admin.post(f"{BASE_URL}/admin/users/{alice_uid}/toggle", headers=JSON, allow_redirects=False)
|
||||
denied = _find(
|
||||
admin,
|
||||
"admin.user.active.disable",
|
||||
lambda e: e.get("target_uid") == alice_uid and e.get("result") == "denied",
|
||||
)
|
||||
assert denied is not None
|
||||
|
||||
|
||||
def test_admin_ai_quota_resets_recorded(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
_, target = _member()
|
||||
uid = _db_user(target)["uid"]
|
||||
admin.post(f"{BASE_URL}/admin/users/{uid}/reset-ai-quota", headers=JSON, allow_redirects=False)
|
||||
admin.post(f"{BASE_URL}/admin/ai-quota/reset-guests", headers=JSON, allow_redirects=False)
|
||||
admin.post(f"{BASE_URL}/admin/ai-quota/reset-all", headers=JSON, allow_redirects=False)
|
||||
assert _find(admin, "admin.user.ai_quota.reset", lambda e: e.get("target_uid") == uid) is not None
|
||||
assert _find(admin, "admin.ai_quota.reset_guests", lambda e: True) is not None
|
||||
assert _find(admin, "admin.ai_quota.reset_all", lambda e: True) is not None
|
||||
|
||||
|
||||
def _seed_news():
|
||||
uid = generate_uid()
|
||||
title = _unique("aunews")
|
||||
get_table("news").insert(
|
||||
{
|
||||
"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_news_actions_recorded(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
uid = _seed_news()
|
||||
admin.post(f"{BASE_URL}/admin/news/{uid}/toggle", headers=JSON, allow_redirects=False)
|
||||
admin.post(f"{BASE_URL}/admin/news/{uid}/publish", headers=JSON, allow_redirects=False)
|
||||
admin.post(f"{BASE_URL}/admin/news/{uid}/landing", headers=JSON, allow_redirects=False)
|
||||
admin.post(f"{BASE_URL}/admin/news/{uid}/delete", headers=JSON, allow_redirects=False)
|
||||
for key in ("news.featured.toggle", "news.publish.toggle", "news.landing.toggle", "news.delete"):
|
||||
assert _find(admin, key, lambda e: e.get("target_uid") == uid) is not None, key
|
||||
|
||||
|
||||
def test_service_lifecycle_recorded(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
try:
|
||||
admin.post(f"{BASE_URL}/admin/services/audit/stop", headers=JSON, allow_redirects=False)
|
||||
admin.post(f"{BASE_URL}/admin/services/audit/start", headers=JSON, allow_redirects=False)
|
||||
admin.post(f"{BASE_URL}/admin/services/audit/run", headers=JSON, allow_redirects=False)
|
||||
assert _find(admin, "service.stop", lambda e: e.get("target_uid") == "audit") is not None
|
||||
assert _find(admin, "service.start", lambda e: e.get("target_uid") == "audit") is not None
|
||||
assert _find(admin, "service.run_now", lambda e: e.get("target_uid") == "audit") is not None
|
||||
finally:
|
||||
admin.post(f"{BASE_URL}/admin/services/audit/start", allow_redirects=False)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# emission - security middleware and request-less side-effects #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_maintenance_block_recorded(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
set_setting("maintenance_mode", "1")
|
||||
try:
|
||||
requests.get(f"{BASE_URL}/feed", allow_redirects=False)
|
||||
finally:
|
||||
set_setting("maintenance_mode", "0")
|
||||
event = _find(admin, "security.maintenance.block", lambda e: True)
|
||||
assert event is not None
|
||||
assert event["result"] == "denied"
|
||||
|
||||
|
||||
def test_rate_limit_block_recorded(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
set_setting("rate_limit_per_minute", "1")
|
||||
try:
|
||||
for _ in range(4):
|
||||
requests.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
data={"email": "x@y.dev", "password": "nope"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
finally:
|
||||
set_setting("rate_limit_per_minute", "1000000")
|
||||
event = _find(admin, "security.rate_limit.block", lambda e: True)
|
||||
assert event is not None
|
||||
assert event["result"] == "denied"
|
||||
|
||||
|
||||
def test_post_create_emits_reward_side_effects(seeded_db):
|
||||
s, name = _member()
|
||||
uid = _db_user(name)["uid"]
|
||||
_new_post(s)
|
||||
admin = _admin(seeded_db)
|
||||
assert _find(admin, "reward.xp.grant", lambda e: e.get("actor_uid") == uid) is not None
|
||||
assert _find(admin, "reward.badge.award", lambda e: e.get("actor_uid") == uid) is not None
|
||||
|
||||
|
||||
def test_follow_emits_notification_create(seeded_db):
|
||||
s, name = _member()
|
||||
s.post(f"{BASE_URL}/follow/bob_test", headers=JSON, allow_redirects=False)
|
||||
admin = _admin(seeded_db)
|
||||
event = _find(
|
||||
admin, "notification.create", lambda e: name in (e.get("summary") or "")
|
||||
)
|
||||
assert event is not None
|
||||
assert event["actor_kind"] == "system"
|
||||
|
||||
|
||||
def test_audit_docs_page_admin_only(seeded_db):
|
||||
# guest cannot see the admin-only audit-log docs page
|
||||
assert requests.get(f"{BASE_URL}/docs/audit-log.html").status_code == 404
|
||||
admin = _db_user("alice_test")
|
||||
r = requests.get(
|
||||
f"{BASE_URL}/docs/audit-log.html", headers={"X-API-KEY": admin["api_key"]}
|
||||
)
|
||||
assert r.status_code == 200
|
||||
assert "Audit Log" in r.text
|
||||
@@ -0,0 +1,393 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from devplacepy.database import get_table
|
||||
from devplacepy.services.audit import store, query
|
||||
from devplacepy.services.audit import record as audit
|
||||
from devplacepy.services.audit.categories import category_for
|
||||
from devplacepy.services.audit.record import _sanitize_summary, _actor_from, _coerce_scalar
|
||||
|
||||
|
||||
def _now():
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _tables():
|
||||
store.ensure_tables()
|
||||
yield
|
||||
|
||||
|
||||
def test_category_for_known_prefixes():
|
||||
assert category_for("auth.login.success") == "auth"
|
||||
assert category_for("vote.post.up") == "engagement"
|
||||
assert category_for("reaction.add") == "engagement"
|
||||
assert category_for("file.write.create") == "project_files"
|
||||
assert category_for("dir.create") == "project_files"
|
||||
assert category_for("job.zip.complete") == "project_files"
|
||||
assert category_for("admin.setting.update") == "admin"
|
||||
assert category_for("container.instance.start") == "container"
|
||||
assert category_for("devii.turn") == "devii"
|
||||
assert category_for("cli.role.set") == "cli"
|
||||
assert category_for("reward.xp.grant") == "reward"
|
||||
assert category_for("security.authz.denied") == "security"
|
||||
assert category_for("proxy.access") == "ingress"
|
||||
assert category_for("ai.gateway.call") == "ai"
|
||||
|
||||
|
||||
def test_category_for_unknown_falls_back_to_first_segment():
|
||||
assert category_for("totallyunknown.event") == "totallyunknown"
|
||||
assert category_for("") == "other"
|
||||
|
||||
|
||||
def test_sanitize_summary_strips_html_and_truncates():
|
||||
assert _sanitize_summary("<b>hi</b> <script>x()</script> there") == "hi x() there"
|
||||
assert _sanitize_summary(None) is None
|
||||
assert _sanitize_summary("") is None
|
||||
long = "a" * 500
|
||||
out = _sanitize_summary(long)
|
||||
assert len(out) <= 140
|
||||
assert out.endswith("...")
|
||||
|
||||
|
||||
def test_actor_from_resolves_role():
|
||||
assert _actor_from(None)["actor_kind"] == "guest"
|
||||
assert _actor_from(None)["actor_role"] == "guest"
|
||||
member = _actor_from({"uid": "u1", "username": "m", "role": "Member"})
|
||||
assert member["actor_kind"] == "user" and member["actor_role"] == "member"
|
||||
admin = _actor_from({"uid": "u2", "username": "a", "role": "Admin"})
|
||||
assert admin["actor_role"] == "admin"
|
||||
|
||||
|
||||
def test_coerce_scalar():
|
||||
assert _coerce_scalar(None) is None
|
||||
assert _coerce_scalar(True) == "1"
|
||||
assert _coerce_scalar(False) == "0"
|
||||
assert _coerce_scalar(5) == "5"
|
||||
assert _coerce_scalar("x") == "x"
|
||||
|
||||
|
||||
def test_record_system_inserts_row_with_fields():
|
||||
key = f"test.unit.basic.{audit_uid_marker()}"
|
||||
uid = audit.record_system(
|
||||
key,
|
||||
actor_kind="user",
|
||||
actor_uid="actor-1",
|
||||
actor_username="alice",
|
||||
actor_role="admin",
|
||||
target_type="post",
|
||||
target_uid="post-1",
|
||||
target_label="Hello",
|
||||
old_value=0,
|
||||
new_value=1,
|
||||
summary="<i>did</i> a thing",
|
||||
metadata={"k": "v", "n": 3},
|
||||
result="success",
|
||||
)
|
||||
assert uid
|
||||
row = store.get_event(uid)
|
||||
assert row["event_key"] == key
|
||||
assert row["category"] == "test"
|
||||
assert row["actor_kind"] == "user"
|
||||
assert row["actor_uid"] == "actor-1"
|
||||
assert row["actor_role"] == "admin"
|
||||
assert row["target_type"] == "post"
|
||||
assert row["old_value"] == "0" and row["new_value"] == "1"
|
||||
assert row["summary"] == "did a thing"
|
||||
assert row["result"] == "success"
|
||||
import json
|
||||
|
||||
assert json.loads(row["metadata"]) == {"k": "v", "n": 3}
|
||||
|
||||
|
||||
def test_record_system_auto_appends_actor_link():
|
||||
uid = audit.record_system(
|
||||
f"test.unit.actorlink.{audit_uid_marker()}",
|
||||
actor_kind="user",
|
||||
actor_uid="actor-2",
|
||||
actor_username="bob",
|
||||
)
|
||||
links = store.get_links(uid)
|
||||
actor_links = [link for link in links if link["relation"] == "actor"]
|
||||
assert len(actor_links) == 1
|
||||
assert actor_links[0]["object_uid"] == "actor-2"
|
||||
|
||||
|
||||
def test_record_system_explicit_links_preserved():
|
||||
uid = audit.record_system(
|
||||
f"test.unit.links.{audit_uid_marker()}",
|
||||
actor_kind="user",
|
||||
actor_uid="actor-3",
|
||||
links=[audit.target("project", "proj-1", "P"), audit.author("author-1")],
|
||||
)
|
||||
links = store.get_links(uid)
|
||||
relations = sorted(link["relation"] for link in links)
|
||||
assert relations == ["actor", "author", "target"]
|
||||
|
||||
|
||||
def test_record_never_raises(monkeypatch):
|
||||
def boom(_row):
|
||||
raise RuntimeError("db exploded")
|
||||
|
||||
monkeypatch.setattr(store, "insert_event", boom)
|
||||
# must swallow the error and return None rather than propagate
|
||||
assert audit.record_system("test.unit.raise", actor_kind="system") is None
|
||||
assert audit.record(None, "test.unit.raise2", user=None) is None
|
||||
|
||||
|
||||
def test_record_with_none_request_uses_explicit_user():
|
||||
uid = audit.record(
|
||||
None,
|
||||
f"test.unit.norequest.{audit_uid_marker()}",
|
||||
user={"uid": "u9", "username": "carol", "role": "Member"},
|
||||
summary="no request object",
|
||||
)
|
||||
row = store.get_event(uid)
|
||||
assert row["actor_uid"] == "u9"
|
||||
assert row["actor_role"] == "member"
|
||||
assert row["origin"] == "web"
|
||||
assert row["request_method"] is None
|
||||
|
||||
|
||||
def test_query_list_events_filters():
|
||||
marker = f"qfilter{audit_uid_marker()}"
|
||||
key = f"test.qf.{marker}"
|
||||
for i in range(3):
|
||||
audit.record_system(
|
||||
key,
|
||||
actor_kind="user",
|
||||
actor_uid=f"u{i}",
|
||||
actor_role="member",
|
||||
result="success" if i < 2 else "failure",
|
||||
summary=f"event {marker} number {i}",
|
||||
)
|
||||
rows, pagination = query.list_events({"event_key": key}, 1)
|
||||
assert pagination["total"] == 3
|
||||
rows_fail, pag_fail = query.list_events({"event_key": key, "result": "failure"}, 1)
|
||||
assert pag_fail["total"] == 1
|
||||
rows_q, pag_q = query.list_events({"q": f"{marker} number 1"}, 1)
|
||||
assert pag_q["total"] >= 1
|
||||
assert all(marker in (r["summary"] or "") for r in rows_q)
|
||||
|
||||
|
||||
def test_query_list_events_pagination():
|
||||
key = f"test.qpage.{audit_uid_marker()}"
|
||||
for i in range(30):
|
||||
audit.record_system(key, actor_kind="system", summary=f"page row {i}")
|
||||
rows1, pag1 = query.list_events({"event_key": key}, 1)
|
||||
assert pag1["total"] == 30
|
||||
assert pag1["total_pages"] == 2
|
||||
assert len(rows1) == 25
|
||||
rows2, pag2 = query.list_events({"event_key": key}, 2)
|
||||
assert len(rows2) == 5
|
||||
# newest first ordering: page 1 first row is newer than page 2 last row
|
||||
assert rows1[0]["created_at"] >= rows2[-1]["created_at"]
|
||||
|
||||
|
||||
def test_query_date_range():
|
||||
key = f"test.qdate.{audit_uid_marker()}"
|
||||
audit.record_system(key, actor_kind="system", summary="dated")
|
||||
rows, pag = query.list_events(
|
||||
{"event_key": key, "date_from": "1999-01-01", "date_to": "2099-12-31"}, 1
|
||||
)
|
||||
assert pag["total"] == 1
|
||||
rows_none, pag_none = query.list_events(
|
||||
{"event_key": key, "date_to": "2000-01-01"}, 1
|
||||
)
|
||||
assert pag_none["total"] == 0
|
||||
|
||||
|
||||
def test_filter_options_returns_distinct():
|
||||
key = f"test.opts.{audit_uid_marker()}"
|
||||
audit.record_system(key, actor_kind="cli", actor_role="system", origin="cli")
|
||||
query._options_cache.clear()
|
||||
options = query.filter_options()
|
||||
assert key in options["event_key"]
|
||||
assert "cli" in options["origin"]
|
||||
assert "category" in options and "result" in options
|
||||
|
||||
|
||||
def test_get_event_with_links():
|
||||
uid = audit.record_system(
|
||||
f"test.unit.detail.{audit_uid_marker()}",
|
||||
actor_kind="user",
|
||||
actor_uid="u-detail",
|
||||
links=[audit.target("gist", "g1", "G")],
|
||||
)
|
||||
data = query.get_event_with_links(uid)
|
||||
assert data is not None
|
||||
assert data["event"]["uid"] == uid
|
||||
relations = {link["relation"] for link in data["links"]}
|
||||
assert "target" in relations and "actor" in relations
|
||||
assert query.get_event_with_links("nonexistent-uid") is None
|
||||
|
||||
|
||||
def test_sweep_removes_old_rows_and_links():
|
||||
ancient = "1990-01-01T00:00:00+00:00"
|
||||
recent_key = f"test.sweep.recent.{audit_uid_marker()}"
|
||||
old_uids = []
|
||||
for i in range(3):
|
||||
uid = store.insert_event(
|
||||
{
|
||||
"event_key": f"test.sweep.old.{i}",
|
||||
"category": "test",
|
||||
"actor_kind": "system",
|
||||
"created_at": ancient,
|
||||
"result": "success",
|
||||
}
|
||||
)
|
||||
store.insert_links(uid, [audit.target("post", f"p{i}")])
|
||||
old_uids.append(uid)
|
||||
recent_uid = audit.record_system(recent_key, actor_kind="system")
|
||||
removed_links, removed_events = store.sweep("2000-01-01T00:00:00+00:00")
|
||||
assert removed_events >= 3
|
||||
assert removed_links >= 3
|
||||
for uid in old_uids:
|
||||
assert store.get_event(uid) is None
|
||||
assert store.get_links(uid) == []
|
||||
# the recent row survives
|
||||
assert store.get_event(recent_uid) is not None
|
||||
|
||||
|
||||
def test_sweep_zero_cutoff_is_safe():
|
||||
# a cutoff far in the past removes nothing recent
|
||||
recent_uid = audit.record_system(
|
||||
f"test.sweep.keep.{audit_uid_marker()}", actor_kind="system"
|
||||
)
|
||||
store.sweep("1970-01-01T00:00:00+00:00")
|
||||
assert store.get_event(recent_uid) is not None
|
||||
|
||||
|
||||
def test_retention_service_prunes_old_rows():
|
||||
from tests.conftest import run_async
|
||||
from devplacepy.database import set_setting
|
||||
from devplacepy.services.audit import AuditService
|
||||
|
||||
ancient = "1991-01-01T00:00:00+00:00"
|
||||
old_uid = store.insert_event(
|
||||
{
|
||||
"event_key": f"test.retention.old.{audit_uid_marker()}",
|
||||
"category": "test",
|
||||
"actor_kind": "system",
|
||||
"created_at": ancient,
|
||||
"result": "success",
|
||||
}
|
||||
)
|
||||
recent_uid = audit.record_system(
|
||||
f"test.retention.recent.{audit_uid_marker()}", actor_kind="system"
|
||||
)
|
||||
set_setting("audit_log_retention_days", "30")
|
||||
run_async(AuditService().run_once())
|
||||
assert store.get_event(old_uid) is None
|
||||
assert store.get_event(recent_uid) is not None
|
||||
|
||||
|
||||
def test_retention_service_disabled_when_zero():
|
||||
from tests.conftest import run_async
|
||||
from devplacepy.database import set_setting
|
||||
from devplacepy.services.audit import AuditService
|
||||
|
||||
ancient_uid = store.insert_event(
|
||||
{
|
||||
"event_key": f"test.retention.keep.{audit_uid_marker()}",
|
||||
"category": "test",
|
||||
"actor_kind": "system",
|
||||
"created_at": "1992-01-01T00:00:00+00:00",
|
||||
"result": "success",
|
||||
}
|
||||
)
|
||||
set_setting("audit_log_retention_days", "0")
|
||||
try:
|
||||
run_async(AuditService().run_once())
|
||||
assert store.get_event(ancient_uid) is not None
|
||||
finally:
|
||||
set_setting("audit_log_retention_days", "90")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# CLI emission (in-process, via the cli command functions) #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _cli_user(role="Member"):
|
||||
from devplacepy.utils import generate_uid
|
||||
|
||||
username = f"auditcli_{generate_uid()[:8]}"
|
||||
get_table("users").insert(
|
||||
{
|
||||
"uid": generate_uid(),
|
||||
"username": username,
|
||||
"email": f"{username}@t.dev",
|
||||
"api_key": generate_uid(),
|
||||
"role": role,
|
||||
"is_active": True,
|
||||
"created_at": _now(),
|
||||
}
|
||||
)
|
||||
return username
|
||||
|
||||
|
||||
def _latest(event_key, **match):
|
||||
rows = sorted(
|
||||
get_table(store.AUDIT_TABLE).find(event_key=event_key),
|
||||
key=lambda r: r.get("created_at", ""),
|
||||
reverse=True,
|
||||
)
|
||||
for row in rows:
|
||||
if all(row.get(k) == v for k, v in match.items()):
|
||||
return row
|
||||
return None
|
||||
|
||||
|
||||
def test_cli_role_set_recorded(local_db):
|
||||
import argparse
|
||||
from devplacepy import cli
|
||||
|
||||
username = _cli_user(role="Member")
|
||||
uid = get_table("users").find_one(username=username)["uid"]
|
||||
cli.cmd_role_set(argparse.Namespace(username=username, role="admin"))
|
||||
event = _latest("cli.role.set", target_uid=uid)
|
||||
assert event is not None
|
||||
assert event["actor_kind"] == "cli"
|
||||
assert event["origin"] == "cli"
|
||||
import json
|
||||
|
||||
meta = json.loads(event["metadata"])
|
||||
assert meta["old"] == "Member" and meta["new"] == "Admin"
|
||||
links = store.get_links(event["uid"])
|
||||
assert any(link["object_uid"] == uid for link in links)
|
||||
|
||||
|
||||
def test_cli_apikey_reset_recorded(local_db):
|
||||
import argparse
|
||||
from devplacepy import cli
|
||||
|
||||
username = _cli_user()
|
||||
uid = get_table("users").find_one(username=username)["uid"]
|
||||
cli.cmd_apikey_reset(argparse.Namespace(username=username))
|
||||
event = _latest("cli.apikey.reset", target_uid=uid)
|
||||
assert event is not None
|
||||
# the new key value is never written into the audit row
|
||||
new_key = get_table("users").find_one(uid=uid)["api_key"]
|
||||
assert new_key not in (event.get("summary") or "")
|
||||
assert new_key not in (event.get("metadata") or "")
|
||||
|
||||
|
||||
def test_cli_apikey_backfill_recorded(local_db):
|
||||
import argparse
|
||||
from devplacepy import cli
|
||||
|
||||
cli.cmd_apikey_backfill(argparse.Namespace())
|
||||
event = _latest("cli.apikey.backfill")
|
||||
assert event is not None
|
||||
assert event["actor_kind"] == "cli"
|
||||
assert (event.get("metadata") or "").find("count") != -1
|
||||
|
||||
|
||||
_counter = [0]
|
||||
|
||||
|
||||
def audit_uid_marker():
|
||||
_counter[0] += 1
|
||||
return f"{int(datetime.now(timezone.utc).timestamp() * 1000)}_{_counter[0]}"
|
||||
@@ -0,0 +1,278 @@
|
||||
import re
|
||||
from tests.conftest import BASE_URL
|
||||
|
||||
|
||||
def create_post(page, content, topic="random", title=None):
|
||||
page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
|
||||
page.locator(".feed-fab").first.wait_for(state="visible", timeout=10000)
|
||||
page.locator(".feed-fab").first.click()
|
||||
page.check(f"input[value='{topic}']")
|
||||
page.fill("#post-content", content)
|
||||
if title:
|
||||
page.fill("#post-title", title)
|
||||
page.locator("#create-post-modal button.btn-primary:has-text('Post')").click()
|
||||
page.wait_for_url("**/posts/*", timeout=10000, wait_until="domcontentloaded")
|
||||
|
||||
|
||||
def signup(page, username, email, password):
|
||||
page.goto(f"{BASE_URL}/auth/signup", wait_until="domcontentloaded")
|
||||
page.fill("#username", username)
|
||||
page.fill("#email", email)
|
||||
page.fill("#password", password)
|
||||
page.fill("#confirm_password", password)
|
||||
page.click("button:has-text('Create account')")
|
||||
page.wait_for_url("**/feed", timeout=10000, wait_until="domcontentloaded")
|
||||
|
||||
|
||||
def test_full_user_journey(browser, app_server):
|
||||
ctx_a = browser.new_context(viewport={"width": 1400, "height": 900})
|
||||
ctx_b = browser.new_context(viewport={"width": 1400, "height": 900})
|
||||
pa = ctx_a.new_page()
|
||||
pb = ctx_b.new_page()
|
||||
pa.set_default_timeout(10000)
|
||||
pb.set_default_timeout(10000)
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 1 — LANDING PAGE
|
||||
# ═══════════════════════════════════════════════
|
||||
pa.goto(f"{BASE_URL}/", wait_until="domcontentloaded")
|
||||
assert pa.is_visible("text=Devplace.net")
|
||||
assert pa.is_visible("text=The Developer Social Network")
|
||||
assert pa.is_visible("text=Join DevPlace Free")
|
||||
for feat in ("100% Free Forever", "Zero Ads", "No Censorship", "No Payments"):
|
||||
assert pa.is_visible(f"text={feat}")
|
||||
assert pa.is_visible("text=Daily Topic")
|
||||
assert pa.is_visible("text=Read More")
|
||||
|
||||
# Navigate to signup via link
|
||||
pa.click("a:has-text('Sign Up')", no_wait_after=True)
|
||||
pa.wait_for_url("**/auth/signup", timeout=10000, wait_until="domcontentloaded")
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 2 — TWO USERS SIGN UP
|
||||
# ═══════════════════════════════════════════════
|
||||
signup(pa, "alice_demo", "alice_demo@test.dev", "alice_pass_1")
|
||||
signup(pb, "bob_demo", "bob_demo@test.dev", "bob_pass_1")
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 3 — ALICE EXPLORES FEED
|
||||
# ═══════════════════════════════════════════════
|
||||
pa.bring_to_front()
|
||||
pa.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
|
||||
for tab_text in ("Topics", "All", "Trending", "Recent", "Following",
|
||||
"Devlog", "Showcase", "Question", "Rant", "Fun"):
|
||||
assert pa.is_visible(f"text={tab_text}")
|
||||
assert pa.is_visible("text=Total Members")
|
||||
assert pa.is_visible("text=Top Authors")
|
||||
|
||||
# Switch all feed tabs
|
||||
for tab in ("Trending", "Recent"):
|
||||
pa.click(f"a:has-text('{tab}')")
|
||||
pa.wait_for_url(f"**/feed?tab={tab.lower()}", wait_until="domcontentloaded")
|
||||
pa.click("a.feed-nav-btn:has-text('All')")
|
||||
pa.wait_for_url("**/feed**", timeout=10000, wait_until="domcontentloaded")
|
||||
|
||||
# Switch all topic filters
|
||||
for topic in ("showcase", "question", "rant", "fun"):
|
||||
pa.click(f"a:has-text('{topic.capitalize()}')")
|
||||
pa.wait_for_url(f"**/feed?topic={topic}", wait_until="domcontentloaded")
|
||||
pa.click("a:has-text('All')")
|
||||
pa.wait_for_url("**/feed**", timeout=10000, wait_until="domcontentloaded")
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 4 — ALICE CREATES 4 POSTS (ALL TOPICS)
|
||||
# ═══════════════════════════════════════════════
|
||||
create_post(pa, "Just shipped a new feature for my REST API! Took me 3 days.", "devlog", "API Progress Update")
|
||||
post1_url = pa.url
|
||||
|
||||
create_post(pa, "Check out this pixel art editor I built in WASM. Runs at 60fps!", "showcase", "Pixel Editor WASM")
|
||||
|
||||
create_post(pa, "Does anyone else think tabs are better than spaces? Fight me.", "rant", "Tabs vs Spaces")
|
||||
|
||||
create_post(pa, "Why do we call it tech debt when really we just cut corners?", "fun")
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 5 — ALICE CREATES TWO PROJECTS
|
||||
# ═══════════════════════════════════════════════
|
||||
pa.bring_to_front()
|
||||
pa.goto(f"{BASE_URL}/projects", wait_until="domcontentloaded")
|
||||
pa.locator("#create-project-btn").click()
|
||||
pa.fill("#title", "N8Nme")
|
||||
pa.fill("#description", "Visual workflow builder with 300+ integrations.")
|
||||
pa.fill("#release_date", "2026-04-15")
|
||||
pa.check("input[value='software']")
|
||||
for plat in ("Linux", "Docker", "Web"):
|
||||
pa.locator("#platforms-input").fill(plat)
|
||||
pa.locator("#platforms-input").press("Enter")
|
||||
pa.click("button:has-text('Create Project')")
|
||||
pa.wait_for_url(f"{BASE_URL}/projects/*", wait_until="domcontentloaded")
|
||||
assert pa.is_visible("text=N8Nme")
|
||||
|
||||
# Second project — released
|
||||
pa.goto(f"{BASE_URL}/projects", wait_until="domcontentloaded")
|
||||
pa.locator("#create-project-btn").click()
|
||||
pa.fill("#title", "DWN")
|
||||
pa.fill("#description", "Decentralized web node in Rust.")
|
||||
pa.check("input[value='software']")
|
||||
pa.check("input[value='Released']")
|
||||
pa.locator("#platforms-input").fill("Linux")
|
||||
pa.locator("#platforms-input").press("Enter")
|
||||
pa.click("button:has-text('Create Project')")
|
||||
pa.wait_for_url(f"{BASE_URL}/projects/*", wait_until="domcontentloaded")
|
||||
assert pa.is_visible("text=DWN")
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 6 — ALICE EDITS HER PROFILE
|
||||
# ═══════════════════════════════════════════════
|
||||
pa.bring_to_front()
|
||||
pa.goto(f"{BASE_URL}/profile/alice_demo", wait_until="domcontentloaded")
|
||||
assert pa.is_visible("text=alice_demo")
|
||||
assert pa.is_visible("text=Member")
|
||||
assert pa.is_visible("text=Posts")
|
||||
assert pa.is_visible("text=Level")
|
||||
assert pa.is_visible("text=Stars")
|
||||
assert pa.is_visible("text=Progress to next level")
|
||||
assert pa.is_visible("text=Bio")
|
||||
|
||||
pa.evaluate("""
|
||||
document.querySelector('textarea[name="bio"]').value = 'Full-stack developer and open source enthusiast.';
|
||||
document.querySelector('input[name="location"]').value = 'Berlin, Germany';
|
||||
document.querySelector('input[name="git_link"]').value = 'https://github.com/alice_demo';
|
||||
document.querySelector('input[name="website"]').value = 'https://alicedemo.dev';
|
||||
document.querySelector('form[action="/profile/update"]').submit();
|
||||
""")
|
||||
pa.wait_for_timeout(500)
|
||||
assert pa.is_visible("text=Full-stack developer")
|
||||
assert pa.is_visible("text=Berlin, Germany")
|
||||
assert pa.is_visible("text=https://github.com/alice_demo")
|
||||
|
||||
# Switch profile tabs
|
||||
pa.goto(f"{BASE_URL}/profile/alice_demo?tab=projects", wait_until="domcontentloaded")
|
||||
pa.wait_for_timeout(300)
|
||||
assert pa.is_visible("text=N8Nme")
|
||||
assert pa.is_visible("text=DWN")
|
||||
pa.goto(f"{BASE_URL}/profile/alice_demo?tab=posts", wait_until="domcontentloaded")
|
||||
pa.wait_for_timeout(300)
|
||||
assert pa.is_visible("text=API Progress Update")
|
||||
assert pa.is_visible("text=Pixel Editor WASM")
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 7 — BOB EXPLORES ALICE'S STUFF
|
||||
# ═══════════════════════════════════════════════
|
||||
pb.bring_to_front()
|
||||
pb.goto(f"{BASE_URL}/profile/alice_demo", wait_until="domcontentloaded")
|
||||
assert pb.is_visible("text=alice_demo")
|
||||
assert pb.is_visible("text=Full-stack developer")
|
||||
assert pb.is_visible("text=Berlin, Germany")
|
||||
|
||||
pb.goto(f"{BASE_URL}/profile/alice_demo?tab=projects", wait_until="domcontentloaded")
|
||||
assert pb.is_visible("text=N8Nme")
|
||||
assert pb.is_visible("text=DWN")
|
||||
|
||||
# Browse projects, search
|
||||
pb.goto(f"{BASE_URL}/projects", wait_until="domcontentloaded")
|
||||
assert pb.is_visible("text=Recently Released")
|
||||
assert pb.is_visible("text=N8Nme")
|
||||
pb.fill("input[placeholder='Search projects...']", "N8Nme")
|
||||
pb.locator("input[placeholder='Search projects...']").press("Enter")
|
||||
pb.wait_for_timeout(300)
|
||||
assert pb.is_visible("text=N8Nme")
|
||||
|
||||
# Bob creates his own project
|
||||
pb.locator("#create-project-btn").click()
|
||||
pb.fill("#title", "ClaudeCode")
|
||||
pb.fill("#description", "AI-powered code generation for the terminal.")
|
||||
pb.check("input[value='software']")
|
||||
pb.check("input[value='Released']")
|
||||
pb.locator("#platforms-input").fill("Linux")
|
||||
pb.locator("#platforms-input").press("Enter")
|
||||
pb.click("button:has-text('Create Project')")
|
||||
pb.wait_for_timeout(500)
|
||||
assert pb.is_visible("text=ClaudeCode")
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 8 — BOB VOTES AND COMMENTS
|
||||
# ═══════════════════════════════════════════════
|
||||
pb.bring_to_front()
|
||||
pb.goto(post1_url)
|
||||
assert pb.is_visible("text=API Progress Update")
|
||||
assert pb.is_visible("text=Just shipped a new feature")
|
||||
|
||||
# Vote
|
||||
pb.locator("button").filter(has_text=re.compile(r"\+")).first.click()
|
||||
pb.wait_for_timeout(300)
|
||||
|
||||
# Comment
|
||||
pb.locator("textarea[name='content']").fill("Nice work! What framework did you use?")
|
||||
pb.click("button:has-text('Post')")
|
||||
pb.wait_for_timeout(500)
|
||||
assert pb.is_visible("text=Nice work! What framework did you use?")
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 9 — ALICE REPLIES TO BOB'S COMMENT
|
||||
# ═══════════════════════════════════════════════
|
||||
pa.bring_to_front()
|
||||
pa.goto(post1_url)
|
||||
assert pa.is_visible("text=Nice work! What framework did you use?")
|
||||
pa.locator(".comment-form textarea[name='content']").fill("Thanks! I used FastAPI with SQLAlchemy. Vanilla JS frontend.")
|
||||
pa.click("button:has-text('Post')")
|
||||
pa.wait_for_timeout(500)
|
||||
assert pa.is_visible("text=Thanks! I used FastAPI")
|
||||
|
||||
# Alice votes on her own post
|
||||
pa.locator("button").filter(has_text=re.compile(r"\+")).first.click()
|
||||
pa.wait_for_timeout(300)
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 10 — ALICE CHECKS NOTIFICATIONS
|
||||
# ═══════════════════════════════════════════════
|
||||
pa.bring_to_front()
|
||||
pa.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded")
|
||||
assert pa.is_visible("h2:has-text('Notifications')")
|
||||
mark_all = pa.locator("button:has-text('Mark all read')")
|
||||
if mark_all.is_visible():
|
||||
mark_all.click()
|
||||
pa.wait_for_timeout(300)
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 11 — BOB CHECKS NOTIFICATIONS TOO
|
||||
# ═══════════════════════════════════════════════
|
||||
pb.bring_to_front()
|
||||
pb.goto(f"{BASE_URL}/notifications", wait_until="domcontentloaded")
|
||||
assert pb.is_visible("h2:has-text('Notifications')")
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 12 — LOGOUT, LOGIN AGAIN
|
||||
# ═══════════════════════════════════════════════
|
||||
pa.goto(f"{BASE_URL}/auth/logout", wait_until="domcontentloaded")
|
||||
pa.wait_for_url("**/", wait_until="domcontentloaded")
|
||||
|
||||
# Log back in
|
||||
pa.click("text=Log In")
|
||||
pa.wait_for_url("**/auth/login", wait_until="domcontentloaded")
|
||||
assert pa.is_visible("h2:has-text('Welcome Back')")
|
||||
pa.fill("#email", "alice_demo@test.dev")
|
||||
pa.fill("#password", "alice_pass_1")
|
||||
pa.click("button:has-text('Sign in')")
|
||||
pa.wait_for_url("**/feed", timeout=10000, wait_until="domcontentloaded")
|
||||
assert pa.is_visible("text=alice_demo")
|
||||
|
||||
# Bob logs out
|
||||
pb.goto(f"{BASE_URL}/auth/logout", wait_until="domcontentloaded")
|
||||
pb.wait_for_url("**/", wait_until="domcontentloaded")
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# ACT 13 — FINAL VERIFICATION
|
||||
# ═══════════════════════════════════════════════
|
||||
pa.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
|
||||
assert pa.is_visible("text=Topics")
|
||||
assert pa.is_visible("text=alice_demo")
|
||||
|
||||
pa.close()
|
||||
pb.close()
|
||||
ctx_a.close()
|
||||
ctx_b.close()
|
||||
|
||||
print("\n═══════════════════════════════════════════════════")
|
||||
print(" ABSURD DEMO COMPLETE — 13 ACTS, 0 MISTAKES")
|
||||
print("═══════════════════════════════════════════════════")
|
||||
Reference in New Issue
Block a user