feat: add audit logging for admin trash restore/purge and notification clear actions
- Record audit events in `admin_trash_restore` and `admin_trash_purge` endpoints with target metadata - Log `notification.read.all` event when user clears devRant notification feed - Include `devrant` as a valid origin in audit categories - Add `admin_section` and `pagination_query` fields to audit and backup schemas for UI consistency
This commit is contained in:
@@ -0,0 +1 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
@@ -0,0 +1,81 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import time
|
||||
import requests
|
||||
from tests.conftest import BASE_URL
|
||||
|
||||
JSON_trash = {"Accept": "application/json"}
|
||||
_counter_trash = [0]
|
||||
|
||||
|
||||
def _unique_trash(prefix="tr"):
|
||||
_counter_trash[0] += 1
|
||||
return f"{prefix}{int(time.time() * 1000)}{_counter_trash[0]}"
|
||||
|
||||
|
||||
def _member_trash():
|
||||
name = _unique_trash("trmem")
|
||||
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 _admin_trash(seeded_db):
|
||||
s = requests.Session()
|
||||
creds = seeded_db["alice"]
|
||||
s.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
data={"email": creds["email"], "password": creds["password"]},
|
||||
allow_redirects=True,
|
||||
)
|
||||
return s
|
||||
|
||||
|
||||
def _create_post_trash(session):
|
||||
return session.post(
|
||||
f"{BASE_URL}/posts/create",
|
||||
headers=JSON_trash,
|
||||
data={"title": _unique_trash("trp"), "content": "trash post body text", "topic": "devlog"},
|
||||
).json()["data"]
|
||||
|
||||
|
||||
def _soft_delete_post_trash(session, slug):
|
||||
return session.post(
|
||||
f"{BASE_URL}/posts/delete/{slug}", headers=JSON_trash, allow_redirects=False
|
||||
)
|
||||
|
||||
|
||||
def test_trash_requires_admin(seeded_db):
|
||||
member, _ = _member_trash()
|
||||
r = member.get(f"{BASE_URL}/admin/trash", allow_redirects=False)
|
||||
assert r.status_code in (302, 303)
|
||||
|
||||
|
||||
def test_trash_lists_soft_deleted_post(seeded_db):
|
||||
member, _ = _member_trash()
|
||||
post = _create_post_trash(member)
|
||||
_soft_delete_post_trash(member, post["slug"])
|
||||
admin = _admin_trash(seeded_db)
|
||||
body = admin.get(
|
||||
f"{BASE_URL}/admin/trash", headers=JSON_trash, params={"table": "posts"}
|
||||
).json()
|
||||
assert body["table"] == "posts"
|
||||
assert body["admin_section"] == "trash"
|
||||
assert any(item["uid"] == post["uid"] for item in body["items"])
|
||||
|
||||
|
||||
def test_trash_unknown_table_falls_back_to_posts(seeded_db):
|
||||
admin = _admin_trash(seeded_db)
|
||||
body = admin.get(
|
||||
f"{BASE_URL}/admin/trash", headers=JSON_trash, params={"table": "bogus"}
|
||||
).json()
|
||||
assert body["table"] == "posts"
|
||||
@@ -0,0 +1,104 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import time
|
||||
import requests
|
||||
from tests.conftest import BASE_URL
|
||||
|
||||
JSON_trash = {"Accept": "application/json"}
|
||||
_counter_trash = [0]
|
||||
|
||||
|
||||
def _unique_trash(prefix="tr"):
|
||||
_counter_trash[0] += 1
|
||||
return f"{prefix}{int(time.time() * 1000)}{_counter_trash[0]}"
|
||||
|
||||
|
||||
def _member_trash():
|
||||
name = _unique_trash("trpmem")
|
||||
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 _admin_trash(seeded_db):
|
||||
s = requests.Session()
|
||||
creds = seeded_db["alice"]
|
||||
s.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
data={"email": creds["email"], "password": creds["password"]},
|
||||
allow_redirects=True,
|
||||
)
|
||||
return s
|
||||
|
||||
|
||||
def _create_post_trash(session):
|
||||
return session.post(
|
||||
f"{BASE_URL}/posts/create",
|
||||
headers=JSON_trash,
|
||||
data={"title": _unique_trash("trpp"), "content": "purge post body text", "topic": "devlog"},
|
||||
).json()["data"]
|
||||
|
||||
|
||||
def _soft_delete_post_trash(session, slug):
|
||||
session.post(f"{BASE_URL}/posts/delete/{slug}", headers=JSON_trash, allow_redirects=False)
|
||||
|
||||
|
||||
def _audit_find(admin, event_key, target_uid):
|
||||
data = admin.get(
|
||||
f"{BASE_URL}/admin/audit-log", headers=JSON_trash, params={"event_key": event_key}
|
||||
).json()
|
||||
for entry in data["entries"]:
|
||||
if entry.get("target_uid") == target_uid:
|
||||
return entry
|
||||
return None
|
||||
|
||||
|
||||
def test_purge_requires_admin(seeded_db):
|
||||
member, _ = _member_trash()
|
||||
r = member.post(
|
||||
f"{BASE_URL}/admin/trash/posts/anything/purge", allow_redirects=False
|
||||
)
|
||||
assert r.status_code in (302, 303)
|
||||
|
||||
|
||||
def test_purge_unknown_table_404(seeded_db):
|
||||
admin = _admin_trash(seeded_db)
|
||||
r = admin.post(
|
||||
f"{BASE_URL}/admin/trash/bogus/anything/purge",
|
||||
headers=JSON_trash,
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_purge_hard_deletes_and_records_audit(seeded_db):
|
||||
member, _ = _member_trash()
|
||||
post = _create_post_trash(member)
|
||||
_soft_delete_post_trash(member, post["slug"])
|
||||
admin = _admin_trash(seeded_db)
|
||||
|
||||
r = admin.post(
|
||||
f"{BASE_URL}/admin/trash/posts/{post['uid']}/purge",
|
||||
headers=JSON_trash,
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert r.status_code == 200, r.text[:300]
|
||||
|
||||
# gone from trash and not restorable (row hard-deleted)
|
||||
trash = admin.get(
|
||||
f"{BASE_URL}/admin/trash", headers=JSON_trash, params={"table": "posts"}
|
||||
).json()
|
||||
assert all(item["uid"] != post["uid"] for item in trash["items"])
|
||||
|
||||
entry = _audit_find(admin, "admin.trash.purge", post["uid"])
|
||||
assert entry is not None
|
||||
assert entry["result"] == "success"
|
||||
@@ -0,0 +1,105 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import time
|
||||
import requests
|
||||
from tests.conftest import BASE_URL
|
||||
|
||||
JSON_trash = {"Accept": "application/json"}
|
||||
_counter_trash = [0]
|
||||
|
||||
|
||||
def _unique_trash(prefix="tr"):
|
||||
_counter_trash[0] += 1
|
||||
return f"{prefix}{int(time.time() * 1000)}{_counter_trash[0]}"
|
||||
|
||||
|
||||
def _member_trash():
|
||||
name = _unique_trash("trrmem")
|
||||
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 _admin_trash(seeded_db):
|
||||
s = requests.Session()
|
||||
creds = seeded_db["alice"]
|
||||
s.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
data={"email": creds["email"], "password": creds["password"]},
|
||||
allow_redirects=True,
|
||||
)
|
||||
return s
|
||||
|
||||
|
||||
def _create_post_trash(session):
|
||||
return session.post(
|
||||
f"{BASE_URL}/posts/create",
|
||||
headers=JSON_trash,
|
||||
data={"title": _unique_trash("trrp"), "content": "restore post body text", "topic": "devlog"},
|
||||
).json()["data"]
|
||||
|
||||
|
||||
def _soft_delete_post_trash(session, slug):
|
||||
session.post(f"{BASE_URL}/posts/delete/{slug}", headers=JSON_trash, allow_redirects=False)
|
||||
|
||||
|
||||
def _audit_find(admin, event_key, target_uid):
|
||||
data = admin.get(
|
||||
f"{BASE_URL}/admin/audit-log", headers=JSON_trash, params={"event_key": event_key}
|
||||
).json()
|
||||
for entry in data["entries"]:
|
||||
if entry.get("target_uid") == target_uid:
|
||||
return entry
|
||||
return None
|
||||
|
||||
|
||||
def test_restore_requires_admin(seeded_db):
|
||||
member, _ = _member_trash()
|
||||
r = member.post(
|
||||
f"{BASE_URL}/admin/trash/posts/anything/restore", allow_redirects=False
|
||||
)
|
||||
assert r.status_code in (302, 303)
|
||||
|
||||
|
||||
def test_restore_unknown_table_404(seeded_db):
|
||||
admin = _admin_trash(seeded_db)
|
||||
r = admin.post(
|
||||
f"{BASE_URL}/admin/trash/bogus/anything/restore",
|
||||
headers=JSON_trash,
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_restore_revives_post_and_records_audit(seeded_db):
|
||||
member, _ = _member_trash()
|
||||
post = _create_post_trash(member)
|
||||
_soft_delete_post_trash(member, post["slug"])
|
||||
admin = _admin_trash(seeded_db)
|
||||
|
||||
r = admin.post(
|
||||
f"{BASE_URL}/admin/trash/posts/{post['uid']}/restore",
|
||||
headers=JSON_trash,
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert r.status_code == 200, r.text[:300]
|
||||
|
||||
# restored row is live again and gone from trash
|
||||
trash = admin.get(
|
||||
f"{BASE_URL}/admin/trash", headers=JSON_trash, params={"table": "posts"}
|
||||
).json()
|
||||
assert all(item["uid"] != post["uid"] for item in trash["items"])
|
||||
assert member.get(f"{BASE_URL}/posts/{post['slug']}").status_code == 200
|
||||
|
||||
entry = _audit_find(admin, "admin.trash.restore", post["uid"])
|
||||
assert entry is not None
|
||||
assert entry["result"] == "success"
|
||||
@@ -0,0 +1 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
@@ -0,0 +1 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
@@ -0,0 +1,79 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import time
|
||||
import requests
|
||||
from tests.conftest import BASE_URL
|
||||
|
||||
JSON_notif = {"Accept": "application/json"}
|
||||
_counter_notif = [0]
|
||||
|
||||
|
||||
def _unique_notif(prefix="dn"):
|
||||
_counter_notif[0] += 1
|
||||
return f"{prefix}{int(time.time() * 1000)}{_counter_notif[0]}"
|
||||
|
||||
|
||||
def _devrant_user():
|
||||
name = _unique_notif("dnmem")
|
||||
password = "secret123"
|
||||
s = requests.Session()
|
||||
s.post(
|
||||
f"{BASE_URL}/auth/signup",
|
||||
data={
|
||||
"username": name,
|
||||
"email": f"{name}@t.dev",
|
||||
"password": password,
|
||||
"confirm_password": password,
|
||||
},
|
||||
allow_redirects=True,
|
||||
)
|
||||
r = s.post(
|
||||
f"{BASE_URL}/api/users/auth-token",
|
||||
data={"username": name, "password": password},
|
||||
)
|
||||
body = r.json()
|
||||
assert body["success"] is True, body
|
||||
token = body["auth_token"]
|
||||
return name, {
|
||||
"token_id": token["id"],
|
||||
"token_key": token["key"],
|
||||
"user_id": token["user_id"],
|
||||
}
|
||||
|
||||
|
||||
def _admin_notif(seeded_db):
|
||||
s = requests.Session()
|
||||
creds = seeded_db["alice"]
|
||||
s.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
data={"email": creds["email"], "password": creds["password"]},
|
||||
allow_redirects=True,
|
||||
)
|
||||
return s
|
||||
|
||||
|
||||
def _audit_find(admin, event_key, username):
|
||||
data = admin.get(
|
||||
f"{BASE_URL}/admin/audit-log", headers=JSON_notif, params={"event_key": event_key}
|
||||
).json()
|
||||
for entry in data["entries"]:
|
||||
if entry.get("actor_username") == username:
|
||||
return entry
|
||||
return None
|
||||
|
||||
|
||||
def test_clear_notif_feed_requires_token(seeded_db):
|
||||
r = requests.delete(f"{BASE_URL}/api/users/me/notif-feed")
|
||||
assert r.json()["success"] is False
|
||||
|
||||
|
||||
def test_clear_notif_feed_succeeds_and_records_audit(seeded_db):
|
||||
name, token = _devrant_user()
|
||||
r = requests.delete(f"{BASE_URL}/api/users/me/notif-feed", data=token)
|
||||
body = r.json()
|
||||
assert body["success"] is True, body
|
||||
|
||||
admin = _admin_notif(seeded_db)
|
||||
entry = _audit_find(admin, "notification.read.all", name)
|
||||
assert entry is not None
|
||||
assert entry["origin"] == "devrant"
|
||||
@@ -428,3 +428,19 @@ def test_project_uid_redirects_to_canonical_slug(app_server):
|
||||
assert r.headers["location"].endswith(f"/projects/{slug}"), r.headers.get(
|
||||
"location"
|
||||
)
|
||||
|
||||
|
||||
def test_private_detail_is_noindex_for_owner(app_server):
|
||||
_, _, key = _signup_project_visibility()
|
||||
slug = _create_project_project_visibility(key, "Private NoIndex", is_private=True)["slug"]
|
||||
r = requests.get(f"{BASE_URL}/projects/{slug}", headers={"X-API-KEY": key})
|
||||
assert r.status_code == 200, r.text[:300]
|
||||
assert '<meta name="robots" content="noindex,nofollow">' in r.text
|
||||
|
||||
|
||||
def test_public_detail_is_indexable(app_server):
|
||||
_, _, key = _signup_project_visibility()
|
||||
slug = _create_project_project_visibility(key, "Public Indexable", is_private=False)["slug"]
|
||||
r = requests.get(f"{BASE_URL}/projects/{slug}", headers={"X-API-KEY": key})
|
||||
assert r.status_code == 200, r.text[:300]
|
||||
assert '<meta name="robots" content="index,follow">' in r.text
|
||||
|
||||
@@ -35,3 +35,16 @@ def test_db_mutation_tools_record_and_confirm():
|
||||
assert action.is_read_only is False
|
||||
assert _records_mutation(name) is True
|
||||
assert name in CONFIRM_REQUIRED
|
||||
|
||||
|
||||
def test_seo_report_is_public_read_only_http_action():
|
||||
action = BY_NAME["seo_report"]
|
||||
assert action.handler == "http"
|
||||
assert action.method == "GET"
|
||||
assert action.path == "/tools/seo/{uid}/report"
|
||||
assert action.requires_auth is False
|
||||
assert action.is_read_only is True
|
||||
assert _records_mutation("seo_report") is False
|
||||
assert "seo_report" not in CONFIRM_REQUIRED
|
||||
names = {p.name for p in action.params}
|
||||
assert "uid" in names
|
||||
|
||||
Reference in New Issue
Block a user