fc3g: integration tests for /api/extension/* and /extension/<filename>
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,277 @@
|
||||
"""FC-3g: /api/extension and /extension/<filename> 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/<filename> -----------------------------------------
|
||||
|
||||
|
||||
@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
|
||||
Reference in New Issue
Block a user