feat: retire pixiv entirely — delete its code, its ledgers, its credential (3977, 3978, 3979)

Milestone #406 phase 2, with issue #3980 folded in. Phase 1 (2026-09-13)
unregistered pixiv so nothing could reach it; the code has sat in the tree
uncalled since. DeviantArt is why the second half is not left for later —
#3069 retired it in code on 2026-08-27 and its stored session was still in
the database seven weeks on.

Step 5 — the code. Deletes pixiv_client, pixiv_downloader, pixiv_ingester,
platforms/pixiv and their three test modules and fixture, then edits out
every remaining reference: the dispatch entry, the campaign-id and verify
branches in download_backends, the display-name branch in extension_service,
and the comments that still described pixiv as live.

The consolidation check the step asked for comes back negative:
native_ingest_common has seven non-pixiv callers (patreon, subscribestar,
membership_reconcile, membership_roster, ingest_core), so nothing there
drops to a single user.

Step 6 — the data, alembic 0102. Drops pixiv_seen_media and
pixiv_failed_media, and deletes credential rows whose platform is not
registered. Written as "not registered" rather than "pixiv" at the step's
explicit ask, which is what makes one migration cover two retirements: the
pixiv OAuth refresh token and DeviantArt's leftover session (#3980). It is
also the only way either row can go — the credentials UI renders one card
per platform from /api/platforms and looks the credential up by key, so an
unregistered platform's row has no card and no Remove button.

Pixiv's Source rows are KEPT, changing the milestone's original data table
on the operator's call. `platform` is stored only on Source; neither Post
nor ImageRecord carries it. Both FKs are ON DELETE SET NULL, so a delete
would not lose the art — but it would drop every pixiv image into the
gallery's __unsourced__ bucket and strip the platform chip off every pixiv
post. The rows stay disabled (0097) and unregistered, so nothing schedules
or downloads through them. Keeping them costs nothing and keeps the
attribution that "the art already downloaded from pixiv stays" is about.

Step 7 — the guard. test_pixiv_code_and_tables_are_gone asserts absence from
the module table and from Base.metadata, not from prose (snippet #3352's
trap). The extension and registry negative assertions were already in place
from phase 1. The final sweep found one real residue step 4 missed:
extension/README.md still advertised pixiv support and carried a "Pixiv
OAuth" manual-test item.

Also replaces the two deleted dispatch tests with one over the whole
NATIVE_INGESTER_PLATFORMS set, so adding a platform and forgetting its
ingester class now fails at unit level rather than as a mid-download
KeyError.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

tests/fixtures/pixiv_user_illusts_page1.json

@@ -1,129 +0,0 @@
-{
-  "illusts": [
-    {
-      "id": 111,
-      "title": "Multi Page Adventure",
-      "type": "illust",
-      "caption": "Two-page set. <a href=\"https://example.com/wip\">WIP thread</a>",
-      "create_date": "2026-06-20T18:00:00+09:00",
-      "user": {"id": 99, "name": "Example Artist", "account": "exartist"},
-      "tags": [
-        {"name": "オリジナル", "translated_name": "original"},
-        {"name": "女の子", "translated_name": "girl"}
-      ],
-      "page_count": 2,
-      "width": 1200,
-      "height": 1600,
-      "x_restrict": 0,
-      "series": {"id": 4242, "title": "Adventure Series"},
-      "total_view": 1000,
-      "total_bookmarks": 250,
-      "is_bookmarked": false,
-      "illust_ai_type": 1,
-      "meta_single_page": {},
-      "meta_pages": [
-        {
-          "image_urls": {
-            "square_medium": "https://i.pximg.net/c/360x360_70/img-master/img/2026/06/20/18/00/00/111_p0_square1200.jpg",
-            "original": "https://i.pximg.net/img-original/img/2026/06/20/18/00/00/111_p0.png"
-          }
-        },
-        {
-          "image_urls": {
-            "square_medium": "https://i.pximg.net/c/360x360_70/img-master/img/2026/06/20/18/00/00/111_p1_square1200.jpg",
-            "original": "https://i.pximg.net/img-original/img/2026/06/20/18/00/00/111_p1.png"
-          }
-        }
-      ]
-    },
-    {
-      "id": 222,
-      "title": "Single Piece",
-      "type": "illust",
-      "caption": "",
-      "create_date": "2026-06-18T12:30:00+09:00",
-      "user": {"id": 99, "name": "Example Artist", "account": "exartist"},
-      "tags": [{"name": "落書き", "translated_name": "doodle"}],
-      "page_count": 1,
-      "width": 900,
-      "height": 900,
-      "x_restrict": 1,
-      "series": null,
-      "total_view": 500,
-      "total_bookmarks": 60,
-      "is_bookmarked": true,
-      "illust_ai_type": 0,
-      "meta_single_page": {
-        "original_image_url": "https://i.pximg.net/img-original/img/2026/06/18/12/30/00/222_p0.jpg"
-      },
-      "meta_pages": []
-    },
-    {
-      "id": 333,
-      "title": "Wiggle Loop",
-      "type": "ugoira",
-      "caption": "animated",
-      "create_date": "2026-06-15T09:00:00+09:00",
-      "user": {"id": 99, "name": "Example Artist", "account": "exartist"},
-      "tags": [{"name": "うごイラ", "translated_name": "ugoira"}],
-      "page_count": 1,
-      "width": 600,
-      "height": 600,
-      "x_restrict": 0,
-      "series": null,
-      "total_view": 300,
-      "total_bookmarks": 40,
-      "is_bookmarked": false,
-      "illust_ai_type": 0,
-      "meta_single_page": {
-        "original_image_url": "https://i.pximg.net/img-original/img/2026/06/15/09/00/00/333_ugoira0.jpg"
-      },
-      "meta_pages": []
-    },
-    {
-      "id": 444,
-      "title": "Blocked Work",
-      "type": "illust",
-      "caption": "",
-      "create_date": "2026-06-10T00:00:00+09:00",
-      "user": {"id": 99, "name": "Example Artist", "account": "exartist"},
-      "tags": [],
-      "page_count": 1,
-      "width": 0,
-      "height": 0,
-      "x_restrict": 2,
-      "series": null,
-      "total_view": 0,
-      "total_bookmarks": 0,
-      "is_bookmarked": false,
-      "illust_ai_type": 0,
-      "meta_single_page": {
-        "original_image_url": "https://s.pximg.net/common/images/limit_sanity_level_360.png"
-      },
-      "meta_pages": []
-    },
-    {
-      "id": 555,
-      "title": "Ghost Work",
-      "type": "illust",
-      "caption": "",
-      "create_date": "2026-06-01T00:00:00+09:00",
-      "user": {"id": 0, "name": "", "account": ""},
-      "tags": [],
-      "page_count": 1,
-      "width": 0,
-      "height": 0,
-      "x_restrict": 0,
-      "series": null,
-      "total_view": 0,
-      "total_bookmarks": 0,
-      "is_bookmarked": false,
-      "illust_ai_type": 0,
-      "meta_single_page": {
-        "original_image_url": "https://i.pximg.net/img-original/img/2026/06/01/00/00/00/555_p0.png"
-      },
-      "meta_pages": []
-    }
-  ],
-  "next_url": "https://app-api.pixiv.net/v1/user/illusts?user_id=99&offset=30"
-}

tests/test_api_extension.py

@@ -87,29 +87,21 @@ async def test_quick_add_reuses_source_artist_after_rename(client, ext_key):
 @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.
+    # (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(
@@ -117,7 +109,7 @@ async def test_resolve_artist_name_dispatches_per_platform(db, monkeypatch):
     # 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"
+    assert await ExtensionService(db)._resolve_artist_name("patreon", "maewix", "u") == "maewix"
     # 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"

tests/test_download_backends.py

@@ -7,7 +7,6 @@ import pytest
 
 from backend.app.services.download_backends import (
     NATIVE_INGESTER_PLATFORMS,
-    _campaign_resolution_error,
     _native_ingester_cls,
     _unsupported_platform_message,
     run_download,
@@ -15,7 +14,6 @@ from backend.app.services.download_backends import (
     verify_source_credential,
 )
 from backend.app.services.gallery_dl import ErrorType
-from backend.app.services.pixiv_ingester import PixivIngester
 
 
 def test_native_platforms():
@@ -25,8 +23,9 @@ def test_native_platforms():
 
 
 def test_pixiv_is_no_longer_native():
-    """Retired at milestone #406. The refusal below is what stops it falling
-    through to gallery-dl now that it is not native."""
+    """Retired at milestone #406 — unregistered in phase 1, deleted in phase 2.
+    The refusal below is what stops it falling through to gallery-dl now that
+    it is neither native nor registered."""
     assert uses_native_ingester("pixiv") is False
     assert "pixiv" not in NATIVE_INGESTER_PLATFORMS
 
@@ -113,11 +112,14 @@ def test_unknown_platform_is_not_native():
     assert uses_native_ingester("nonsense") is False
 
 
-def test_pixiv_dispatches_to_its_ingester():
-    assert _native_ingester_cls("pixiv") is PixivIngester
+def test_every_native_platform_dispatches_to_an_ingester():
+    """The dispatch table and NATIVE_INGESTER_PLATFORMS have to agree, or a
+    platform that routes native raises KeyError mid-download instead of being
+    refused up front. Written over the set rather than per-platform so adding
+    one to NATIVE_INGESTER_PLATFORMS and forgetting the class fails here.
 
+    (This replaces the per-platform dispatch tests, one of which was pixiv's;
+    it was deleted with pixiv at milestone #406 phase 2.)"""
+    for platform in NATIVE_INGESTER_PLATFORMS:
+        assert _native_ingester_cls(platform) is not None
 
-def test_pixiv_resolution_error_names_the_expected_url_shape():
-    msg = _campaign_resolution_error("pixiv", "https://www.pixiv.net/artworks/1")
-    assert "pixiv user id" in msg
-    assert "users/<id>" in msg

tests/test_pixiv_client.py

@@ -1,419 +0,0 @@
-"""PixivClient tests — parsing + iteration against canned pages, no network.
-
-The fixture mirrors a real `/v1/user/illusts` page (multi-page work, single
-page, ugoira, sanity-limited placeholder, deleted-author ghost). HTTP is
-stubbed at the requests-session seam (oauth POST + API GET), so the exact
-gallery-dl-parity request profile — headers, oauth form, pagination params —
-is asserted rather than assumed.
-"""
-
-import json
-from pathlib import Path
-
-import pytest
-
-from backend.app.services.pixiv_client import (
-    PIXIV_APP_HEADERS,
-    MediaItem,
-    PixivAPIError,
-    PixivAuthError,
-    PixivClient,
-    PixivDriftError,
-    rating_label,
-    user_id_from_url,
-)
-
-_FIXTURE = Path(__file__).parent / "fixtures" / "pixiv_user_illusts_page1.json"
-
-
-class FakeResponse:
-    def __init__(self, status_code=200, json_data=None, headers=None):
-        self.status_code = status_code
-        self._json = json_data
-        self.headers = headers or {}
-
-    def json(self):
-        if self._json is None:
-            raise ValueError("no JSON")
-        return self._json
-
-
-class FakeSession:
-    """Minimal requests.Session stand-in: canned responses per (method, url
-    fragment), recording every call for profile assertions."""
-
-    def __init__(self):
-        self.headers = dict(PIXIV_APP_HEADERS)
-        self.responses = []
-        self.calls = []
-
-    def queue(self, response):
-        self.responses.append(response)
-        return self
-
-    def _next(self):
-        if not self.responses:
-            raise AssertionError("FakeSession ran out of queued responses")
-        return self.responses.pop(0)
-
-    def post(self, url, data=None, headers=None, timeout=None):
-        self.calls.append(("POST", url, data, headers))
-        return self._next()
-
-    def get(self, url, params=None, timeout=None, headers=None):
-        self.calls.append(("GET", url, params, headers))
-        return self._next()
-
-
-def _oauth_ok():
-    return FakeResponse(200, {
-        "response": {
-            "access_token": "acc-token",
-            "expires_in": 3600,
-            "user": {"id": "77", "account": "operator", "name": "Op"},
-        }
-    })
-
-
-@pytest.fixture
-def page1():
-    return json.loads(_FIXTURE.read_text())
-
-
-@pytest.fixture
-def client():
-    # Never issues a request in pure-parsing tests.
-    return PixivClient("refresh-tok", session=FakeSession())
-
-
-def _post_for(client, page1, work_id):
-    for work in page1["illusts"]:
-        if work["id"] == work_id:
-            return client._normalize(work)
-    raise AssertionError(f"no work {work_id} in fixture")
-
-
-# -- URL → user id ----------------------------------------------------------
-
-def test_user_id_from_url_variants():
-    assert user_id_from_url("https://www.pixiv.net/users/12345678") == "12345678"
-    assert user_id_from_url("https://www.pixiv.net/en/users/42") == "42"
-    assert user_id_from_url("https://pixiv.net/users/7/artworks") == "7"
-    assert user_id_from_url("https://www.pixiv.net/member.php?id=99") == "99"
-
-
-def test_user_id_from_url_rejects_non_matches():
-    assert user_id_from_url("https://www.pixiv.net/artworks/111") is None
-    assert user_id_from_url("https://www.pixiv.net/users/notdigits") is None
-    assert user_id_from_url("https://example.com/users/5") is None
-    assert user_id_from_url("") is None
-
-
-# -- normalization ------------------------------------------------------------
-
-def test_normalize_maps_attributes(client, page1):
-    post = _post_for(client, page1, 111)
-    attrs = post["attributes"]
-    assert post["id"] == 111
-    assert attrs["title"] == "Multi Page Adventure"
-    assert "Two-page set." in attrs["content"]
-    assert attrs["published_at"] == "2026-06-20T18:00:00+09:00"
-    assert attrs["post_type"] == "illust"
-    assert post["_work"]["total_bookmarks"] == 250
-
-
-def test_post_record_key_and_meta(client, page1):
-    post = _post_for(client, page1, 222)
-    assert client.post_record_key(post) == ("post:222", "222")
-    meta = client.post_meta(post)
-    assert meta["title"] == "Single Piece"
-    assert meta["date"] == "2026-06-18T12:30:00+09:00"
-    assert client.post_record_key({"id": None}) is None
-
-
-# -- gating -------------------------------------------------------------------
-
-def test_post_is_gated_limit_placeholder(client, page1):
-    assert client.post_is_gated(_post_for(client, page1, 444)) is True
-
-
-def test_post_is_gated_deleted_author(client, page1):
-    assert client.post_is_gated(_post_for(client, page1, 555)) is True
-
-
-def test_post_is_gated_normal_works(client, page1):
-    assert client.post_is_gated(_post_for(client, page1, 111)) is False
-    assert client.post_is_gated(_post_for(client, page1, 222)) is False
-
-
-# -- extract_media --------------------------------------------------------------
-
-def test_extract_media_multi_page(client, page1):
-    items = client.extract_media(_post_for(client, page1, 111), {})
-    assert len(items) == 2
-    assert all(isinstance(m, MediaItem) for m in items)
-    assert [m.media_id for m in items] == ["p0", "p1"]
-    assert items[0].url.endswith("/111_p0.png")
-    assert items[1].url.endswith("/111_p1.png")
-    # gallery-dl filename parity: {id}_{title[:50]}_{num:>02}.{extension}
-    assert items[0].filename == "111_Multi Page Adventure_00.png"
-    assert items[1].filename == "111_Multi Page Adventure_01.png"
-    assert all(m.post_id == "111" for m in items)
-
-
-def test_extract_media_single_page(client, page1):
-    items = client.extract_media(_post_for(client, page1, 222), {})
-    assert len(items) == 1
-    assert items[0].media_id == "p0"
-    assert items[0].url.endswith("/222_p0.jpg")
-    assert items[0].filename == "222_Single Piece_00.jpg"
-
-
-def test_extract_media_gated_yields_nothing(client, page1):
-    assert client.extract_media(_post_for(client, page1, 444), {}) == []
-    assert client.extract_media(_post_for(client, page1, 555), {}) == []
-
-
-def test_extract_media_ugoira_zip_swap(client, page1, monkeypatch):
-    frames = [{"file": "000000.jpg", "delay": 90}, {"file": "000001.jpg", "delay": 90}]
-
-    def fake_call(endpoint, params):
-        assert endpoint == "/v1/ugoira/metadata"
-        assert params == {"illust_id": "333"}
-        return {"ugoira_metadata": {
-            "zip_urls": {"medium": (
-                "https://i.pximg.net/img-zip-ugoira/img/2026/06/15/09/00/00/"
-                "333_ugoira600x600.zip"
-            )},
-            "frames": frames,
-        }}
-
-    monkeypatch.setattr(client, "_call", fake_call)
-    post = _post_for(client, page1, 333)
-    items = client.extract_media(post, {})
-    assert len(items) == 1
-    assert items[0].media_id == "ugoira"
-    assert items[0].kind == "ugoira"
-    assert items[0].url.endswith("333_ugoira1920x1080.zip")
-    assert items[0].filename == "333_Wiggle Loop_00.zip"
-    # Frame delays memoized for the post record (future video conversion).
-    assert post["_work"]["_ugoira_frames"] == frames
-
-
-def test_fetch_ugoira_frames_memoizes_and_shares_one_call(client, page1, monkeypatch):
-    # fetch_ugoira_frames (called by write_post_record, which the core runs
-    # BEFORE extract_media) populates frames; extract_media then REUSES the
-    # memoized metadata — exactly one /v1/ugoira/metadata call total.
-    frames = [{"file": "000000.jpg", "delay": 90}]
-    calls = []
-
-    def fake_call(endpoint, params):
-        calls.append(endpoint)
-        return {"ugoira_metadata": {
-            "zip_urls": {"medium": (
-                "https://i.pximg.net/img-zip-ugoira/x/333_ugoira600x600.zip"
-            )},
-            "frames": frames,
-        }}
-
-    monkeypatch.setattr(client, "_call", fake_call)
-    post = _post_for(client, page1, 333)
-    client.fetch_ugoira_frames(post)
-    assert post["_work"]["_ugoira_frames"] == frames
-    items = client.extract_media(post, {})          # reuses memoized meta
-    assert items[0].media_id == "ugoira"
-    assert calls == ["/v1/ugoira/metadata"]          # ONE fetch, not two
-
-
-def test_fetch_ugoira_frames_noop_for_non_ugoira(client, page1, monkeypatch):
-    monkeypatch.setattr(client, "_call", lambda e, p: pytest.fail("should not fetch"))
-    post = _post_for(client, page1, 222)             # a single-page illust
-    client.fetch_ugoira_frames(post)
-    assert "_ugoira_frames" not in post["_work"]
-
-
-def test_extract_media_ugoira_metadata_failure_downgrades(
-    client, page1, monkeypatch
-):
-    def fake_call(endpoint, params):
-        raise PixivAPIError("boom", status_code=500)
-
-    monkeypatch.setattr(client, "_call", fake_call)
-    assert client.extract_media(_post_for(client, page1, 333), {}) == []
-
-
-def test_extract_media_ugoira_auth_failure_stays_loud(
-    client, page1, monkeypatch
-):
-    def fake_call(endpoint, params):
-        raise PixivAuthError("token dead", status_code=400)
-
-    monkeypatch.setattr(client, "_call", fake_call)
-    with pytest.raises(PixivAuthError):
-        client.extract_media(_post_for(client, page1, 333), {})
-
-
-# -- iteration ------------------------------------------------------------------
-
-def test_iter_posts_paginates_and_carries_cursor(page1, monkeypatch):
-    client = PixivClient("refresh-tok", session=FakeSession())
-    page2 = {
-        "illusts": [dict(page1["illusts"][1], id=666, title="Older")],
-        "next_url": None,
-    }
-    seen_params = []
-
-    def fake_call(endpoint, params):
-        assert endpoint == "/v1/user/illusts"
-        seen_params.append(dict(params))
-        return page1 if len(seen_params) == 1 else page2
-
-    monkeypatch.setattr(client, "_call", fake_call)
-    rows = list(client.iter_posts("99"))
-
-    assert seen_params[0] == {"user_id": "99"}
-    # Page 2 params come verbatim from next_url's query string.
-    assert seen_params[1] == {"user_id": "99", "offset": "30"}
-    # Page-1 posts carry cursor None; page-2 posts carry the fetching cursor.
-    assert [c for _, _, c in rows[:5]] == [None] * 5
-    assert rows[5][2] == "user_id=99&offset=30"
-    assert rows[5][0]["id"] == 666
-
-
-def test_iter_posts_resumes_from_cursor(page1, monkeypatch):
-    client = PixivClient("refresh-tok", session=FakeSession())
-    captured = {}
-
-    def fake_call(endpoint, params):
-        captured["params"] = dict(params)
-        return {"illusts": [], "next_url": None}
-
-    monkeypatch.setattr(client, "_call", fake_call)
-    list(client.iter_posts("99", cursor="user_id=99&offset=60"))
-    assert captured["params"] == {"user_id": "99", "offset": "60"}
-
-
-def test_iter_posts_rejects_non_numeric_campaign():
-    client = PixivClient("refresh-tok", session=FakeSession())
-    with pytest.raises(PixivDriftError):
-        next(client.iter_posts("https://www.pixiv.net/users/99"))
-
-
-def test_iter_posts_drift_on_missing_illusts(monkeypatch):
-    client = PixivClient("refresh-tok", session=FakeSession())
-    monkeypatch.setattr(client, "_call", lambda e, p: {"error": None, "body": 1})
-    with pytest.raises(PixivDriftError):
-        next(client.iter_posts("99"))
-
-
-# -- auth ------------------------------------------------------------------------
-
-def test_login_sends_gallery_dl_profile_and_sets_bearer():
-    session = FakeSession().queue(_oauth_ok())
-    client = PixivClient("refresh-tok", session=session)
-    client._login()
-
-    method, url, data, headers = session.calls[0]
-    assert method == "POST"
-    assert url == "https://oauth.secure.pixiv.net/auth/token"
-    assert data["grant_type"] == "refresh_token"
-    assert data["refresh_token"] == "refresh-tok"
-    assert data["client_id"] == "MOBrBDS8blbauoSck0ZfDbtuzpyT"
-    assert data["get_secure_url"] == "1"
-    # X-Client-Time/-Hash pair: ISO + literal +00:00, md5 hex digest.
-    assert headers["X-Client-Time"].endswith("+00:00")
-    assert len(headers["X-Client-Hash"]) == 32
-    assert session.headers["Authorization"] == "Bearer acc-token"
-    # Token is fresh → a second login is a no-op (no extra POST).
-    client._login()
-    assert len(session.calls) == 1
-
-
-def test_login_maps_rejection_to_auth_error():
-    session = FakeSession().queue(FakeResponse(400, {"has_error": True}))
-    client = PixivClient("refresh-tok", session=session)
-    with pytest.raises(PixivAuthError):
-        client._login()
-
-
-def test_login_without_token_is_auth_error():
-    client = PixivClient(None, session=FakeSession())
-    with pytest.raises(PixivAuthError):
-        client._login()
-
-
-def test_call_maps_rate_limit_message():
-    session = FakeSession().queue(_oauth_ok()).queue(FakeResponse(403, {
-        "error": {"message": "Rate Limit", "user_message": ""},
-    }))
-    client = PixivClient("refresh-tok", session=session)
-    with pytest.raises(PixivAPIError) as exc_info:
-        client._call("/v1/user/illusts", {"user_id": "1"})
-    err = exc_info.value
-    assert not isinstance(err, PixivAuthError)
-    assert err.status_code == 429
-    assert err.retry_after == 300.0
-
-
-def test_call_maps_403_to_auth_error():
-    session = FakeSession().queue(_oauth_ok()).queue(FakeResponse(403, {
-        "error": {"message": "invalid access token", "user_message": ""},
-    }))
-    client = PixivClient("refresh-tok", session=session)
-    with pytest.raises(PixivAuthError):
-        client._call("/v1/user/illusts", {"user_id": "1"})
-
-
-def test_call_maps_404_status():
-    session = FakeSession().queue(_oauth_ok()).queue(FakeResponse(404, {
-        "error": {"message": "Not Found", "user_message": ""},
-    }))
-    client = PixivClient("refresh-tok", session=session)
-    with pytest.raises(PixivAPIError) as exc_info:
-        client._call("/v1/user/illusts", {"user_id": "1"})
-    assert exc_info.value.status_code == 404
-
-
-def test_verify_auth_reports_account():
-    session = FakeSession().queue(_oauth_ok())
-    client = PixivClient("refresh-tok", session=session)
-    ok, message = client.verify_auth()
-    assert ok is True
-    assert "operator" in message
-
-
-def test_verify_auth_bad_token():
-    session = FakeSession().queue(FakeResponse(400, {"has_error": True}))
-    client = PixivClient("bad", session=session)
-    ok, message = client.verify_auth()
-    assert ok is False
-    assert "rotate" in message.lower() or "rejected" in message.lower()
-
-
-def test_resolve_display_name(client, monkeypatch):
-    monkeypatch.setattr(
-        client, "_call",
-        lambda e, p: {"user": {"name": "Kurotsuchi Machi", "id": p["user_id"]}},
-    )
-    assert client.resolve_display_name("99") == "Kurotsuchi Machi"
-
-
-def test_resolve_display_name_none_on_failure(client, monkeypatch):
-    def boom(endpoint, params):
-        raise PixivAPIError("nope", status_code=404)
-    monkeypatch.setattr(client, "_call", boom)
-    assert client.resolve_display_name("99") is None
-    # Empty/whitespace name → None (caller falls back to the id).
-    monkeypatch.setattr(client, "_call", lambda e, p: {"user": {"name": "  "}})
-    assert client.resolve_display_name("99") is None
-
-
-# -- rating ------------------------------------------------------------------------
-
-def test_rating_label():
-    assert rating_label(0) == "General"
-    assert rating_label(1) == "R-18"
-    assert rating_label(2) == "R-18G"
-    assert rating_label(None) is None
-    assert rating_label(True) is None
-    assert rating_label(9) is None

tests/test_pixiv_downloader.py

@@ -1,309 +0,0 @@
-"""Unit tests for PixivDownloader — no network, no DB.
-
-The HTTP layer is stubbed via the `session` seam (a fake session whose `.get`
-returns canned streaming bytes). Media items come from the real
-PixivClient.extract_media over the shared fixture page, so the downloader is
-exercised against the exact shapes the client produces.
-"""
-
-from __future__ import annotations
-
-import json
-from pathlib import Path
-
-import requests
-
-from backend.app.services.pixiv_client import MediaItem, PixivClient
-from backend.app.services.pixiv_downloader import PixivDownloader
-
-_FIXTURE = Path(__file__).parent / "fixtures" / "pixiv_user_illusts_page1.json"
-
-# Minimal valid PNG so the file_validator passes on the happy path.
-_PNG_HEAD = b"\x89PNG\r\n\x1a\n"
-_PNG_TAIL = b"\x00\x00\x00\x00IEND\xaeB`\x82"
-_PNG_BYTES = _PNG_HEAD + (b"\x00" * 16) + _PNG_TAIL
-
-
-class _FakeResponse:
-    def __init__(self, payload: bytes, status_code: int = 200, headers=None):
-        self._payload = payload
-        self.status_code = status_code
-        self.headers = headers or {}
-
-    def raise_for_status(self):
-        if self.status_code >= 400:
-            raise requests.HTTPError(f"HTTP {self.status_code}")
-        return None
-
-    def iter_content(self, chunk_size=65536):
-        for i in range(0, len(self._payload), chunk_size):
-            yield self._payload[i : i + chunk_size]
-
-
-class _FakeSession:
-    """Records GETs and returns canned bytes (per-URL, default _PNG_BYTES)."""
-
-    def __init__(self, payloads: dict[str, bytes] | None = None):
-        self.payloads = payloads or {}
-        self.calls: list[str] = []
-
-    def get(self, url, stream=False, timeout=None, headers=None):
-        self.calls.append(url)
-        return _FakeResponse(self.payloads.get(url, _PNG_BYTES))
-
-
-def _client():
-    # Parsing only — extract_media for non-ugoira works never issues a request.
-    return PixivClient("tok", session=_FakeSession())
-
-
-def _post_and_media(work_id):
-    page = json.loads(_FIXTURE.read_text())
-    client = _client()
-    for work in page["illusts"]:
-        if work["id"] == work_id:
-            post = client._normalize(work)
-            return post, client.extract_media(post, {})
-    raise AssertionError(f"no work {work_id} in fixture")
-
-
-def _post_only(work_id):
-    """Normalized post WITHOUT media extraction — the write_post_record tests
-    don't need media, and extracting an ugoira would hit the API seam."""
-    page = json.loads(_FIXTURE.read_text())
-    for work in page["illusts"]:
-        if work["id"] == work_id:
-            return _client()._normalize(work)
-    raise AssertionError(f"no work {work_id} in fixture")
-
-
-def _downloader(tmp_path, session=None):
-    # validate=False: the stub payload is PNG bytes regardless of the target
-    # extension, so real validation would quarantine the .jpg cases. The
-    # validate/quarantine plumbing is BaseNativeDownloader's, covered by the
-    # Patreon downloader tests.
-    return PixivDownloader(
-        tmp_path, validate=False,
-        session=session if session is not None else _FakeSession(),
-    )
-
-
-def test_download_post_writes_gallery_dl_layout(tmp_path):
-    post, media = _post_and_media(111)
-    dl = _downloader(tmp_path)
-    outcomes = dl.download_post(post, media, "artist-a")
-
-    assert [o.status for o in outcomes] == ["downloaded", "downloaded"]
-    flat = tmp_path / "artist-a" / "pixiv" / "pixiv"
-    p0 = flat / "111_Multi Page Adventure_00.png"
-    p1 = flat / "111_Multi Page Adventure_01.png"
-    assert p0.is_file()
-    assert p1.is_file()
-    # No per-post directory — pixiv's layout is flat (gallery-dl parity).
-    assert {p.parent for p in (p0, p1)} == {flat}
-
-
-def test_download_post_writes_minimal_sidecar(tmp_path):
-    post, media = _post_and_media(222)
-    dl = _downloader(tmp_path)
-    outcomes = dl.download_post(post, media, "artist-a")
-
-    assert outcomes[0].status == "downloaded"
-    sidecar = outcomes[0].path.with_suffix(".json")
-    data = json.loads(sidecar.read_text())
-    # Post-first: image identity ONLY — the body lives in the post record.
-    assert data == {
-        "category": "pixiv",
-        "id": "222",
-        "source_url": media[0].url,
-    }
-
-
-def test_download_post_skips_seen(tmp_path):
-    post, media = _post_and_media(222)
-    session = _FakeSession()
-    dl = _downloader(tmp_path, session=session)
-    outcomes = dl.download_post(post, media, "artist-a", is_seen=lambda m: True)
-    assert [o.status for o in outcomes] == ["skipped_seen"]
-    assert session.calls == []
-
-
-def test_download_post_skips_on_disk(tmp_path):
-    post, media = _post_and_media(222)
-    flat = tmp_path / "artist-a" / "pixiv" / "pixiv"
-    flat.mkdir(parents=True)
-    existing = flat / "222_Single Piece_00.jpg"
-    existing.write_bytes(b"already here")
-
-    session = _FakeSession()
-    dl = _downloader(tmp_path, session=session)
-    outcomes = dl.download_post(post, media, "artist-a")
-    assert [o.status for o in outcomes] == ["skipped_disk"]
-    assert outcomes[0].path == existing
-    assert session.calls == []
-    assert existing.read_bytes() == b"already here"
-
-
-def test_recapture_surfaces_on_disk_path_for_seen_media(tmp_path):
-    post, media = _post_and_media(222)
-    flat = tmp_path / "artist-a" / "pixiv" / "pixiv"
-    flat.mkdir(parents=True)
-    (flat / "222_Single Piece_00.jpg").write_bytes(b"kept")
-
-    dl = _downloader(tmp_path)
-    outcomes = dl.download_post(
-        post, media, "artist-a", is_seen=lambda m: True, recapture=True,
-    )
-    # On disk → surfaced with its path (relink channel); seen-but-missing
-    # stays skipped_seen (recovery's job, not recapture's).
-    assert outcomes[0].status == "skipped_disk"
-    assert outcomes[0].path is not None
-
-
-def test_recapture_does_not_redownload_missing_seen_media(tmp_path):
-    post, media = _post_and_media(222)
-    session = _FakeSession()
-    dl = _downloader(tmp_path, session=session)
-    outcomes = dl.download_post(
-        post, media, "artist-a", is_seen=lambda m: True, recapture=True,
-    )
-    assert [o.status for o in outcomes] == ["skipped_seen"]
-    assert session.calls == []
-
-
-def test_download_post_honours_should_stop(tmp_path):
-    post, media = _post_and_media(111)
-    dl = _downloader(tmp_path)
-    outcomes = dl.download_post(post, media, "artist-a", should_stop=lambda: True)
-    assert outcomes == []
-
-
-def test_filename_matches_gallery_dl_linux_restrict(tmp_path):
-    # gallery-dl on Linux replaces ONLY "/" and deletes control chars — the
-    # Windows-forbidden set (<>:"|?*) stays RAW in the on-disk name. Matching
-    # that byte-for-byte is what lets the tier-2 disk-skip recognize
-    # pre-cutover files; a stricter sanitizer would re-download them.
-    post, _ = _post_and_media(222)
-    weird = MediaItem(
-        url="https://i.pximg.net/img-original/img/2026/06/18/12/30/00/222_p0.jpg",
-        filename='222_What? A "Title"/Slash\x0900.jpg',  # ? " stay, / → _, \t dropped
-        kind="image",
-        filehash=None,
-        post_id="222",
-        media_id="p0",
-    )
-    dl = _downloader(tmp_path)
-    outcomes = dl.download_post(post, [weird], "artist-a")
-    assert outcomes[0].status == "downloaded"
-    assert outcomes[0].path.name == '222_What? A "Title"_Slash00.jpg'
-
-
-def test_gdl_clean_filename_parity():
-    from backend.app.services.pixiv_downloader import gdl_clean_filename
-    # Only "/" is replaced; the Windows-forbidden set is untouched.
-    assert gdl_clean_filename('a<>:"|?*b.png') == 'a<>:"|?*b.png'
-    assert gdl_clean_filename("a/b/c.png") == "a_b_c.png"
-    # Control chars (newline, tab, DEL) are deleted, not replaced.
-    assert gdl_clean_filename("a\nb\tc\x7fd.png") == "abcd.png"
-    # Trailing dots/spaces are NOT stripped on Linux.
-    assert gdl_clean_filename("title . .png") == "title . .png"
-
-
-def test_own_session_carries_app_referer(tmp_path):
-    # Built session (no injection) must carry the app profile — i.pximg.net
-    # 403s GETs without the app-api Referer.
-    dl = PixivDownloader(tmp_path)
-    assert dl.session.headers["Referer"] == "https://app-api.pixiv.net/"
-    assert dl.session.headers["App-OS"] == "ios"
-
-
-def test_write_post_record_enriched(tmp_path):
-    post = _post_only(111)
-    dl = _downloader(tmp_path)
-    outcome = dl.write_post_record(post, "artist-a")
-
-    assert outcome.path == (
-        tmp_path / "artist-a" / "pixiv" / "pixiv" / "_post_111.json"
-    )
-    assert outcome.title == "Multi Page Adventure"
-    assert outcome.post_type == "illust"
-    assert outcome.body_chars == len(post["attributes"]["content"])
-
-    data = json.loads(outcome.path.read_text())
-    assert data["category"] == "pixiv"
-    assert data["id"] == "111"
-    assert data["published_at"] == "2026-06-20T18:00:00+09:00"
-    assert data["rating"] == "General"
-    assert data["page_count"] == 2
-    assert data["total_bookmarks"] == 250
-    assert data["total_view"] == 1000
-    assert data["illust_ai_type"] == 1
-    assert data["series"] == {"id": 4242, "title": "Adventure Series"}
-    assert data["tags"] == [
-        {"name": "オリジナル", "translated_name": "original"},
-        {"name": "女の子", "translated_name": "girl"},
-    ]
-    assert data["user"] == {"id": 99, "account": "exartist", "name": "Example Artist"}
-    # JP tags stay human-readable on disk (ensure_ascii=False).
-    assert "オリジナル" in outcome.path.read_text()
-
-
-def test_write_post_record_rating_r18(tmp_path):
-    post = _post_only(222)
-    dl = _downloader(tmp_path)
-    outcome = dl.write_post_record(post, "artist-a")
-    data = json.loads(outcome.path.read_text())
-    assert data["rating"] == "R-18"
-    assert data["is_bookmarked"] is True
-
-
-def test_write_post_record_includes_ugoira_frames(tmp_path):
-    post = _post_only(333)
-    frames = [{"file": "000000.jpg", "delay": 90}]
-    post["_work"]["_ugoira_frames"] = frames
-    dl = _downloader(tmp_path)
-    outcome = dl.write_post_record(post, "artist-a")
-    data = json.loads(outcome.path.read_text())
-    assert data["type"] == "ugoira"
-    assert data["ugoira_frames"] == frames
-
-
-def test_write_post_record_fetches_ugoira_frames_when_absent(tmp_path):
-    # The core writes the post record BEFORE extract_media, so frames aren't
-    # memoized yet — the injected fetcher must populate them so the record keeps
-    # the timings (regression: they were silently always empty).
-    post = _post_only(333)
-    assert "_ugoira_frames" not in post["_work"]
-    frames = [{"file": "000000.jpg", "delay": 120}]
-    calls = []
-
-    def fetcher(p):
-        calls.append(p)
-        p["_work"]["_ugoira_frames"] = frames
-
-    dl = PixivDownloader(
-        tmp_path, validate=False, session=_FakeSession(),
-        ugoira_frames_fetcher=fetcher,
-    )
-    outcome = dl.write_post_record(post, "artist-a")
-    data = json.loads(outcome.path.read_text())
-    assert data["ugoira_frames"] == frames
-    assert len(calls) == 1
-
-
-def test_write_post_record_non_ugoira_does_not_call_fetcher(tmp_path):
-    post = _post_only(111)                       # a plain multi-page illust
-    calls = []
-    dl = PixivDownloader(
-        tmp_path, validate=False, session=_FakeSession(),
-        ugoira_frames_fetcher=lambda p: calls.append(p),
-    )
-    dl.write_post_record(post, "artist-a")
-    assert calls == []
-
-
-def test_write_post_record_without_id(tmp_path):
-    dl = _downloader(tmp_path)
-    outcome = dl.write_post_record({"id": None, "attributes": {}}, "artist-a")
-    assert outcome.path is None
-    assert outcome.body_chars == 0

tests/test_pixiv_ingester.py

@@ -1,305 +0,0 @@
-"""PixivIngester tests — the adapter's wiring over the shared core.
-
-The core's walk/skip/cursor/budget behavior is exercised exhaustively by
-test_patreon_ingester / test_subscribestar_native; these cover what is
-pixiv-SPECIFIC: the synthesized ledger key, the real Postgres pixiv ledgers
-(the upserts' constraint names must match migration 0076), the failure
-mapping including the rate-limit retry_after carry, and the #862 body-canary
-opt-out (caption-less pixiv feeds must not fail API_DRIFT).
-"""
-
-import pytest
-from sqlalchemy import func, select
-from sqlalchemy.orm import sessionmaker
-
-from backend.app.models import Artist, PixivFailedMedia, PixivSeenMedia, Source
-from backend.app.services.gallery_dl import ErrorType
-from backend.app.services.ingest_core import _CANARY_MIN_SAMPLE
-from backend.app.services.native_ingest_common import MediaOutcome, PostRecordOutcome
-from backend.app.services.pixiv_client import (
-    MediaItem,
-    PixivAPIError,
-    PixivAuthError,
-    PixivDriftError,
-)
-from backend.app.services.pixiv_ingester import PixivIngester, _ledger_key
-
-pytestmark = pytest.mark.integration
-
-
-# --- fakes ----------------------------------------------------------------
-
-
-def _media(post_id, num, *, filehash=None, media_id=None):
-    mid = media_id if media_id is not None else f"p{num}"
-    return MediaItem(
-        url=f"https://i.pximg.net/img-original/img/2026/07/01/00/00/00/{post_id}_{mid}.png",
-        filename=f"{post_id}_Work_{num:02d}.png",
-        kind="image",
-        filehash=filehash,
-        post_id=str(post_id),
-        media_id=mid,
-    )
-
-
-class _FakeClient:
-    """Stub PixivClient. `pages` is a list of (page_cursor, [posts]); each post
-    is (post_id, [MediaItem]). `raise_exc` trips a client-level failure;
-    `captions` toggles per-post body text (the canary input)."""
-
-    def __init__(self, pages, raise_exc=None, captions=True):
-        self._pages = pages
-        self._raise_exc = raise_exc
-        self._captions = captions
-
-    def iter_posts(self, campaign_id, cursor=None):
-        if self._raise_exc is not None:
-            raise self._raise_exc
-        for page_cursor, posts in self._pages:
-            for post_id, media in posts:
-                caption = f"<p>caption {post_id}</p>" if self._captions else ""
-                yield (
-                    {
-                        "id": post_id,
-                        "_media": media,
-                        "attributes": {
-                            "title": f"Work {post_id}",
-                            "post_type": "illust",
-                            "content": caption,
-                            "published_at": "2026-07-01T00:00:00+09:00",
-                        },
-                        "_work": {},
-                    },
-                    {},
-                    page_cursor,
-                )
-
-    def extract_media(self, post, included_index):
-        return post["_media"]
-
-    def post_meta(self, post):
-        return {"title": post.get("id"), "date": None}
-
-    @staticmethod
-    def post_is_gated(post):
-        return False
-
-    @staticmethod
-    def post_record_key(post):
-        pid = str(post.get("id") or "")
-        return (f"post:{pid}", pid) if pid else None
-
-
-class _FakeDownloader:
-    """Stub PixivDownloader. Honors the injected is_seen; media whose ledger
-    key is in `error` report a download failure."""
-
-    def __init__(self, tmp_path, error=None):
-        self.tmp_path = tmp_path
-        self.error = set(error or ())
-        self.download_calls = 0
-
-    def download_post(self, post, media_items, artist_slug, *, is_seen,
-                      should_stop=lambda: False, recapture=False):
-        outcomes = []
-        for m in media_items:
-            if should_stop():
-                break
-            if is_seen(m) and not recapture:
-                outcomes.append(
-                    MediaOutcome(media=m, status="skipped_seen", path=None, error=None)
-                )
-            elif _ledger_key(m) in self.error:
-                self.download_calls += 1
-                outcomes.append(
-                    MediaOutcome(media=m, status="error", path=None, error="403 pximg")
-                )
-            else:
-                self.download_calls += 1
-                p = self.tmp_path / m.filename
-                p.write_bytes(b"x")
-                outcomes.append(
-                    MediaOutcome(media=m, status="downloaded", path=p, error=None)
-                )
-        return outcomes
-
-    def write_post_record(self, post, artist_slug):
-        p = self.tmp_path / f"_post_{post.get('id')}.json"
-        p.write_text("{}")
-        attrs = post.get("attributes") or {}
-        body = attrs.get("content")
-        return PostRecordOutcome(
-            path=p,
-            post_type=attrs.get("post_type"),
-            title=attrs.get("title"),
-            body_chars=len(body) if isinstance(body, str) else 0,
-        )
-
-
-@pytest.fixture
-async def source_id(db):
-    artist = Artist(name="Pixland", slug="pixland")
-    db.add(artist)
-    await db.flush()
-    source = Source(
-        artist_id=artist.id, platform="pixiv",
-        url="https://www.pixiv.net/users/99", enabled=True, config_overrides={},
-    )
-    db.add(source)
-    await db.commit()
-    return source.id
-
-
-def _ingester(sync_engine, tmp_path, client, downloader):
-    factory = sessionmaker(sync_engine, expire_on_commit=False)
-    return PixivIngester(
-        images_root=tmp_path, cookies_path=None, session_factory=factory,
-        client=client, downloader=downloader,
-    )
-
-
-def _run(ing, source_id, mode="tick"):
-    return ing.run(
-        source_id=source_id, campaign_id="99", artist_slug="pixland",
-        url="https://www.pixiv.net/users/99", mode=mode,
-    )
-
-
-def _seen_keys(sync_engine, source_id):
-    factory = sessionmaker(sync_engine, expire_on_commit=False)
-    with factory() as s:
-        return set(s.execute(
-            select(PixivSeenMedia.filehash).where(
-                PixivSeenMedia.source_id == source_id
-            )
-        ).scalars().all())
-
-
-# --- ledger key -------------------------------------------------------------
-
-
-def test_ledger_key_shapes():
-    assert _ledger_key(_media(111, 0)) == "111:p0"
-    assert _ledger_key(_media(111, 4)) == "111:p4"
-    assert _ledger_key(_media(333, 0, media_id="ugoira")) == "333:ugoira"
-    assert _ledger_key(_media(1, 0, filehash="f" * 32)) == "f" * 32
-    long_key = _ledger_key(_media("9" * 200, 0))
-    assert len(long_key) <= 128
-
-
-# --- walk + ledgers ---------------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_tick_downloads_and_marks_seen(source_id, sync_engine, tmp_path):
-    m0, m1 = _media(111, 0), _media(111, 1)
-    client = _FakeClient([(None, [(111, [m0, m1])])])
-    downloader = _FakeDownloader(tmp_path)
-    ing = _ingester(sync_engine, tmp_path, client, downloader)
-
-    result = _run(ing, source_id)
-
-    assert result.success is True
-    assert result.files_downloaded == 2
-    assert len(result.post_record_paths) == 1
-    # 2 media keys + the synthetic post key, in the pixiv-shaped key format.
-    assert _seen_keys(sync_engine, source_id) == {"111:p0", "111:p1", "post:111"}
-
-
-@pytest.mark.asyncio
-async def test_tick_skips_seen_via_ledger(source_id, sync_engine, tmp_path):
-    m0 = _media(111, 0)
-    factory = sessionmaker(sync_engine, expire_on_commit=False)
-    with factory() as s:
-        s.add(PixivSeenMedia(
-            source_id=source_id, filehash=_ledger_key(m0), post_id="111",
-        ))
-        s.commit()
-
-    client = _FakeClient([(None, [(111, [m0])])])
-    downloader = _FakeDownloader(tmp_path)
-    ing = _ingester(sync_engine, tmp_path, client, downloader)
-
-    result = _run(ing, source_id)
-    assert result.files_downloaded == 0
-    assert downloader.download_calls == 0
-
-
-@pytest.mark.asyncio
-async def test_failed_media_lands_in_dead_letter_ledger(
-    source_id, sync_engine, tmp_path
-):
-    m0 = _media(111, 0)
-    client = _FakeClient([(None, [(111, [m0])])])
-    downloader = _FakeDownloader(tmp_path, error={_ledger_key(m0)})
-    ing = _ingester(sync_engine, tmp_path, client, downloader)
-
-    result = _run(ing, source_id)
-    assert result.run_stats["per_item_failures"] == 1
-
-    factory = sessionmaker(sync_engine, expire_on_commit=False)
-    with factory() as s:
-        row = s.execute(
-            select(PixivFailedMedia).where(
-                PixivFailedMedia.source_id == source_id
-            )
-        ).scalar_one()
-    assert row.filehash == "111:p0"
-    assert row.attempts == 1
-    assert "403" in row.last_error
-
-
-# --- failure mapping ----------------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_auth_error_maps_to_auth_error(source_id, sync_engine, tmp_path):
-    client = _FakeClient([], raise_exc=PixivAuthError(
-        "rotate the token", status_code=400,
-    ))
-    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
-    result = _run(ing, source_id)
-    assert result.success is False
-    assert result.error_type == ErrorType.AUTH_ERROR
-
-
-@pytest.mark.asyncio
-async def test_drift_error_maps_to_api_drift(source_id, sync_engine, tmp_path):
-    client = _FakeClient([], raise_exc=PixivDriftError("no illusts list"))
-    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
-    result = _run(ing, source_id)
-    assert result.error_type == ErrorType.API_DRIFT
-    assert "Pixiv app API changed" in result.error_message
-
-
-@pytest.mark.asyncio
-async def test_rate_limit_carries_retry_after(source_id, sync_engine, tmp_path):
-    client = _FakeClient([], raise_exc=PixivAPIError(
-        "rate limited", status_code=429, retry_after=300.0,
-    ))
-    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
-    result = _run(ing, source_id)
-    assert result.error_type == ErrorType.RATE_LIMITED
-    assert result.retry_after_seconds == 300.0
-
-
-# --- body canary opt-out --------------------------------------------------------
-
-
-@pytest.mark.asyncio
-async def test_captionless_feed_does_not_trip_body_canary(
-    source_id, sync_engine, tmp_path
-):
-    """#862 canary opt-out: a pixiv backfill recording ≥ the canary sample of
-    caption-less works is NORMAL (many artists never write captions) — it must
-    complete, not fail API_DRIFT the way a zero-body Patreon walk does."""
-    n = _CANARY_MIN_SAMPLE + 5
-    posts = [(1000 + i, [_media(1000 + i, 0)]) for i in range(n)]
-    client = _FakeClient([(None, posts)], captions=False)
-    downloader = _FakeDownloader(tmp_path)
-    ing = _ingester(sync_engine, tmp_path, client, downloader)
-
-    result = _run(ing, source_id, mode="backfill")
-    assert result.success is True
-    assert result.error_type is None
-    assert result.files_downloaded == n

tests/test_platforms_registry.py

@@ -26,6 +26,40 @@ def test_pixiv_is_retired():
     assert "pixiv" not in known_platform_keys()
 
 
+def test_pixiv_code_and_tables_are_gone():
+    """Milestone #406 phase 2 (step 7). Phase 1 made pixiv unreachable; this is
+    the guard that it stays deleted rather than drifting back in.
+
+    Both halves assert absence from a DATA STRUCTURE — the module table and the
+    declarative metadata — not from prose. Snippet #3352's trap is an absence
+    check against a comment, which passes the moment someone rewords the
+    comment; a re-added module or a re-declared table cannot hide from these.
+
+    DeviantArt is the reason this exists: #3069 retired it in code on
+    2026-08-27 and its credential row was still sitting in the database seven
+    weeks later (#3980). A retirement nothing asserts is a retirement that
+    half-happens.
+    """
+    import importlib
+
+    from backend.app.models.base import Base
+
+    for module in (
+        "backend.app.services.pixiv_client",
+        "backend.app.services.pixiv_downloader",
+        "backend.app.services.pixiv_ingester",
+        "backend.app.services.platforms.pixiv",
+        "backend.app.models.pixiv_seen_media",
+        "backend.app.models.pixiv_failed_media",
+    ):
+        with pytest.raises(ModuleNotFoundError):
+            importlib.import_module(module)
+
+    # Alembic 0102 drops these; nothing may re-declare them.
+    assert "pixiv_seen_media" not in Base.metadata.tables
+    assert "pixiv_failed_media" not in Base.metadata.tables
+
+
 def test_fanbox_not_in_registry():
     # Sanity check — FC-3a added 'fanbox' by mistake; it's not a GS platform.
     assert "fanbox" not in PLATFORMS