From cd838ec904009a3fa8ad4d718e8404453487d49e Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 23 May 2026 23:40:00 -0400 Subject: [PATCH] fc3g: integration tests for /api/extension/* and /extension/ Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_api_extension.py | 277 ++++++++++++++++++++++++++++++++++++ 1 file changed, 277 insertions(+) create mode 100644 tests/test_api_extension.py diff --git a/tests/test_api_extension.py b/tests/test_api_extension.py new file mode 100644 index 0000000..ca3671f --- /dev/null +++ b/tests/test_api_extension.py @@ -0,0 +1,277 @@ +"""FC-3g: /api/extension and /extension/ integration tests.""" + +import hashlib + +import pytest +import pytest_asyncio +from sqlalchemy import func, select + +from backend.app import create_app +from backend.app import frontend as frontend_module +from backend.app.api import extension as extension_module +from backend.app.models import AppSetting, Artist, Source + +pytestmark = pytest.mark.integration + + +@pytest.fixture +async def app(): + return create_app() + + +@pytest.fixture +async def client(app): + async with app.test_client() as c: + yield c + + +@pytest_asyncio.fixture +async def ext_key(db): + db.add(AppSetting(key="extension_api_key", value="test-ext-key")) + await db.commit() + return "test-ext-key" + + +# --- /api/extension/quick-add-source --------------------------------- + + +@pytest.mark.asyncio +async def test_quick_add_source_creates_artist_and_source(client, ext_key, db_sync): + resp = await client.post( + "/api/extension/quick-add-source", + json={"url": "https://www.patreon.com/maewix"}, + headers={"X-Extension-Key": ext_key}, + ) + assert resp.status_code == 201 + body = await resp.get_json() + assert body["created_artist"] is True + assert body["created_source"] is True + assert body["artist"]["slug"] == "maewix" + assert body["source"]["platform"] == "patreon" + assert body["source"]["enabled"] is True + + # Async Core-DML assertion: column select, not ORM attribute access. + artist_count = db_sync.execute( + select(func.count(Artist.id)).where(Artist.slug == "maewix") + ).scalar_one() + assert artist_count == 1 + source_count = db_sync.execute( + select(func.count(Source.id)).where( + Source.platform == "patreon", + Source.url == "https://www.patreon.com/maewix", + ) + ).scalar_one() + assert source_count == 1 + + +@pytest.mark.asyncio +async def test_quick_add_source_idempotent(client, ext_key): + body = {"url": "https://www.subscribestar.com/some-creator"} + headers = {"X-Extension-Key": ext_key} + r1 = await client.post("/api/extension/quick-add-source", json=body, headers=headers) + r2 = await client.post("/api/extension/quick-add-source", json=body, headers=headers) + assert r1.status_code == 201 + assert r2.status_code == 200 + body2 = await r2.get_json() + assert body2["created_source"] is False + assert body2["created_artist"] is False + + +@pytest.mark.parametrize("url,platform,slug", [ + ("https://www.patreon.com/maewix", "patreon", "maewix"), + ("https://patreon.com/maewix", "patreon", "maewix"), + ("https://www.subscribestar.com/foobar", "subscribestar", "foobar"), + ("https://subscribestar.adult/foobar", "subscribestar", "foobar"), + ("https://www.hentai-foundry.com/user/Foo/profile", "hentaifoundry", "Foo"), + ("https://www.deviantart.com/baz", "deviantart", "baz"), + ("https://www.pixiv.net/users/12345", "pixiv", "12345"), + ("https://www.pixiv.net/en/users/12345", "pixiv", "12345"), +]) +@pytest.mark.asyncio +async def test_quick_add_source_url_patterns(client, ext_key, url, platform, slug): + resp = await client.post( + "/api/extension/quick-add-source", + json={"url": url}, + headers={"X-Extension-Key": ext_key}, + ) + assert resp.status_code == 201, await resp.get_json() + body = await resp.get_json() + assert body["source"]["platform"] == platform + # slugify lowercases — the Artist slug should reflect that. + assert body["artist"]["slug"] == slug.lower() + + +@pytest.mark.asyncio +async def test_quick_add_source_unknown_url_400(client, ext_key): + resp = await client.post( + "/api/extension/quick-add-source", + json={"url": "https://example.com/foo"}, + headers={"X-Extension-Key": ext_key}, + ) + assert resp.status_code == 400 + body = await resp.get_json() + assert body["error"] == "unknown_platform" + assert "known" in body + + +@pytest.mark.asyncio +async def test_quick_add_source_invalid_url_400(client, ext_key): + resp = await client.post( + "/api/extension/quick-add-source", + json={"url": "not-a-url"}, + headers={"X-Extension-Key": ext_key}, + ) + assert resp.status_code == 400 + body = await resp.get_json() + assert body["error"] == "invalid_url" + + +@pytest.mark.asyncio +async def test_quick_add_source_missing_key_401(client): + resp = await client.post( + "/api/extension/quick-add-source", + json={"url": "https://www.patreon.com/maewix"}, + ) + assert resp.status_code == 401 + body = await resp.get_json() + assert body["error"] == "unauthorized" + + +@pytest.mark.asyncio +async def test_quick_add_source_wrong_key_401(client, ext_key): + resp = await client.post( + "/api/extension/quick-add-source", + json={"url": "https://www.patreon.com/maewix"}, + headers={"X-Extension-Key": "wrong-key"}, + ) + assert resp.status_code == 401 + + +@pytest.mark.asyncio +async def test_quick_add_source_missing_body_400(client, ext_key): + resp = await client.post( + "/api/extension/quick-add-source", + headers={"X-Extension-Key": ext_key}, + ) + assert resp.status_code == 400 + body = await resp.get_json() + assert body["error"] == "invalid_body" + + +@pytest.mark.asyncio +async def test_quick_add_source_attaches_to_existing_artist(client, ext_key, db, db_sync): + db.add(Artist(name="Maewix Original", slug="maewix", is_subscription=False)) + await db.commit() + + resp = await client.post( + "/api/extension/quick-add-source", + json={"url": "https://www.patreon.com/maewix"}, + headers={"X-Extension-Key": ext_key}, + ) + assert resp.status_code == 201 + body = await resp.get_json() + assert body["created_artist"] is False + assert body["created_source"] is True + + artist_count = db_sync.execute( + select(func.count(Artist.id)).where(Artist.slug == "maewix") + ).scalar_one() + assert artist_count == 1 # no duplicate created + + +# --- /api/extension/manifest --------------------------------------- + + +@pytest.mark.asyncio +async def test_extension_manifest_returns_404_when_dir_missing(client, monkeypatch, tmp_path): + monkeypatch.setattr(extension_module, "XPI_DIR", tmp_path / "does-not-exist") + resp = await client.get("/api/extension/manifest") + assert resp.status_code == 404 + body = await resp.get_json() + assert body == {"installed": False} + + +@pytest.mark.asyncio +async def test_extension_manifest_returns_404_when_no_xpi_files(client, monkeypatch, tmp_path): + monkeypatch.setattr(extension_module, "XPI_DIR", tmp_path) + resp = await client.get("/api/extension/manifest") + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_extension_manifest_returns_metadata_when_xpi_present(client, monkeypatch, tmp_path): + xpi = tmp_path / "fabledcurator-1.2.3.xpi" + xpi.write_bytes(b"fake-xpi-content") + monkeypatch.setattr(extension_module, "XPI_DIR", tmp_path) + resp = await client.get("/api/extension/manifest") + assert resp.status_code == 200 + body = await resp.get_json() + assert body["installed"] is True + assert body["version"] == "1.2.3" + assert body["xpi_url"] == "/extension/fabledcurator-1.2.3.xpi" + assert body["latest_url"] == "/extension/fabledcurator-latest.xpi" + assert body["sha256"] == hashlib.sha256(b"fake-xpi-content").hexdigest() + + +# --- /extension/ ----------------------------------------- + + +@pytest.mark.asyncio +async def test_serve_extension_rejects_non_xpi_filename(client, monkeypatch, tmp_path): + monkeypatch.setattr(frontend_module, "XPI_DIR", tmp_path) + resp = await client.get("/extension/passwd") + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_serve_extension_rejects_path_traversal(client, monkeypatch, tmp_path): + monkeypatch.setattr(frontend_module, "XPI_DIR", tmp_path) + # Path-traversal attempt — the route regex alone catches this since + # `..` characters aren't in [\w.-]+, but cover the case anyway. + resp = await client.get("/extension/fabledcurator-..%2Fetc%2Fpasswd.xpi") + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_serve_extension_404_when_xpi_missing(client, monkeypatch, tmp_path): + monkeypatch.setattr(frontend_module, "XPI_DIR", tmp_path) + resp = await client.get("/extension/fabledcurator-1.0.0.xpi") + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_serve_extension_serves_specific_xpi_with_correct_mime( + client, monkeypatch, tmp_path, +): + xpi = tmp_path / "fabledcurator-1.0.0.xpi" + xpi.write_bytes(b"xpi-bytes") + monkeypatch.setattr(frontend_module, "XPI_DIR", tmp_path) + resp = await client.get("/extension/fabledcurator-1.0.0.xpi") + assert resp.status_code == 200 + assert resp.headers["Content-Type"].startswith("application/x-xpinstall") + + +@pytest.mark.asyncio +async def test_serve_extension_latest_returns_most_recent_xpi( + client, monkeypatch, tmp_path, +): + import os + import time + + older = tmp_path / "fabledcurator-1.0.0.xpi" + newer = tmp_path / "fabledcurator-1.0.1.xpi" + older.write_bytes(b"old") + newer.write_bytes(b"new") + os.utime(older, (time.time() - 10, time.time() - 10)) + monkeypatch.setattr(frontend_module, "XPI_DIR", tmp_path) + resp = await client.get("/extension/fabledcurator-latest.xpi") + assert resp.status_code == 200 + data = await resp.get_data() + assert data == b"new" + + +@pytest.mark.asyncio +async def test_serve_extension_latest_404_when_dir_empty(client, monkeypatch, tmp_path): + monkeypatch.setattr(frontend_module, "XPI_DIR", tmp_path) + resp = await client.get("/extension/fabledcurator-latest.xpi") + assert resp.status_code == 404