volumen/tests/test_admin.py

"""Tests for the admin routes."""

from __future__ import annotations

import os
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_page_omits_warning_when_users_exist(self, admin_client: TestClient) -> None:
        # Regression: `users_exist` was missing from the admin template
        # context, so the "No users configured" warning was shown on every
        # login page render regardless of whether authentication worked.
        response = admin_client.get("/admin/login")
        assert response.status_code == 200
        assert "No users configured" not in response.text

    def test_login_page_shows_warning_when_no_users(self, config, store) -> None:
        # No bootstrap_hash, no users.toml → `users_exist` must be False
        # so the warning block is rendered.
        from volumen.app import create_app

        users_path = config.users_file
        # Make sure no stale users file exists.
        if os.path.exists(users_path):
            os.remove(users_path)
        app = create_app(config, store)
        client = TestClient(app)
        response = client.get("/admin/login")
        assert response.status_code == 200
        assert "No users configured" 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": "volumen-test-password",
                "_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": "volumen-test-password",
            "_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": "volumen-test-password",
                "new_password": "volumen-brand-new-pw",
            },
        )
        assert "Password updated" in resp.text

        users_path = str(admin_config_dir / "users.toml")
        fresh = Users(path=users_path)
        assert fresh.authenticate("admin", "volumen-test-password") is None
        assert fresh.authenticate("admin", "volumen-brand-new-pw") 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()

    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": "editor-test-pw",
                "role": "author",
            },
        )
        assert "User added" in resp.text

        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": "editor-test-pw"},
        )

        # 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", "editor-test-pw")
        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:
        import tomli_w

        from volumen.password import hash_password as hash_pw

        _login(admin_client)
        resp = admin_client.get("/admin/")
        assert resp.status_code == 200

        # Replace users file with a different admin so the logged-in
        # 'admin' is truly orphaned. An empty file would resurrect the
        # bootstrap admin via the password_hash fallback, so we must
        # write a real users list that does not include 'admin'.
        users_path = admin_config_dir / "users.toml"
        payload = tomli_w.dumps(
            {
                "users": [
                    {
                        "username": "other",
                        "password_hash": hash_pw("other-password"),
                        "role": "admin",
                    }
                ]
            }
        ).encode("utf-8")
        users_path.write_bytes(payload)

        resp = admin_client.get("/admin/", follow_redirects=False)
        assert resp.status_code in (302, 307, 303)