volumen/tests/test_server.py

"""Tests for the public API endpoints."""

from __future__ import annotations

from datetime import date
from pathlib import Path

from fastapi.testclient import TestClient


class TestSiteEndpoint:
    def test_site_endpoint(self, client: TestClient) -> None:
        response = client.get("/api/volumen/site")
        assert response.status_code == 200
        data = response.json()
        assert data["title"] == "My Blog"

    def test_site_endpoint_exposes_fediverse_creator_default(self, client: TestClient) -> None:
        response = client.get("/api/volumen/site")
        assert response.status_code == 200
        assert "fediverse_creator" in response.json()


class TestPostsEndpoint:
    def test_posts_excludes_drafts(self, client: TestClient) -> None:
        response = client.get("/api/volumen/posts")
        data = response.json()
        assert data["total"] == 1
        assert data["posts"][0]["slug"] == "hello"

    def test_post_detail_renders_html(self, client: TestClient) -> None:
        response = client.get("/api/volumen/posts/hello")
        assert response.status_code == 200
        assert "<strong>world</strong>" in response.json()["html"]

    def test_post_not_found(self, client: TestClient) -> None:
        response = client.get("/api/volumen/posts/missing")
        assert response.status_code == 404
        assert response.json()["detail"]["error"] == "not_found"

    def test_draft_post_detail_returns_error(self, client: TestClient) -> None:
        response = client.get("/api/volumen/posts/draft")
        assert response.status_code == 404
        assert response.json()["detail"]["error"] == "draft"


class TestTagsEndpoint:
    def test_tags_endpoint(self, client: TestClient) -> None:
        response = client.get("/api/volumen/tags")
        data = response.json()
        assert data["tags"] == [{"name": "intro", "count": 1}]


class TestCORS:
    def test_cors_header_present(self, client: TestClient) -> None:
        response = client.get("/api/volumen/site")
        assert response.headers.get("access-control-allow-origin") == "*"

    def test_cors_preflight(self, client: TestClient) -> None:
        response = client.options("/api/volumen/site")
        assert response.status_code == 200
        assert response.headers["access-control-allow-origin"] == "*"
        assert "GET" in response.headers["access-control-allow-methods"]


class TestLanguageFilter:
    def test_all_langs_post_appears_in_other_languages(
        self, client: TestClient, tmp_content_dir: Path
    ) -> None:
        (tmp_content_dir / "global.md").write_text(
            '+++\ntitle = "Global"\nlang = "en"\nall_langs = true\n+++\n\nEverywhere.\n'
        )
        response = client.get("/api/volumen/posts?lang=cs")
        slugs = [p["slug"] for p in response.json()["posts"]]
        assert "global" in slugs
        assert "hello" not in slugs

    def test_lang_filter_default_shows_default_lang_posts(self, client: TestClient) -> None:
        response = client.get("/api/volumen/posts")
        slugs = [p["slug"] for p in response.json()["posts"]]
        assert "hello" in slugs


class TestScheduledPosts:
    def test_scheduled_post_is_hidden(self, client: TestClient, tmp_content_dir: Path) -> None:
        future = date.today().replace(year=date.today().year + 1).isoformat()
        (tmp_content_dir / "later.md").write_text(
            f'+++\ntitle = "Later"\npublish_at = {future}\n+++\n\nNot yet.\n'
        )
        response = client.get("/api/volumen/posts")
        slugs = [p["slug"] for p in response.json()["posts"]]
        assert "later" not in slugs


class TestSearch:
    def test_search_by_title(self, client: TestClient) -> None:
        response = client.get("/api/volumen/posts?q=Hello")
        data = response.json()
        assert data["total"] == 1
        assert data["posts"][0]["slug"] == "hello"

    def test_search_by_body(self, client: TestClient) -> None:
        response = client.get("/api/volumen/posts?q=world")
        data = response.json()
        assert data["total"] == 1

    def test_search_no_match(self, client: TestClient) -> None:
        response = client.get("/api/volumen/posts?q=nonexistent")
        assert response.json()["total"] == 0

    def test_search_is_case_insensitive(self, client: TestClient) -> None:
        response = client.get("/api/volumen/posts?q=HELLO")
        assert response.json()["total"] == 1


class TestPagination:
    def test_pagination_has_next_and_prev(self, client: TestClient, tmp_content_dir: Path) -> None:
        for i in range(5):
            (tmp_content_dir / f"post-{i}.md").write_text(
                f'+++\ntitle = "Post {i}"\n+++\n\nBody {i}.\n'
            )

        resp = client.get("/api/volumen/posts?page=1&limit=2")
        data = resp.json()
        assert data["has_next"] is True
        assert data["has_prev"] is False

        resp = client.get("/api/volumen/posts?page=2&limit=2")
        data = resp.json()
        assert data["has_next"] is True
        assert data["has_prev"] is True

        resp = client.get("/api/volumen/posts?page=3&limit=2")
        data = resp.json()
        assert data["has_next"] is False
        assert data["has_prev"] is True


class TestRSSFeed:
    def test_feed_endpoint(self, client: TestClient) -> None:
        response = client.get("/api/volumen/feed.xml")
        assert response.status_code == 200
        ct = response.headers.get("content-type", "")
        assert "application/rss+xml" in ct or "application/xml" in ct
        body = response.text
        assert '<rss version="2.0"' in body
        assert "<title>Hello</title>" in body
        assert "Draft" not in body
        assert "<language>en</language>" in body

    def test_rss_feed_includes_dc_creator_when_set(
        self, client: TestClient, tmp_content_dir: Path
    ) -> None:
        (tmp_content_dir / "rss.md").write_text(
            '+++\ntitle = "Rss"\nslug = "rss"\n'
            'fediverse_creator = "@rss@example.com"\n+++\n\nBody.\n'
        )
        response = client.get("/api/volumen/feed.xml")
        assert "<dc:creator>@rss@example.com</dc:creator>" in response.text
        assert "xmlns:dc" in response.text

    def test_rss_feed_omits_dc_creator_when_not_set(self, client: TestClient) -> None:
        response = client.get("/api/volumen/feed.xml")
        assert "<dc:creator>" not in response.text


class TestJSONFeed:
    def test_json_feed_endpoint(self, client: TestClient) -> None:
        response = client.get("/api/volumen/feed.json")
        assert response.status_code == 200
        data = response.json()
        assert data["version"] == "https://jsonfeed.org/version/1.1"
        assert data["title"] == "My Blog"
        assert len(data["items"]) == 1
        assert data["items"][0]["title"] == "Hello"
        assert "<strong>world</strong>" in data["items"][0]["content_html"]
        titles = [i["title"] for i in data["items"]]
        assert "Draft" not in titles

    def test_json_feed_includes_authors_from_post(
        self, client: TestClient, tmp_content_dir: Path
    ) -> None:
        (tmp_content_dir / "jsonfed.md").write_text(
            '+++\ntitle = "Jsonfed"\nslug = "jsonfed"\n'
            'fediverse_creator = "@jf@example.io"\n+++\n\nBody.\n'
        )
        response = client.get("/api/volumen/feed.json")
        items = response.json()["items"]
        item = next(i for i in items if i.get("id", "").endswith("jsonfed"))
        assert item["authors"] == [{"name": "@jf@example.io"}]

    def test_json_feed_falls_back_to_site_default_fediverse_creator(
        self, tmp_path: Path, tmp_content_dir: Path
    ) -> None:
        """When a post has no fediverse_creator, fall back to site.fediverse_creator."""
        from fastapi.testclient import TestClient

        from volumen.app import create_app
        from volumen.config import Config
        from volumen.store import Store

        config_path = tmp_path / "custom.toml"
        users_path = tmp_path / "users.toml"
        users_path.write_text("")
        config_path.write_text(
            f'content_dir = "{tmp_content_dir}"\n'
            f'users_file = "{users_path}"\n'
            "\n[site]\n"
            'fediverse_creator = "@siteowner@example.social"\n'
        )
        custom_config = Config.load(path=str(config_path))
        custom_store = Store(str(tmp_content_dir), default_lang="en")
        custom_app = create_app(custom_config, custom_store)
        custom_client = TestClient(custom_app)

        response = custom_client.get("/api/volumen/feed.json")
        assert response.status_code == 200
        item = response.json()["items"][0]
        assert "authors" in item
        assert item["authors"] == [{"name": "@siteowner@example.social"}]


class TestSitemap:
    def test_sitemap_endpoint(self, client: TestClient) -> None:
        response = client.get("/api/volumen/sitemap.xml")
        assert response.status_code == 200
        ct = response.headers.get("content-type", "")
        assert "application/xml" in ct or "text/xml" in ct
        body = response.text
        assert "urlset" in body
        assert "<loc>https://example.com/hello</loc>" in body
        assert "draft" not in body


class TestFediverseCreator:
    def test_post_summary_exposes_fediverse_creator(
        self, client: TestClient, tmp_content_dir: Path
    ) -> None:
        (tmp_content_dir / "attributed.md").write_text(
            '+++\ntitle = "Attributed"\nslug = "attributed"\n'
            'fediverse_creator = "@ada@lovelace.org"\n+++\n\nBody.\n'
        )
        response = client.get("/api/volumen/posts/attributed")
        assert response.status_code == 200
        assert response.json()["fediverse_creator"] == "@ada@lovelace.org"

    def test_post_summary_fediverse_creator_is_null_when_not_set(self, client: TestClient) -> None:
        response = client.get("/api/volumen/posts/hello")
        assert response.status_code == 200
        assert response.json()["fediverse_creator"] is None

    def test_posts_list_summary_exposes_fediverse_creator(
        self, client: TestClient, tmp_content_dir: Path
    ) -> None:
        (tmp_content_dir / "fed.md").write_text(
            '+++\ntitle = "Fed"\nslug = "fed"\n'
            'fediverse_creator = "@user@example.social"\n+++\n\nBody.\n'
        )
        response = client.get("/api/volumen/posts")
        fed = next(p for p in response.json()["posts"] if p["slug"] == "fed")
        assert fed["fediverse_creator"] == "@user@example.social"


class TestCacheHeaders:
    def test_cache_control_on_api_responses(self, client: TestClient) -> None:
        response = client.get("/api/volumen/site")
        assert "cache-control" in response.headers