FabledCurator/tests/test_api_extension.py

"""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 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_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.asyncio
async def test_quick_add_reuses_source_artist_after_rename(client, ext_key):
    # #130 identity-by-source: after renaming the artist (slug frozen ≠ new
    # name-slug), re-adding the SAME source must reuse it — a slug-based lookup
    # would miss and duplicate the artist.
    url = "https://www.subscribestar.com/renamed-creator"
    h = {"X-Extension-Key": ext_key}
    a1 = (await (await client.post(
        "/api/extension/quick-add-source", json={"url": url}, headers=h
    )).get_json())["artist"]
    await client.patch(f"/api/artists/{a1['id']}", json={"name": "Totally Different"})
    b2 = await (await client.post(
        "/api/extension/quick-add-source", json={"url": url}, headers=h
    )).get_json()
    assert b2["created_artist"] is False
    assert b2["artist"]["id"] == a1["id"]
    assert b2["artist"]["name"] == "Totally Different"


@pytest.mark.asyncio
async def test_resolve_artist_name_dispatches_per_platform(db, monkeypatch):
    # #130: each native platform resolves its real display name at add-time
    # (pixiv=token API, patreon=campaigns API, subscribestar=profile page);
    # gallery-dl platforms and any failure fall back to the URL handle.
    from backend.app.services import patreon_resolver
    from backend.app.services.credential_service import CredentialService
    from backend.app.services.extension_service import ExtensionService
    from backend.app.services.pixiv_client import PixivClient
    from backend.app.services.subscribestar_client import SubscribeStarClient

    async def _tok(self, platform):
        return "tok"

    async def _cookies(self, platform):
        return "/tmp/cookies.txt"

    monkeypatch.setattr(CredentialService, "get_token", _tok)
    monkeypatch.setattr(CredentialService, "get_cookies_path", _cookies)
    monkeypatch.setattr(PixivClient, "resolve_display_name", lambda self, uid: "Pixiv Name")
    monkeypatch.setattr(patreon_resolver, "resolve_display_name", lambda v, c: "Patreon Name")
    monkeypatch.setattr(SubscribeStarClient, "resolve_display_name", lambda self, u: "SS Name")

    svc = ExtensionService(db, crypto=object())     # crypto seam only (calls stubbed)
    assert await svc._resolve_artist_name(
        "pixiv", "555", "https://www.pixiv.net/users/555") == "Pixiv Name"
    assert await svc._resolve_artist_name(
        "patreon", "maewix", "https://patreon.com/maewix") == "Patreon Name"
    assert await svc._resolve_artist_name(
        "subscribestar", "sabu", "https://subscribestar.adult/sabu") == "SS Name"
    # gallery-dl platform → readable handle passthrough (no resolver).
    assert await svc._resolve_artist_name("hentaifoundry", "Foo", "u") == "Foo"
    # No crypto → no resolution attempt → the raw handle.
    assert await ExtensionService(db)._resolve_artist_name("pixiv", "555", "u") == "555"
    # Resolver returns None → fall back to the handle.
    monkeypatch.setattr(patreon_resolver, "resolve_display_name", lambda v, c: None)
    assert await svc._resolve_artist_name("patreon", "maewix", "u") == "maewix"


@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.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_rejects_retired_deviantart(client, ext_key):
    """#3069: a DeviantArt creator URL used to derive cleanly. Now that the
    platform is retired, the extension's own gate should never offer the
    button — but a stale content script on an un-updated browser still can,
    so the backend has to refuse it rather than create an unusable source."""
    resp = await client.post(
        "/api/extension/quick-add-source",
        json={"url": "https://www.deviantart.com/baz"},
        headers={"X-Extension-Key": ext_key},
    )
    assert resp.status_code == 400
    body = await resp.get_json()
    assert body["error"] == "unknown_platform"
    assert "deviantart" not in body["known"]


@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


# --- /api/extension/probe ---------------------------------------------


@pytest.mark.asyncio
async def test_probe_returns_new_when_nothing_exists(client, ext_key):
    resp = await client.get(
        "/api/extension/probe",
        query_string={"url": "https://www.patreon.com/freshcreator"},
        headers={"X-Extension-Key": ext_key},
    )
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["state"] == "new"
    assert body["platform"] == "patreon"
    assert body["slug"] == "freshcreator"


@pytest.mark.asyncio
async def test_probe_returns_source_match_for_already_added(client, ext_key, db):
    artist = Artist(name="Alice", slug="alice", is_subscription=True)
    db.add(artist)
    await db.flush()
    src = Source(
        artist_id=artist.id, platform="patreon",
        url="https://www.patreon.com/alice", enabled=True, config_overrides={},
    )
    db.add(src)
    await db.commit()

    resp = await client.get(
        "/api/extension/probe",
        query_string={"url": "https://www.patreon.com/alice"},
        headers={"X-Extension-Key": ext_key},
    )
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["state"] == "source_match"
    assert body["artist"]["slug"] == "alice"
    assert body["source"]["url"] == "https://www.patreon.com/alice"
    assert body["source"]["platform"] == "patreon"


@pytest.mark.asyncio
async def test_probe_returns_artist_match_when_only_synthetic_anchor_exists(
    client, ext_key, db,
):
    """Filesystem-imported artist with only a sidecar synthetic Source
    for the (artist, platform) — the URL the operator's browsing isn't
    yet a real Source. The probe should collapse this into artist_match
    so the chip says '+ Add Patreon source to Dymkens' rather than
    '+ Add to FabledCurator' (which would re-create the artist)."""
    artist = Artist(name="Dymkens", slug="dymkens", is_subscription=False)
    db.add(artist)
    await db.flush()
    synthetic = Source(
        artist_id=artist.id, platform="patreon",
        url="sidecar:patreon:dymkens", enabled=False, config_overrides={},
    )
    db.add(synthetic)
    await db.commit()

    resp = await client.get(
        "/api/extension/probe",
        query_string={"url": "https://www.patreon.com/dymkens"},
        headers={"X-Extension-Key": ext_key},
    )
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["state"] == "artist_match"
    assert body["artist"]["slug"] == "dymkens"
    assert "source" not in body


@pytest.mark.asyncio
async def test_probe_returns_unknown_platform_for_non_artist_url(client, ext_key):
    """A patreon URL that isn't an artist page (e.g. /home, /posts/N)
    shouldn't trigger the button. Sentinel 'unknown_platform' state
    tells the content script to skip injection."""
    resp = await client.get(
        "/api/extension/probe",
        query_string={"url": "https://www.patreon.com/posts/12345"},
        headers={"X-Extension-Key": ext_key},
    )
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["state"] == "unknown_platform"


@pytest.mark.asyncio
async def test_probe_missing_key_401(client):
    resp = await client.get(
        "/api/extension/probe",
        query_string={"url": "https://www.patreon.com/maewix"},
    )
    assert resp.status_code == 401


@pytest.mark.asyncio
async def test_probe_missing_url_400(client, ext_key):
    resp = await client.get(
        "/api/extension/probe",
        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_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