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.
77 lines
2.9 KiB
Python
77 lines
2.9 KiB
Python
"""Test TPDB connectora z mockowanym httpx (respx) — paginacja, limit, delta."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from app.connectors.tpdb import TPDBConnector
|
|
|
|
|
|
@pytest.fixture
|
|
def fixture_scene() -> dict:
|
|
return json.loads((Path(__file__).parent / "fixtures" / "tpdb_scene.json").read_text())
|
|
|
|
|
|
def _page(scenes: list[dict], current: int, last: int) -> dict:
|
|
return {
|
|
"data": scenes,
|
|
"meta": {"current_page": current, "last_page": last, "per_page": 100, "total": len(scenes) * last},
|
|
}
|
|
|
|
|
|
def test_paginates_and_stops_at_last_page(fixture_scene: dict) -> None:
|
|
page1 = _page([fixture_scene], 1, 2)
|
|
page2_scene = {**fixture_scene, "id": "ddddffff-1111-4111-8111-222222222222", "title": "Two"}
|
|
page2 = _page([page2_scene], 2, 2)
|
|
|
|
with respx.mock(base_url="https://api.theporndb.net") as mock:
|
|
route = mock.get("/scenes").mock(
|
|
side_effect=[httpx.Response(200, json=page1), httpx.Response(200, json=page2)]
|
|
)
|
|
c = TPDBConnector(token="t")
|
|
scenes = list(c.fetch_scenes())
|
|
assert [s.title for s in scenes] == ["The Great Heist", "Two"]
|
|
assert route.call_count == 2
|
|
|
|
|
|
def test_respects_limit_and_does_not_fetch_more(fixture_scene: dict) -> None:
|
|
page1 = _page([fixture_scene, {**fixture_scene, "id": "00000000-0000-4000-8000-000000000001", "title": "B"}], 1, 5)
|
|
with respx.mock(base_url="https://api.theporndb.net") as mock:
|
|
route = mock.get("/scenes").mock(return_value=httpx.Response(200, json=page1))
|
|
c = TPDBConnector(token="t")
|
|
scenes = list(c.fetch_scenes(limit=1))
|
|
assert len(scenes) == 1
|
|
# only one page hit because limit reached mid-page
|
|
assert route.call_count == 1
|
|
|
|
|
|
def test_passes_date_filter_when_since_provided(fixture_scene: dict) -> None:
|
|
from datetime import datetime
|
|
|
|
with respx.mock(base_url="https://api.theporndb.net") as mock:
|
|
route = mock.get("/scenes").mock(return_value=httpx.Response(200, json=_page([], 1, 1)))
|
|
c = TPDBConnector(token="t")
|
|
list(c.fetch_scenes(since=datetime(2024, 6, 1, 12, 0, 0)))
|
|
request = route.calls[0].request
|
|
assert "date=2024-06-01" in str(request.url)
|
|
|
|
|
|
def test_token_required(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
# nadpisz wartość z .env (settings cache trzeba wyczyścić)
|
|
from app.config import get_settings
|
|
monkeypatch.setenv("TPDB_API_TOKEN", "")
|
|
get_settings.cache_clear()
|
|
with pytest.raises(RuntimeError, match="TPDB_API_TOKEN"):
|
|
TPDBConnector(token=None)
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_handles_empty_data_payload() -> None:
|
|
with respx.mock(base_url="https://api.theporndb.net") as mock:
|
|
mock.get("/scenes").mock(return_value=httpx.Response(200, json={"data": [], "meta": {}}))
|
|
c = TPDBConnector(token="t")
|
|
assert list(c.fetch_scenes()) == []
|