chore: reorganize test files into domain-specific subdirectories under tests/
Split the monolithic test directory into three tiers (unit, api, e2e) with a path-mirroring directory structure. Added corresponding Makefile targets (test-unit, test-api, test-e2e) and updated all documentation references (CLAUDE.md, README.md, testing-cicd.html, testing-framework.html, testing-make.html) to reflect the new layout and naming conventions.
This commit is contained in:
@@ -0,0 +1,74 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from types import SimpleNamespace
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import get_table, set_setting
|
||||
from devplacepy.utils import generate_uid
|
||||
from devplacepy.services.devii.config import (
|
||||
effective_daily_limit,
|
||||
FIELD_USER_DAILY_USD,
|
||||
FIELD_GUEST_DAILY_USD,
|
||||
FIELD_ADMIN_DAILY_USD,
|
||||
)
|
||||
from devplacepy.services.devii.store import UsageLedger
|
||||
LEDGER = "devii_usage_ledger"
|
||||
def _now_iso_devii_quota():
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
def _seed_ledger(owner_kind, owner_id, cost, n=1):
|
||||
table = get_table(LEDGER)
|
||||
for _ in range(n):
|
||||
table.insert(
|
||||
{
|
||||
"owner_kind": owner_kind,
|
||||
"owner_id": owner_id,
|
||||
"created_at": _now_iso_devii_quota(),
|
||||
"cost_usd": cost,
|
||||
}
|
||||
)
|
||||
def _ensure_devii():
|
||||
from devplacepy.services.manager import service_manager
|
||||
from devplacepy.services.devii import DeviiService
|
||||
|
||||
if service_manager.get_service("devii") is None:
|
||||
service_manager.register(DeviiService())
|
||||
return service_manager.get_service("devii")
|
||||
def _make_user_devii_quota(role="Member"):
|
||||
uid = generate_uid()
|
||||
username = f"quota_{uid[-12:]}"
|
||||
get_table("users").insert(
|
||||
{
|
||||
"uid": uid,
|
||||
"username": username,
|
||||
"email": f"{username}@t.dev",
|
||||
"api_key": generate_uid(),
|
||||
"role": role,
|
||||
"is_active": True,
|
||||
}
|
||||
)
|
||||
return uid, username
|
||||
|
||||
|
||||
def test_admin_reset_all_endpoint(alice):
|
||||
page, _ = alice
|
||||
_seed_ledger("guest", f"guest-{generate_uid()[:8]}", 0.01)
|
||||
_seed_ledger("user", _make_user_devii_quota()[0], 0.5)
|
||||
resp = page.request.post(f"{BASE_URL}/admin/ai-quota/reset-all")
|
||||
assert resp.ok
|
||||
assert get_table(LEDGER).count() == 0
|
||||
|
||||
|
||||
def test_non_admin_cannot_reset_quota(bob):
|
||||
page, _ = bob
|
||||
marker = f"guest-{generate_uid()[:8]}"
|
||||
get_table(LEDGER).insert(
|
||||
{
|
||||
"owner_kind": "guest",
|
||||
"owner_id": marker,
|
||||
"created_at": _now_iso_devii_quota(),
|
||||
"cost_usd": 0.01,
|
||||
}
|
||||
)
|
||||
page.request.post(f"{BASE_URL}/admin/ai-quota/reset-all")
|
||||
assert get_table(LEDGER).count(owner_kind="guest", owner_id=marker) == 1
|
||||
get_table(LEDGER).delete(owner_kind="guest", owner_id=marker)
|
||||
@@ -0,0 +1,73 @@
|
||||
# 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(
|
||||
{
|
||||
"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}",
|
||||
"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]}",
|
||||
"email": f"target_{uid[:8]}@test.devplace",
|
||||
"password_hash": "x",
|
||||
"role": "Member",
|
||||
"is_active": True,
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
}
|
||||
)
|
||||
return uid
|
||||
|
||||
|
||||
def test_admin_ai_usage_data(alice):
|
||||
page, _ = alice
|
||||
resp = page.request.get(f"{BASE_URL}/admin/ai-usage/data?hours=48")
|
||||
assert resp.status == 200
|
||||
data = resp.json()
|
||||
assert "hourly" in data
|
||||
assert "cost" in data
|
||||
assert "requests" in data
|
||||
@@ -0,0 +1,69 @@
|
||||
# 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(
|
||||
{
|
||||
"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}",
|
||||
"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]}",
|
||||
"email": f"target_{uid[:8]}@test.devplace",
|
||||
"password_hash": "x",
|
||||
"role": "Member",
|
||||
"is_active": True,
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
}
|
||||
)
|
||||
return uid
|
||||
|
||||
|
||||
def test_admin_ai_usage_page_loads(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/admin/ai-usage", wait_until="domcontentloaded")
|
||||
assert page.is_visible("text=AI usage") or page.is_visible("text=AI Usage")
|
||||
@@ -0,0 +1,73 @@
|
||||
# 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(
|
||||
{
|
||||
"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}",
|
||||
"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]}",
|
||||
"email": f"target_{uid[:8]}@test.devplace",
|
||||
"password_hash": "x",
|
||||
"role": "Member",
|
||||
"is_active": True,
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
}
|
||||
)
|
||||
return uid
|
||||
|
||||
|
||||
def test_admin_analytics_page(alice):
|
||||
page, _ = alice
|
||||
resp = page.request.get(f"{BASE_URL}/admin/analytics")
|
||||
assert resp.status == 200
|
||||
data = resp.json()
|
||||
assert "total_members" in data
|
||||
assert "totals" in data
|
||||
assert "posts" in data["totals"]
|
||||
@@ -0,0 +1,274 @@
|
||||
# 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(
|
||||
{
|
||||
"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}",
|
||||
"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]}",
|
||||
"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",
|
||||
},
|
||||
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(
|
||||
{
|
||||
"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"
|
||||
@@ -0,0 +1,21 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import requests
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import get_table
|
||||
|
||||
|
||||
def test_containers_admin_data_returns_json(alice, seeded_db):
|
||||
s = requests.Session()
|
||||
s.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
data={
|
||||
"email": seeded_db["alice"]["email"],
|
||||
"password": seeded_db["alice"]["password"],
|
||||
},
|
||||
)
|
||||
r = s.get(f"{BASE_URL}/admin/containers/data")
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert "instances" in data
|
||||
assert isinstance(data["instances"], list)
|
||||
@@ -0,0 +1,17 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import requests
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import get_table
|
||||
|
||||
|
||||
def test_containers_admin_page_loads(alice, app_server):
|
||||
page, user = alice
|
||||
page.goto(f"{BASE_URL}/admin/containers", wait_until="domcontentloaded")
|
||||
assert page.is_visible(".admin-toolbar h2:has-text('Containers')")
|
||||
|
||||
|
||||
def test_container_instance_page_404_for_missing(alice, app_server):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/admin/containers/nonexistent", wait_until="domcontentloaded")
|
||||
assert page.is_visible("text=Not Found") or page.is_visible("text=404")
|
||||
@@ -0,0 +1,68 @@
|
||||
# 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(
|
||||
{
|
||||
"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}",
|
||||
"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]}",
|
||||
"email": f"target_{uid[:8]}@test.devplace",
|
||||
"password_hash": "x",
|
||||
"role": "Member",
|
||||
"is_active": True,
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
}
|
||||
)
|
||||
return uid
|
||||
|
||||
|
||||
def test_admin_redirect_unauth(page, app_server):
|
||||
page.goto(f"{BASE_URL}/admin", wait_until="domcontentloaded")
|
||||
assert page.url.startswith(f"{BASE_URL}/auth/login")
|
||||
@@ -0,0 +1,69 @@
|
||||
# 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(
|
||||
{
|
||||
"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}",
|
||||
"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]}",
|
||||
"email": f"target_{uid[:8]}@test.devplace",
|
||||
"password_hash": "x",
|
||||
"role": "Member",
|
||||
"is_active": True,
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
}
|
||||
)
|
||||
return uid
|
||||
|
||||
|
||||
def test_admin_media_page_loads(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/admin/media", wait_until="domcontentloaded")
|
||||
assert page.is_visible("text=Media") or page.is_visible("text=Deleted Media")
|
||||
@@ -0,0 +1,145 @@
|
||||
# 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(
|
||||
{
|
||||
"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}",
|
||||
"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]}",
|
||||
"email": f"target_{uid[:8]}@test.devplace",
|
||||
"password_hash": "x",
|
||||
"role": "Member",
|
||||
"is_active": True,
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
}
|
||||
)
|
||||
return uid
|
||||
|
||||
|
||||
def test_admin_news_loads(alice):
|
||||
page, _ = alice
|
||||
seed_admin_news()
|
||||
page.goto(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
|
||||
assert page.is_visible("text=News")
|
||||
assert page.is_visible("text=Admin News Article")
|
||||
|
||||
|
||||
def test_admin_news_publish_toggle(alice):
|
||||
page, _ = alice
|
||||
seed_admin_news()
|
||||
page.goto(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
|
||||
toggle_sel = "form[action*='/publish'] button.admin-toggle-switch"
|
||||
was_active = "active" in (
|
||||
page.locator(toggle_sel).first.get_attribute("class") or ""
|
||||
)
|
||||
page.locator(toggle_sel).first.click()
|
||||
page.wait_for_url(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
|
||||
is_active = "active" in (
|
||||
page.locator(toggle_sel).first.get_attribute("class") or ""
|
||||
)
|
||||
assert is_active != was_active
|
||||
|
||||
|
||||
def test_admin_news_landing_toggle(alice):
|
||||
page, _ = alice
|
||||
seed_admin_news()
|
||||
page.goto(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
|
||||
toggle_sel = "form[action*='/landing'] button.admin-toggle-switch"
|
||||
was_active = "active" in (
|
||||
page.locator(toggle_sel).first.get_attribute("class") or ""
|
||||
)
|
||||
page.locator(toggle_sel).first.click()
|
||||
page.wait_for_url(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
|
||||
is_active = "active" in (
|
||||
page.locator(toggle_sel).first.get_attribute("class") or ""
|
||||
)
|
||||
assert is_active != was_active
|
||||
|
||||
|
||||
def test_admin_news_delete(alice):
|
||||
page, _ = alice
|
||||
seed_admin_news()
|
||||
page.goto(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
|
||||
initial_count = len(page.locator("table tbody tr").all())
|
||||
page.locator("form[action*='/delete'] button").first.click()
|
||||
page.locator(".dialog-overlay.visible .dialog-confirm").click()
|
||||
page.wait_for_url(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
|
||||
page.wait_for_timeout(1000)
|
||||
new_count = len(page.locator("table tbody tr").all())
|
||||
assert new_count < initial_count
|
||||
|
||||
|
||||
def test_admin_news_pagination(alice):
|
||||
page, _ = alice
|
||||
seed_admin_news(30)
|
||||
page.goto(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
|
||||
assert page.text_content(".pagination-info") is not None
|
||||
assert "Page 1" in page.text_content(".pagination-info")
|
||||
next_btn = page.locator("a.pagination-btn:has-text('Next')")
|
||||
assert next_btn.count() > 0, "30 seeded articles must span more than one page"
|
||||
next_btn.first.click()
|
||||
page.wait_for_url(f"{BASE_URL}/admin/news?page=2", wait_until="domcontentloaded")
|
||||
assert "Page 2" in page.text_content(".pagination-info")
|
||||
|
||||
|
||||
def test_admin_news_featured_toggle(alice):
|
||||
page, _ = alice
|
||||
seed_admin_news()
|
||||
page.goto(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
|
||||
toggle_sel = "form[action$='/toggle'] button.admin-toggle-switch"
|
||||
was_active = "active" in (
|
||||
page.locator(toggle_sel).first.get_attribute("class") or ""
|
||||
)
|
||||
page.locator(toggle_sel).first.click()
|
||||
page.wait_for_url(f"{BASE_URL}/admin/news", wait_until="domcontentloaded")
|
||||
is_active = "active" in (
|
||||
page.locator(toggle_sel).first.get_attribute("class") or ""
|
||||
)
|
||||
assert is_active != was_active
|
||||
@@ -0,0 +1,48 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import requests
|
||||
from playwright.sync_api import expect
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import (
|
||||
get_notification_default,
|
||||
get_notification_prefs,
|
||||
get_table,
|
||||
notification_enabled,
|
||||
reset_notification_prefs,
|
||||
set_notification_default,
|
||||
set_notification_pref,
|
||||
)
|
||||
from devplacepy.utils import create_notification, generate_uid
|
||||
def _user_notification_prefs(username):
|
||||
return get_table("users").find_one(username=username)
|
||||
def _pref(prefs, key):
|
||||
return next(p for p in prefs if p["key"] == key)
|
||||
|
||||
|
||||
def test_admin_default_endpoint(alice):
|
||||
_, admin_user = alice
|
||||
admin = _user_notification_prefs(admin_user["username"])
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{BASE_URL}/admin/notifications",
|
||||
data={"notification_type": "level", "channel": "push", "value": "false"},
|
||||
headers={"Accept": "application/json", "X-API-KEY": admin["api_key"]},
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert get_notification_default("level", "push") is False
|
||||
finally:
|
||||
set_notification_default("level", "push", True)
|
||||
|
||||
|
||||
def test_member_cannot_set_default(bob):
|
||||
_, user = bob
|
||||
member = _user_notification_prefs(user["username"])
|
||||
response = requests.post(
|
||||
f"{BASE_URL}/admin/notifications",
|
||||
data={"notification_type": "level", "channel": "push", "value": "false"},
|
||||
headers={"Accept": "application/json", "X-API-KEY": member["api_key"]},
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert response.status_code in (302, 303, 403)
|
||||
assert get_notification_default("level", "push") is True
|
||||
@@ -0,0 +1,20 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from tests.conftest import BASE_URL, login_user
|
||||
from devplacepy.database import get_table
|
||||
def _promote_to_admin(username: str) -> None:
|
||||
users = get_table("users")
|
||||
user = users.find_one(username=username)
|
||||
if user:
|
||||
users.update({"uid": user["uid"], "role": "Admin"}, ["uid"])
|
||||
|
||||
|
||||
def test_services_clear_logs(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services", wait_until="domcontentloaded")
|
||||
resp = page.request.post(f"{BASE_URL}/admin/services/news/clear-logs")
|
||||
assert resp.status == 200
|
||||
data = resp.json()
|
||||
assert data["ok"] is True
|
||||
@@ -0,0 +1,56 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from tests.conftest import BASE_URL, login_user
|
||||
from devplacepy.database import get_table
|
||||
def _promote_to_admin(username: str) -> None:
|
||||
users = get_table("users")
|
||||
user = users.find_one(username=username)
|
||||
if user:
|
||||
users.update({"uid": user["uid"], "role": "Admin"}, ["uid"])
|
||||
|
||||
|
||||
def test_services_config_validation(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services", wait_until="domcontentloaded")
|
||||
bad = page.request.post(
|
||||
f"{BASE_URL}/admin/services/news/config", form={"news_grade_threshold": "99"}
|
||||
)
|
||||
assert bad.status == 400
|
||||
body = bad.json()
|
||||
assert body["ok"] is False
|
||||
assert "news_grade_threshold" in body["errors"]
|
||||
try:
|
||||
good = page.request.post(
|
||||
f"{BASE_URL}/admin/services/news/config", form={"news_grade_threshold": "8"}
|
||||
)
|
||||
assert good.ok
|
||||
assert good.json()["ok"] is True
|
||||
finally:
|
||||
page.request.post(
|
||||
f"{BASE_URL}/admin/services/news/config", form={"news_grade_threshold": "7"}
|
||||
)
|
||||
|
||||
|
||||
def test_bots_config_float_validation(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services", wait_until="domcontentloaded")
|
||||
bad = page.request.post(
|
||||
f"{BASE_URL}/admin/services/bots/config", form={"bot_input_cost_per_1m": "abc"}
|
||||
)
|
||||
assert bad.status == 400
|
||||
assert "bot_input_cost_per_1m" in bad.json()["errors"]
|
||||
try:
|
||||
good = page.request.post(
|
||||
f"{BASE_URL}/admin/services/bots/config",
|
||||
form={"bot_input_cost_per_1m": "0.42"},
|
||||
)
|
||||
assert good.ok and good.json()["ok"] is True
|
||||
finally:
|
||||
page.request.post(
|
||||
f"{BASE_URL}/admin/services/bots/config",
|
||||
form={"bot_input_cost_per_1m": "0.27"},
|
||||
)
|
||||
@@ -0,0 +1,47 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from tests.conftest import BASE_URL, login_user
|
||||
from devplacepy.database import get_table
|
||||
def _promote_to_admin(username: str) -> None:
|
||||
users = get_table("users")
|
||||
user = users.find_one(username=username)
|
||||
if user:
|
||||
users.update({"uid": user["uid"], "role": "Admin"}, ["uid"])
|
||||
|
||||
|
||||
def test_services_data_json(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services/data", wait_until="domcontentloaded")
|
||||
text = page.inner_text("body > pre") if page.locator("body > pre").count() > 0 else page.content()
|
||||
import json
|
||||
|
||||
data = json.loads(text)
|
||||
assert "services" in data
|
||||
assert isinstance(data["services"], list)
|
||||
|
||||
|
||||
def test_service_detail_unknown_404(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
resp = page.request.get(f"{BASE_URL}/admin/services/nope")
|
||||
assert resp.status == 404
|
||||
assert page.request.get(f"{BASE_URL}/admin/services/nope/data").status == 404
|
||||
|
||||
|
||||
def test_bots_service_registered_and_opt_in(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services", wait_until="domcontentloaded")
|
||||
data = page.request.get(f"{BASE_URL}/admin/services/data").json()
|
||||
bots = next(s for s in data["services"] if s["name"] == "bots")
|
||||
assert bots["enabled"] is False
|
||||
assert bots["status"] == "stopped"
|
||||
assert "metrics" in bots
|
||||
page.goto(f"{BASE_URL}/admin/services/bots", wait_until="domcontentloaded")
|
||||
detail = page.locator("[data-service-detail][data-service='bots']")
|
||||
assert detail.locator("input[name='bot_fleet_size']").count() == 1
|
||||
assert detail.locator("input[name='bot_input_cost_per_1m']").count() == 1
|
||||
@@ -0,0 +1,62 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from tests.conftest import BASE_URL, login_user
|
||||
from devplacepy.database import get_table
|
||||
def _promote_to_admin(username: str) -> None:
|
||||
users = get_table("users")
|
||||
user = users.find_one(username=username)
|
||||
if user:
|
||||
users.update({"uid": user["uid"], "role": "Admin"}, ["uid"])
|
||||
|
||||
|
||||
def test_services_page_non_admin_redirect(bob):
|
||||
page, _ = bob
|
||||
page.goto(f"{BASE_URL}/admin/services", wait_until="domcontentloaded")
|
||||
assert page.url.rstrip("/").endswith("/feed")
|
||||
assert "/admin/services" not in page.url
|
||||
|
||||
|
||||
def test_services_page_admin_loads(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services", wait_until="domcontentloaded")
|
||||
assert page.is_visible("text=Services")
|
||||
|
||||
|
||||
def test_services_sidebar_link(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services", wait_until="domcontentloaded")
|
||||
link = page.locator("a.sidebar-link[href='/admin/services']")
|
||||
assert link.is_visible()
|
||||
assert "active" in (link.get_attribute("class") or "")
|
||||
|
||||
|
||||
def test_services_index_lists_with_links(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services", wait_until="domcontentloaded")
|
||||
row = page.locator(".service-row[data-service='news']")
|
||||
assert row.count() == 1
|
||||
assert row.locator("a[href='/admin/services/news']").count() >= 1
|
||||
assert page.locator(".service-row-desc").count() >= 1
|
||||
# index is a slim list: no config forms or per-service controls
|
||||
assert page.locator("[data-config-form]").count() == 0
|
||||
|
||||
|
||||
def test_service_detail_has_tabs_and_config(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services/news", wait_until="domcontentloaded")
|
||||
detail = page.locator("[data-service-detail][data-service='news']")
|
||||
assert detail.count() == 1
|
||||
assert detail.locator("[data-tabs] [data-tab='config']").count() == 1
|
||||
assert detail.locator("[data-config-form]").count() == 1
|
||||
assert detail.locator("input[name='news_grade_threshold']").count() == 1
|
||||
for action in ("start", "stop", "run", "clear-logs"):
|
||||
assert detail.locator(f"[data-action='{action}']").count() == 1
|
||||
assert detail.locator("fieldset.config-section").count() >= 1
|
||||
@@ -0,0 +1,20 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from tests.conftest import BASE_URL, login_user
|
||||
from devplacepy.database import get_table
|
||||
def _promote_to_admin(username: str) -> None:
|
||||
users = get_table("users")
|
||||
user = users.find_one(username=username)
|
||||
if user:
|
||||
users.update({"uid": user["uid"], "role": "Admin"}, ["uid"])
|
||||
|
||||
|
||||
def test_services_run_now(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services", wait_until="domcontentloaded")
|
||||
resp = page.request.post(f"{BASE_URL}/admin/services/news/run")
|
||||
assert resp.status == 200
|
||||
data = resp.json()
|
||||
assert data["ok"] is True
|
||||
@@ -0,0 +1,28 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from tests.conftest import BASE_URL, login_user
|
||||
from devplacepy.database import get_table
|
||||
def _promote_to_admin(username: str) -> None:
|
||||
users = get_table("users")
|
||||
user = users.find_one(username=username)
|
||||
if user:
|
||||
users.update({"uid": user["uid"], "role": "Admin"}, ["uid"])
|
||||
|
||||
|
||||
def test_services_start_stop_toggles_enabled(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services", wait_until="domcontentloaded")
|
||||
try:
|
||||
assert page.request.post(f"{BASE_URL}/admin/services/news/stop").ok
|
||||
data = page.request.get(f"{BASE_URL}/admin/services/data").json()
|
||||
news = next(s for s in data["services"] if s["name"] == "news")
|
||||
assert news["enabled"] is False
|
||||
assert news["status"] == "stopped"
|
||||
assert page.request.post(f"{BASE_URL}/admin/services/news/start").ok
|
||||
data = page.request.get(f"{BASE_URL}/admin/services/data").json()
|
||||
news = next(s for s in data["services"] if s["name"] == "news")
|
||||
assert news["enabled"] is True
|
||||
finally:
|
||||
page.request.post(f"{BASE_URL}/admin/services/news/start")
|
||||
@@ -0,0 +1,17 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from tests.conftest import BASE_URL, login_user
|
||||
from devplacepy.database import get_table
|
||||
def _promote_to_admin(username: str) -> None:
|
||||
users = get_table("users")
|
||||
user = users.find_one(username=username)
|
||||
if user:
|
||||
users.update({"uid": user["uid"], "role": "Admin"}, ["uid"])
|
||||
|
||||
|
||||
def test_services_unknown_returns_404(page, seeded_db):
|
||||
user = seeded_db["alice"]
|
||||
_promote_to_admin(user["username"])
|
||||
login_user(page, user)
|
||||
page.goto(f"{BASE_URL}/admin/services", wait_until="domcontentloaded")
|
||||
assert page.request.post(f"{BASE_URL}/admin/services/nope/stop").status == 404
|
||||
@@ -0,0 +1,125 @@
|
||||
# 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(
|
||||
{
|
||||
"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}",
|
||||
"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]}",
|
||||
"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 requests
|
||||
DEFAULT_MAINTENANCE_MESSAGE = (
|
||||
"DevPlace is undergoing scheduled maintenance. Please check back shortly."
|
||||
)
|
||||
OPERATIONAL_FIELDS = (
|
||||
"rate_limit_per_minute",
|
||||
"rate_limit_window_seconds",
|
||||
"session_max_age_days",
|
||||
"session_remember_days",
|
||||
"registration_open",
|
||||
"maintenance_mode",
|
||||
"maintenance_message",
|
||||
)
|
||||
def _save_settings(page, **fields):
|
||||
page.goto(f"{BASE_URL}/admin/settings", wait_until="domcontentloaded")
|
||||
for name, value in fields.items():
|
||||
locator = page.locator(f"#{name}")
|
||||
tag = locator.evaluate("el => el.tagName.toLowerCase()")
|
||||
if tag == "select":
|
||||
page.select_option(f"#{name}", value)
|
||||
else:
|
||||
locator.fill(value)
|
||||
page.click("button:has-text('Save Settings')")
|
||||
page.wait_for_url("**/admin/settings", wait_until="domcontentloaded")
|
||||
|
||||
|
||||
def test_admin_settings_save(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/admin/settings", wait_until="domcontentloaded")
|
||||
assert page.is_visible("#site_name")
|
||||
original = page.locator("#site_tagline").input_value()
|
||||
newval = f"tagline-{uuid4().hex[:8]}"
|
||||
try:
|
||||
page.fill("#site_tagline", newval)
|
||||
page.click("button:has-text('Save Settings')")
|
||||
page.wait_for_url("**/admin/settings", wait_until="domcontentloaded")
|
||||
assert page.locator("#site_tagline").input_value() == newval
|
||||
finally:
|
||||
page.fill("#site_tagline", original)
|
||||
page.click("button:has-text('Save Settings')")
|
||||
page.wait_for_url("**/admin/settings", wait_until="domcontentloaded")
|
||||
|
||||
|
||||
def test_operational_fields_render(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/admin/settings", wait_until="domcontentloaded")
|
||||
assert page.is_visible("text=Operational")
|
||||
for field in OPERATIONAL_FIELDS:
|
||||
assert page.is_visible(f"#{field}"), field
|
||||
|
||||
|
||||
def test_maintenance_mode_admin_retains_access(alice):
|
||||
page, _ = alice
|
||||
try:
|
||||
_save_settings(page, maintenance_mode="1")
|
||||
page.goto(f"{BASE_URL}/admin/settings", wait_until="domcontentloaded")
|
||||
assert page.is_visible("#maintenance_mode")
|
||||
login = requests.get(f"{BASE_URL}/auth/login")
|
||||
assert login.status_code == 200
|
||||
finally:
|
||||
_save_settings(page, maintenance_mode="0")
|
||||
@@ -0,0 +1,97 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from starlette.requests import Request
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import get_table, set_setting
|
||||
from devplacepy.utils import generate_uid
|
||||
from devplacepy.services.openai_gateway import GatewayService
|
||||
from devplacepy.services.openai_gateway.analytics import build_user_usage
|
||||
from devplacepy.services.devii.config import (
|
||||
build_settings,
|
||||
FIELD_AI_URL,
|
||||
FIELD_AI_MODEL,
|
||||
FIELD_AI_KEY,
|
||||
FIELD_MAX_ITERATIONS,
|
||||
FIELD_PLAN_REQUIRED,
|
||||
FIELD_VERIFY_REQUIRED,
|
||||
)
|
||||
def _now_iso_ai_usage_profile():
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
def _make_user_ai_usage_profile(role="Member"):
|
||||
username = f"aiu_{generate_uid()[:8]}"
|
||||
api_key = generate_uid()
|
||||
uid = generate_uid()
|
||||
get_table("users").insert(
|
||||
{
|
||||
"uid": uid,
|
||||
"username": username,
|
||||
"email": f"{username}@t.dev",
|
||||
"api_key": api_key,
|
||||
"role": role,
|
||||
"is_active": True,
|
||||
}
|
||||
)
|
||||
return uid, username, api_key
|
||||
def _make_request_ai_usage_profile(headers):
|
||||
raw = [(k.lower().encode(), v.encode()) for k, v in headers.items()]
|
||||
return Request(
|
||||
{
|
||||
"type": "http",
|
||||
"method": "POST",
|
||||
"path": "/openai/v1/chat/completions",
|
||||
"query_string": b"",
|
||||
"headers": raw,
|
||||
"state": {},
|
||||
}
|
||||
)
|
||||
def _seed_gateway(uid, n=3, cost=0.01, kind="user"):
|
||||
table = get_table("gateway_usage_ledger")
|
||||
for _ in range(n):
|
||||
table.insert(
|
||||
{
|
||||
"created_at": _now_iso_ai_usage_profile(),
|
||||
"owner_kind": kind,
|
||||
"owner_id": uid,
|
||||
"backend": "chat",
|
||||
"model": "molodetz",
|
||||
"success": 1,
|
||||
"status_code": 200,
|
||||
"prompt_tokens": 1000,
|
||||
"completion_tokens": 300,
|
||||
"total_tokens": 1300,
|
||||
"upstream_latency_ms": 1500.0,
|
||||
"tokens_per_second": 40.0,
|
||||
"cost_usd": cost,
|
||||
}
|
||||
)
|
||||
def _devii_cfg():
|
||||
return {
|
||||
FIELD_AI_URL: "",
|
||||
FIELD_AI_MODEL: "",
|
||||
FIELD_AI_KEY: "INTERNAL-KEY",
|
||||
FIELD_MAX_ITERATIONS: "40",
|
||||
FIELD_PLAN_REQUIRED: "1",
|
||||
FIELD_VERIFY_REQUIRED: "1",
|
||||
}
|
||||
def _goto_profile(page, username):
|
||||
page.goto(f"{BASE_URL}/profile/{username}", wait_until="domcontentloaded")
|
||||
|
||||
|
||||
def test_admin_user_ai_usage_endpoint_returns_data(alice, seeded_db):
|
||||
page, _ = alice
|
||||
alice_uid = get_table("users").find_one(username=seeded_db["alice"]["username"])[
|
||||
"uid"
|
||||
]
|
||||
resp = page.request.get(f"{BASE_URL}/admin/users/{alice_uid}/ai-usage?hours=24")
|
||||
assert resp.status == 200
|
||||
data = resp.json()
|
||||
assert data["owner_id"] == alice_uid
|
||||
assert "cost" in data and "requests" in data and "tokens" in data
|
||||
|
||||
|
||||
def test_member_forbidden_from_user_ai_usage_endpoint(bob, seeded_db):
|
||||
page, _ = bob
|
||||
bob_uid = get_table("users").find_one(username=seeded_db["bob"]["username"])["uid"]
|
||||
resp = page.request.get(f"{BASE_URL}/admin/users/{bob_uid}/ai-usage")
|
||||
assert resp.status == 403
|
||||
@@ -0,0 +1,366 @@
|
||||
# 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(
|
||||
{
|
||||
"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}",
|
||||
"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]}",
|
||||
"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",
|
||||
},
|
||||
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(
|
||||
{
|
||||
"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
|
||||
import io
|
||||
import uuid
|
||||
from PIL import Image
|
||||
JSON_media = {"Accept": "application/json"}
|
||||
def _png_bytes_media(color=(200, 30, 30)):
|
||||
buf = io.BytesIO()
|
||||
Image.new("RGB", (8, 8), color).save(buf, "PNG")
|
||||
return buf.getvalue()
|
||||
def _signup_media(prefix="media"):
|
||||
s = requests.Session()
|
||||
name = f"{prefix}_{uuid.uuid4().hex[:10]}"
|
||||
s.post(
|
||||
f"{BASE_URL}/auth/signup",
|
||||
data={
|
||||
"username": name,
|
||||
"email": f"{name}@test.devplace",
|
||||
"password": "secret123",
|
||||
"confirm_password": "secret123",
|
||||
},
|
||||
allow_redirects=True,
|
||||
)
|
||||
return s, name
|
||||
def _login_media(seeded_db, who):
|
||||
s = requests.Session()
|
||||
creds = seeded_db[who]
|
||||
s.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
data={"email": creds["email"], "password": creds["password"]},
|
||||
allow_redirects=True,
|
||||
)
|
||||
return s
|
||||
def _upload_media(s, name="pic.png", content=None, mime="image/png"):
|
||||
files = {"file": (name, content if content is not None else _png_bytes_media(), mime)}
|
||||
r = s.post(f"{BASE_URL}/uploads/upload", files=files)
|
||||
assert r.status_code == 201, r.text
|
||||
return r.json()["uid"]
|
||||
def _create_post_media(s, attachment_uids, title="media post"):
|
||||
r = s.post(
|
||||
f"{BASE_URL}/posts/create",
|
||||
data={
|
||||
"content": "A post that carries media for the gallery tests.",
|
||||
"title": title,
|
||||
"topic": "random",
|
||||
"attachment_uids": attachment_uids,
|
||||
},
|
||||
allow_redirects=True,
|
||||
)
|
||||
assert r.status_code == 200, r.text[:300]
|
||||
assert "/posts/" in r.url, r.url
|
||||
return r.url
|
||||
def _create_project_media(s, attachment_uids, title="Media Project"):
|
||||
r = s.post(
|
||||
f"{BASE_URL}/projects/create",
|
||||
data={
|
||||
"title": title,
|
||||
"description": "Project with media for gallery tests.",
|
||||
"project_type": "software",
|
||||
"platforms": "linux",
|
||||
"status": "In Development",
|
||||
"attachment_uids": attachment_uids,
|
||||
},
|
||||
allow_redirects=True,
|
||||
)
|
||||
assert r.status_code == 200, r.text[:300]
|
||||
assert "/projects/" in r.url, r.url
|
||||
return r.url
|
||||
def _media_json(session, username):
|
||||
r = session.get(f"{BASE_URL}/profile/{username}?tab=media", headers=JSON_media)
|
||||
assert r.status_code == 200, r.text[:300]
|
||||
return r.json()
|
||||
def _media_uids(session, username):
|
||||
return [m["uid"] for m in _media_json(session, username)["media"]]
|
||||
def _seed_media_via_browser(page, title="ui media"):
|
||||
up = page.request.post(
|
||||
f"{BASE_URL}/uploads/upload",
|
||||
multipart={
|
||||
"file": {
|
||||
"name": "pic.png",
|
||||
"mimeType": "image/png",
|
||||
"buffer": _png_bytes_media(),
|
||||
}
|
||||
},
|
||||
)
|
||||
assert up.ok, up.text()
|
||||
uid = up.json()["uid"]
|
||||
post = page.request.post(
|
||||
f"{BASE_URL}/posts/create",
|
||||
form={
|
||||
"content": "media for the ui gallery test",
|
||||
"title": title,
|
||||
"topic": "random",
|
||||
"attachment_uids": uid,
|
||||
},
|
||||
)
|
||||
assert post.ok, post.text()
|
||||
return uid
|
||||
def _media_uids_via_page(page, username):
|
||||
resp = page.request.get(f"{BASE_URL}/profile/{username}?tab=media", headers=JSON_media)
|
||||
return [m["uid"] for m in resp.json()["media"]]
|
||||
|
||||
|
||||
def test_admin_users_loads(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/admin/users", wait_until="domcontentloaded")
|
||||
assert page.is_visible("text=Users")
|
||||
assert page.is_visible("text=alice_test")
|
||||
|
||||
|
||||
def test_admin_users_pagination(alice):
|
||||
page, _ = alice
|
||||
import urllib.request
|
||||
import urllib.parse
|
||||
|
||||
for i in range(30):
|
||||
data = urllib.parse.urlencode(
|
||||
{
|
||||
"username": f"pagu_{i:04d}",
|
||||
"email": f"pagu{i:04d}@test.devplace",
|
||||
"password": "testpass123",
|
||||
"confirm_password": "testpass123",
|
||||
}
|
||||
).encode()
|
||||
try:
|
||||
urllib.request.urlopen(
|
||||
urllib.request.Request(
|
||||
f"{BASE_URL}/auth/signup",
|
||||
data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
page.goto(f"{BASE_URL}/admin/users", wait_until="domcontentloaded")
|
||||
assert "Page 1" in page.text_content(".pagination-info")
|
||||
page.goto(f"{BASE_URL}/admin/users?page=2", wait_until="domcontentloaded")
|
||||
assert "Page 2" in page.text_content(".pagination-info")
|
||||
|
||||
|
||||
def test_admin_change_user_role(alice):
|
||||
page, _ = alice
|
||||
uid = _seed_target_user()
|
||||
page.goto(f"{BASE_URL}/admin/users", wait_until="domcontentloaded")
|
||||
sel = f"form[action='/admin/users/{uid}/role'] select[name='role']"
|
||||
page.locator(sel).wait_for(state="visible")
|
||||
page.select_option(sel, "admin")
|
||||
page.locator(".dialog-overlay.visible .dialog-confirm").click()
|
||||
page.wait_for_url("**/admin/users**", wait_until="domcontentloaded")
|
||||
page.goto(f"{BASE_URL}/admin/users", wait_until="domcontentloaded")
|
||||
assert page.locator(sel).input_value() == "admin"
|
||||
|
||||
|
||||
def test_admin_toggle_user_active(alice):
|
||||
page, _ = alice
|
||||
uid = _seed_target_user()
|
||||
page.goto(f"{BASE_URL}/admin/users", wait_until="domcontentloaded")
|
||||
btn = f"form[action='/admin/users/{uid}/toggle'] button"
|
||||
before = page.locator(btn).inner_text()
|
||||
page.locator(btn).click()
|
||||
page.locator(".dialog-overlay.visible .dialog-confirm").click()
|
||||
page.wait_for_url("**/admin/users**", wait_until="domcontentloaded")
|
||||
assert page.locator(btn).inner_text() != before
|
||||
|
||||
|
||||
def test_admin_password_change_redirect(alice):
|
||||
page, _ = alice
|
||||
uid = _seed_target_user()
|
||||
page.goto(f"{BASE_URL}/admin/users", wait_until="domcontentloaded")
|
||||
form_sel = f"form[action='/admin/users/{uid}/password']"
|
||||
pw_form = page.locator(form_sel)
|
||||
if pw_form.count() > 0:
|
||||
page.locator(f"button[data-toggle='pw-{uid}']").click()
|
||||
pw_form.locator("input[name='password']").fill("newpass123")
|
||||
pw_form.locator("button[type='submit']").click()
|
||||
page.wait_for_url("**/admin/users**", wait_until="domcontentloaded")
|
||||
|
||||
|
||||
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_ui_admin_media_sidebar_and_trash(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/admin/users", wait_until="domcontentloaded")
|
||||
link = page.locator(".sidebar-link[href='/admin/media']")
|
||||
assert link.is_visible()
|
||||
link.click()
|
||||
page.wait_for_url("**/admin/media", wait_until="domcontentloaded")
|
||||
assert page.is_visible("text=Media trash")
|
||||
Reference in New Issue
Block a user