bvandeusen
0c4b8aef8c
Final piece of the artist decoupling. (1) Identity-by-source: quick_add_source resolves the artist by an existing (platform, url) Source first, so a re-add reuses the artist even after it was renamed (its frozen slug no longer matches the name) — a slug-based lookup would have duplicated it. (2) Pixiv naming: a new pixiv source resolves the real display name via the app API (PixivClient .resolve_display_name → /v1/user/detail) using the stored token, so the artist is 'Kurotsuchi Machi' not '12345678' — and its name-derived slug matches what a native download produces, unifying them. Falls back to the numeric id when no token/crypto. ExtensionService gains the crypto seam; the endpoint passes it. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM
420 lines
15 KiB
Python
420 lines
15 KiB
Python
"""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
|