ad0284585b
Goon — self-hosted aggregator for adult-content scene metadata. Indexes scenes from TPDB, StashDB, and 30+ public adult tube sites. Cross-source deduplication via perceptual hash + Levenshtein distance. FastAPI backend + APScheduler worker + React Native (Expo) mobile client. FOSS, ad-free, donation-funded. See README for details.
86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
"""Test StashDB connectora — paginate, delta-by-updated, error handling."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from app.connectors.stashdb import StashDBConnector
|
|
|
|
|
|
@pytest.fixture
|
|
def stashdb_raw() -> dict:
|
|
return json.loads(
|
|
(Path(__file__).parent / "fixtures" / "stashdb_scene.json").read_text()
|
|
)
|
|
|
|
|
|
def _gql_response(scenes: list[dict]) -> dict:
|
|
return {"data": {"queryScenes": {"count": len(scenes), "scenes": scenes}}}
|
|
|
|
|
|
def test_paginates_until_short_page(stashdb_raw: dict) -> None:
|
|
s2 = {**stashdb_raw, "id": "scene-2", "title": "Two", "updated": "2026-04-29T11:00:00Z"}
|
|
s3 = {**stashdb_raw, "id": "scene-3", "title": "Three", "updated": "2026-04-29T10:00:00Z"}
|
|
|
|
with respx.mock(assert_all_called=True) as mock:
|
|
route = mock.post("https://stashdb.org/graphql").mock(
|
|
side_effect=[
|
|
httpx.Response(200, json=_gql_response([stashdb_raw, s2])),
|
|
httpx.Response(200, json=_gql_response([s3])),
|
|
]
|
|
)
|
|
c = StashDBConnector(api_key="k", per_page=2)
|
|
results = list(c.fetch_scenes())
|
|
assert [r.title for r in results] == ["The Great Heist", "Two", "Three"]
|
|
assert route.call_count == 2
|
|
|
|
|
|
def test_stops_when_updated_before_since(stashdb_raw: dict) -> None:
|
|
fresh = {**stashdb_raw, "updated": "2026-04-30T12:00:00Z"}
|
|
stale = {**stashdb_raw, "id": "scene-stale", "updated": "2026-04-01T12:00:00Z"}
|
|
with respx.mock() as mock:
|
|
mock.post("https://stashdb.org/graphql").mock(
|
|
return_value=httpx.Response(200, json=_gql_response([fresh, stale]))
|
|
)
|
|
c = StashDBConnector(api_key="k")
|
|
since = datetime(2026, 4, 15, tzinfo=UTC)
|
|
results = list(c.fetch_scenes(since=since))
|
|
assert len(results) == 1
|
|
assert results[0].external_id == fresh["id"]
|
|
|
|
|
|
def test_respects_limit(stashdb_raw: dict) -> None:
|
|
s2 = {**stashdb_raw, "id": "scene-2", "title": "Two"}
|
|
with respx.mock() as mock:
|
|
mock.post("https://stashdb.org/graphql").mock(
|
|
return_value=httpx.Response(200, json=_gql_response([stashdb_raw, s2]))
|
|
)
|
|
c = StashDBConnector(api_key="k", per_page=10)
|
|
results = list(c.fetch_scenes(limit=1))
|
|
assert len(results) == 1
|
|
|
|
|
|
def test_raises_on_graphql_errors() -> None:
|
|
with respx.mock() as mock:
|
|
mock.post("https://stashdb.org/graphql").mock(
|
|
return_value=httpx.Response(
|
|
200, json={"errors": [{"message": "unauthorized"}]}
|
|
)
|
|
)
|
|
c = StashDBConnector(api_key="k")
|
|
with pytest.raises(RuntimeError, match="stashdb graphql errors"):
|
|
list(c.fetch_scenes())
|
|
|
|
|
|
def test_api_key_required(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from app.config import get_settings
|
|
monkeypatch.setenv("STASHDB_API_KEY", "")
|
|
get_settings.cache_clear()
|
|
with pytest.raises(RuntimeError, match="STASHDB_API_KEY"):
|
|
StashDBConnector(api_key=None)
|
|
get_settings.cache_clear()
|