forked from retoor/devplacepy
feat: add test suites for backup schedules, service config, auth token, bookmarks, content permissions, devii adopt, and devrant auth
Add comprehensive test coverage for seven new API areas: - Backup schedules CRUD operations in admin panel - Service configuration saving and retrieval for news AI model - Token authentication with email/username and JSON body support - Bookmarking for project and news targets with invalid type validation - Content permissions testing with member and admin session fixtures - Devii adopt endpoint redirect behavior and session cookie handling - Devrant authentication flow with user registration and token retrieval
This commit is contained in:
@@ -0,0 +1,154 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import time
|
||||
import pytest
|
||||
import requests
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import get_table, refresh_snapshot, set_setting
|
||||
|
||||
JSON = {"Accept": "application/json"}
|
||||
_counter_bs = [0]
|
||||
|
||||
|
||||
@pytest.fixture(scope="module", autouse=True)
|
||||
def _settings(app_server):
|
||||
set_setting("rate_limit_per_minute", "1000000")
|
||||
set_setting("registration_open", "1")
|
||||
yield
|
||||
|
||||
|
||||
def _admin(seeded_db):
|
||||
refresh_snapshot()
|
||||
key = get_table("users").find_one(username="alice_test")["api_key"]
|
||||
s = requests.Session()
|
||||
s.headers.update({"X-API-KEY": key, **JSON})
|
||||
return s
|
||||
|
||||
|
||||
def _member():
|
||||
_counter_bs[0] += 1
|
||||
name = f"bsmem{int(time.time() * 1000)}{_counter_bs[0]}"
|
||||
requests.post(
|
||||
f"{BASE_URL}/auth/signup",
|
||||
data={
|
||||
"username": name,
|
||||
"email": f"{name}@t.dev",
|
||||
"password": "secret123",
|
||||
"confirm_password": "secret123",
|
||||
},
|
||||
allow_redirects=True,
|
||||
)
|
||||
refresh_snapshot()
|
||||
s = requests.Session()
|
||||
s.headers.update(
|
||||
{"X-API-KEY": get_table("users").find_one(username=name)["api_key"], **JSON}
|
||||
)
|
||||
return s
|
||||
|
||||
|
||||
def _name():
|
||||
_counter_bs[0] += 1
|
||||
return f"sched{int(time.time() * 1000)}{_counter_bs[0]}"
|
||||
|
||||
|
||||
def _create(admin, name, **extra):
|
||||
payload = {"name": name, "target": "database", "kind": "interval", "every_seconds": "3600"}
|
||||
payload.update(extra)
|
||||
r = admin.post(f"{BASE_URL}/admin/backups/schedules/create", data=payload)
|
||||
assert r.status_code in (200, 302), r.text[:200]
|
||||
refresh_snapshot()
|
||||
return get_table("backup_schedules").find_one(name=name, deleted_at=None)
|
||||
|
||||
|
||||
def test_create_schedule(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
name = _name()
|
||||
row = _create(admin, name)
|
||||
assert row is not None
|
||||
assert row["target"] == "database"
|
||||
assert row["deleted_at"] is None
|
||||
|
||||
|
||||
def test_create_cron_without_expression_rejected(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
r = admin.post(
|
||||
f"{BASE_URL}/admin/backups/schedules/create",
|
||||
data={"name": _name(), "target": "full", "kind": "cron", "cron": ""},
|
||||
)
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_toggle_schedule(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
row = _create(admin, _name())
|
||||
before = int(row.get("enabled") or 0)
|
||||
r = admin.post(f"{BASE_URL}/admin/backups/schedules/{row['uid']}/toggle")
|
||||
assert r.status_code in (200, 302)
|
||||
refresh_snapshot()
|
||||
after = int(get_table("backup_schedules").find_one(uid=row["uid"])["enabled"] or 0)
|
||||
assert after != before
|
||||
|
||||
|
||||
def test_edit_schedule(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
row = _create(admin, _name())
|
||||
new_name = _name()
|
||||
r = admin.post(
|
||||
f"{BASE_URL}/admin/backups/schedules/{row['uid']}/edit",
|
||||
data={"name": new_name, "target": "uploads", "kind": "interval", "every_seconds": "7200"},
|
||||
)
|
||||
assert r.status_code in (200, 302)
|
||||
refresh_snapshot()
|
||||
updated = get_table("backup_schedules").find_one(uid=row["uid"])
|
||||
assert updated["name"] == new_name
|
||||
assert updated["target"] == "uploads"
|
||||
|
||||
|
||||
def test_run_schedule_enqueues_backup(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
row = _create(admin, _name())
|
||||
r = admin.post(f"{BASE_URL}/admin/backups/schedules/{row['uid']}/run")
|
||||
assert r.status_code in (200, 302)
|
||||
refresh_snapshot()
|
||||
assert get_table("backups").count(schedule_uid=row["uid"]) >= 1
|
||||
|
||||
|
||||
def test_delete_schedule(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
row = _create(admin, _name())
|
||||
r = admin.post(f"{BASE_URL}/admin/backups/schedules/{row['uid']}/delete")
|
||||
assert r.status_code in (200, 302)
|
||||
refresh_snapshot()
|
||||
assert get_table("backup_schedules").find_one(uid=row["uid"], deleted_at=None) is None
|
||||
|
||||
|
||||
def test_unknown_schedule_404(seeded_db):
|
||||
admin = _admin(seeded_db)
|
||||
for path in ("edit", "toggle", "run", "delete"):
|
||||
if path == "edit":
|
||||
r = admin.post(
|
||||
f"{BASE_URL}/admin/backups/schedules/nope-uid/edit",
|
||||
data={"name": "x", "target": "full", "kind": "interval"},
|
||||
)
|
||||
else:
|
||||
r = admin.post(f"{BASE_URL}/admin/backups/schedules/nope-uid/{path}")
|
||||
assert r.status_code == 404, f"{path}: {r.status_code}"
|
||||
|
||||
|
||||
def test_member_cannot_create_schedule(app_server):
|
||||
member = _member()
|
||||
r = member.post(
|
||||
f"{BASE_URL}/admin/backups/schedules/create",
|
||||
data={"name": _name(), "target": "full", "kind": "interval"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert r.status_code in (302, 303, 403)
|
||||
|
||||
|
||||
def test_guest_cannot_create_schedule(app_server):
|
||||
r = requests.post(
|
||||
f"{BASE_URL}/admin/backups/schedules/create",
|
||||
data={"name": _name(), "target": "full", "kind": "interval"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert r.status_code in (302, 303, 401)
|
||||
Reference in New Issue
Block a user