refactor: rewrite from Ruby/Sinatra to Python/FastAPI
This commit is contained in:
@@ -0,0 +1,423 @@
|
||||
"""Tests for the admin routes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from volumen.users import Users
|
||||
|
||||
|
||||
def _csrf_from_body(body: str) -> str:
|
||||
match = re.search(r'name="_csrf" value="([^"]+)"', body)
|
||||
if match:
|
||||
return match.group(1)
|
||||
return ""
|
||||
|
||||
|
||||
class TestAdminLogin:
|
||||
def test_admin_redirect_to_login(self, admin_client: TestClient) -> None:
|
||||
response = admin_client.get("/admin/", follow_redirects=False)
|
||||
assert response.status_code in (302, 307, 301, 303)
|
||||
|
||||
def test_login_page_renders(self, admin_client: TestClient) -> None:
|
||||
response = admin_client.get("/admin/login")
|
||||
assert response.status_code == 200
|
||||
assert "Sign in" in response.text
|
||||
|
||||
def test_login_with_valid_credentials(self, admin_client: TestClient) -> None:
|
||||
resp = admin_client.get("/admin/login")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/login",
|
||||
data={"username": "admin", "password": "secret", "_csrf": token},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert resp.status_code in (302, 307, 303)
|
||||
|
||||
def test_login_with_invalid_credentials(self, admin_client: TestClient) -> None:
|
||||
resp = admin_client.get("/admin/login")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/login",
|
||||
data={"username": "admin", "password": "nope", "_csrf": token},
|
||||
)
|
||||
assert "Invalid" in resp.text or "invalid" in resp.text.lower()
|
||||
|
||||
def test_wrong_password_is_rejected(self, admin_client: TestClient) -> None:
|
||||
resp = admin_client.get("/admin/login")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/login",
|
||||
data={"username": "admin", "password": "nope", "_csrf": token},
|
||||
)
|
||||
assert "Invalid username or password" in resp.text
|
||||
|
||||
|
||||
def _login(client: TestClient) -> str:
|
||||
"""Log in as admin and return the session cookie value."""
|
||||
resp = client.get("/admin/login")
|
||||
token = _csrf_from_body(resp.text)
|
||||
client.post(
|
||||
"/admin/login",
|
||||
data={"username": "admin", "password": "secret", "_csrf": token},
|
||||
follow_redirects=True,
|
||||
)
|
||||
return token
|
||||
|
||||
|
||||
def _login_as(client: TestClient, username: str, password: str) -> str:
|
||||
resp = client.get("/admin/login")
|
||||
token = _csrf_from_body(resp.text)
|
||||
client.post(
|
||||
"/admin/login",
|
||||
data={"username": username, "password": password, "_csrf": token},
|
||||
follow_redirects=True,
|
||||
)
|
||||
return token
|
||||
|
||||
|
||||
class TestAdminPostList:
|
||||
def test_post_list_requires_login(self, admin_client: TestClient) -> None:
|
||||
resp = admin_client.get("/admin/", follow_redirects=False)
|
||||
assert resp.status_code in (302, 307, 303)
|
||||
|
||||
def test_post_list_after_login(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/")
|
||||
assert resp.status_code == 200
|
||||
assert "Posts" in resp.text
|
||||
|
||||
|
||||
class TestCreatePost:
|
||||
def test_create_post_writes_file(
|
||||
self, admin_client: TestClient, admin_config_dir: Path
|
||||
) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/posts/new")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/posts",
|
||||
data={
|
||||
"_csrf": token,
|
||||
"title": "Test",
|
||||
"slug": "test",
|
||||
"lang": "en",
|
||||
"tags": "a, b",
|
||||
"body": "Hello **world**.",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert resp.status_code in (302, 307, 303)
|
||||
post_path = admin_config_dir / "posts" / "test.md"
|
||||
assert post_path.exists()
|
||||
content = post_path.read_text()
|
||||
assert "Test" in content
|
||||
assert "a, b" in content or "a" in content
|
||||
|
||||
def test_create_post_with_all_langs(
|
||||
self, admin_client: TestClient, admin_config_dir: Path
|
||||
) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/posts/new")
|
||||
token = _csrf_from_body(resp.text)
|
||||
admin_client.post(
|
||||
"/admin/posts",
|
||||
data={
|
||||
"_csrf": token,
|
||||
"title": "Global",
|
||||
"slug": "global",
|
||||
"lang": "en",
|
||||
"all_langs": "on",
|
||||
"body": "Hi",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
post_path = admin_config_dir / "posts" / "global.md"
|
||||
assert post_path.exists()
|
||||
content = post_path.read_text()
|
||||
assert "all_langs" in content
|
||||
|
||||
def test_create_post_with_cover(self, admin_client: TestClient, admin_config_dir: Path) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/posts/new")
|
||||
token = _csrf_from_body(resp.text)
|
||||
admin_client.post(
|
||||
"/admin/posts",
|
||||
data={
|
||||
"_csrf": token,
|
||||
"title": "Covered",
|
||||
"slug": "covered",
|
||||
"lang": "en",
|
||||
"cover": "/media/c.png",
|
||||
"body": "Hi",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
post_path = admin_config_dir / "posts" / "covered.md"
|
||||
content = post_path.read_text()
|
||||
assert "/media/c.png" in content
|
||||
|
||||
def test_create_post_rejects_invalid_slug(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/posts/new")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/posts",
|
||||
data={
|
||||
"_csrf": token,
|
||||
"title": "X",
|
||||
"slug": "../../etc/passwd",
|
||||
"lang": "en",
|
||||
"body": "y",
|
||||
},
|
||||
)
|
||||
assert "Invalid slug" in resp.text
|
||||
|
||||
def test_create_post_rejects_xss_in_slug(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/posts/new")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/posts",
|
||||
data={
|
||||
"_csrf": token,
|
||||
"title": "X",
|
||||
"slug": 'foo" onclick',
|
||||
"lang": "en",
|
||||
"body": "y",
|
||||
},
|
||||
)
|
||||
assert "Invalid slug" in resp.text
|
||||
|
||||
def test_create_post_duplicate_slug(
|
||||
self, admin_client: TestClient, admin_config_dir: Path
|
||||
) -> None:
|
||||
(admin_config_dir / "posts" / "dup.md").write_text(
|
||||
'+++\ntitle = "Existing"\n+++\n\nBody.\n'
|
||||
)
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/posts/new")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/posts",
|
||||
data={
|
||||
"_csrf": token,
|
||||
"title": "Dup",
|
||||
"slug": "dup",
|
||||
"lang": "en",
|
||||
"body": "x",
|
||||
},
|
||||
)
|
||||
assert "already exists" in resp.text.lower()
|
||||
|
||||
|
||||
class TestCSRF:
|
||||
def test_csrf_token_required(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.post(
|
||||
"/admin/posts",
|
||||
data={"title": "X", "slug": "x", "body": "y"},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_invalid_csrf_token_rejected(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/posts/new")
|
||||
resp = admin_client.post(
|
||||
"/admin/posts",
|
||||
data={
|
||||
"_csrf": "invalid-token",
|
||||
"title": "X",
|
||||
"slug": "x",
|
||||
"body": "y",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
class TestLogout:
|
||||
def test_logout_clears_session(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/")
|
||||
assert resp.status_code == 200
|
||||
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
admin_client.post("/admin/logout", data={"_csrf": token})
|
||||
resp = admin_client.get("/admin/", follow_redirects=False)
|
||||
assert resp.status_code in (302, 307, 303)
|
||||
|
||||
|
||||
class TestSettings:
|
||||
def test_settings_requires_login(self, admin_client: TestClient) -> None:
|
||||
resp = admin_client.get("/admin/settings", follow_redirects=False)
|
||||
assert resp.status_code in (302, 307, 303)
|
||||
|
||||
def test_change_password(self, admin_client: TestClient, admin_config_dir: Path) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/settings/password",
|
||||
data={
|
||||
"_csrf": token,
|
||||
"current_password": "secret",
|
||||
"new_password": "brand-new",
|
||||
},
|
||||
)
|
||||
assert "Password updated" in resp.text
|
||||
|
||||
users_path = str(admin_config_dir / "users.toml")
|
||||
fresh = Users(path=users_path)
|
||||
assert fresh.authenticate("admin", "secret") is None
|
||||
assert fresh.authenticate("admin", "brand-new") is not None
|
||||
|
||||
def test_change_password_wrong_current(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/settings/password",
|
||||
data={
|
||||
"_csrf": token,
|
||||
"current_password": "wrong",
|
||||
"new_password": "brand-new",
|
||||
},
|
||||
)
|
||||
assert "incorrect" in resp.text.lower()
|
||||
|
||||
def test_change_username(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/settings/username",
|
||||
data={"_csrf": token, "username": "superadmin"},
|
||||
)
|
||||
assert "updated" in resp.text.lower()
|
||||
|
||||
def test_change_username_accepts_same_name(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/settings/username",
|
||||
data={"_csrf": token, "username": "admin"},
|
||||
)
|
||||
assert "unchanged" in resp.text.lower()
|
||||
|
||||
def test_change_username_rejects_special_chars(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/settings/username",
|
||||
data={"_csrf": token, "username": 'foo" bar'},
|
||||
)
|
||||
assert "letters" in resp.text.lower()
|
||||
|
||||
def test_change_display_name(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/settings/name",
|
||||
data={"_csrf": token, "name": "Admin User"},
|
||||
)
|
||||
assert "Display name updated" in resp.text
|
||||
|
||||
|
||||
class TestFediverseSettings:
|
||||
def test_change_fediverse_creator_valid(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/settings/fediverse",
|
||||
data={"_csrf": token, "fediverse_creator": "@user@example.com"},
|
||||
)
|
||||
assert "updated" in resp.text.lower() or "updated" in resp.text.lower()
|
||||
|
||||
def test_change_fediverse_creator_invalid(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/settings/fediverse",
|
||||
data={"_csrf": token, "fediverse_creator": "not-a-handle"},
|
||||
)
|
||||
assert "handle" in resp.text.lower()
|
||||
|
||||
|
||||
class TestUserManagement:
|
||||
def test_create_user_admin_only(self, admin_client: TestClient, admin_config_dir: Path) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/settings/users",
|
||||
data={"_csrf": token, "username": "editor", "password": "pw12345", "role": "author"},
|
||||
)
|
||||
assert "User added" in resp.text or "added" in resp.text.lower()
|
||||
|
||||
users_path = str(admin_config_dir / "users.toml")
|
||||
users = Users(path=users_path)
|
||||
assert users.find("editor") is not None
|
||||
|
||||
def test_create_user_then_login(self, admin_client: TestClient, admin_config_dir: Path) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
admin_client.post(
|
||||
"/admin/settings/users",
|
||||
data={"_csrf": token, "username": "editor2", "password": "pw12345"},
|
||||
)
|
||||
|
||||
# Logout
|
||||
resp = admin_client.get("/admin/login")
|
||||
token = _csrf_from_body(resp.text)
|
||||
admin_client.post("/admin/logout", data={"_csrf": token})
|
||||
|
||||
# Login as new user
|
||||
_login_as(admin_client, "editor2", "pw12345")
|
||||
resp = admin_client.get("/admin/")
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_cannot_delete_own_account(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/settings/users/admin/delete",
|
||||
data={"_csrf": token},
|
||||
)
|
||||
assert "cannot delete your own" in resp.text.lower()
|
||||
|
||||
def test_cannot_create_user_with_empty_fields(self, admin_client: TestClient) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/settings")
|
||||
token = _csrf_from_body(resp.text)
|
||||
resp = admin_client.post(
|
||||
"/admin/settings/users",
|
||||
data={"_csrf": token, "username": "", "password": ""},
|
||||
)
|
||||
assert "required" in resp.text.lower() or "empty" in resp.text.lower()
|
||||
|
||||
|
||||
class TestOrphanedSession:
|
||||
def test_orphaned_session_is_rejected(
|
||||
self, admin_client: TestClient, admin_config_dir: Path
|
||||
) -> None:
|
||||
_login(admin_client)
|
||||
resp = admin_client.get("/admin/")
|
||||
assert resp.status_code == 200
|
||||
|
||||
# Clear users file to simulate deleted user
|
||||
users_path = admin_config_dir / "users.toml"
|
||||
users_path.write_text("")
|
||||
|
||||
resp = admin_client.get("/admin/", follow_redirects=False)
|
||||
assert resp.status_code in (302, 307, 303)
|
||||
Reference in New Issue
Block a user