tests/unit/test_new_collectors.py (view raw)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 |
"""Tests for the jokepool710 and XIU2 collectors and the shared feed runner."""
from __future__ import annotations
import httpx
import pytest
import respx
from app.collectors.jokepool710 import Jokepool710Collector
from app.collectors.xiu2 import Xiu2Collector
from app.config import Settings
from app.db.models import Tracker, TrackerSource
from sqlalchemy import select
FIXTURE_JOKEPOOL = """
udp://tracker.example:6969/announce
http://tracker.example:80/announce
udp://tracker.example:6969/announce
"""
FIXTURE_XIU2_ALL = """
udp://a.example:6969/announce
http://b.example:80/announce
"""
FIXTURE_XIU2_BEST = """
udp://a.example:6969/announce
"""
@pytest.mark.asyncio
@respx.mock
async def test_jokepool710_import_and_empty_does_not_wipe(session, settings: Settings):
settings = Settings(
APP_NAME="RastroTest",
DATABASE_URL=settings.database_url,
JOKEPOOL710_ALL_URL="https://example.test/joke.txt",
RUN_SCHEDULER=False,
)
respx.get("https://example.test/joke.txt").mock(
return_value=httpx.Response(200, text=FIXTURE_JOKEPOOL)
)
collector = Jokepool710Collector()
async with httpx.AsyncClient() as client:
stats = await collector.import_once(session, client, settings)
assert stats.valid == 2
assert stats.new == 2
assert stats.error is None
rows = (await session.execute(select(Tracker))).scalars().all()
assert len(rows) == 2
sources = (await session.execute(select(TrackerSource))).scalars().all()
assert {s.source_name for s in sources} == {"jokepool710"}
# Empty response must preserve existing data.
respx.get("https://example.test/joke.txt").mock(return_value=httpx.Response(200, text="\n\n"))
async with httpx.AsyncClient() as client:
stats2 = await collector.import_once(session, client, settings)
assert stats2.error is not None
rows2 = (await session.execute(select(Tracker))).scalars().all()
assert len(rows2) == 2
@pytest.mark.asyncio
@respx.mock
async def test_xiu2_import_sets_best_flag(session, settings: Settings):
settings = Settings(
APP_NAME="RastroTest",
DATABASE_URL=settings.database_url,
XIU2_ALL_URL="https://example.test/xiu2-all.txt",
XIU2_BEST_URL="https://example.test/xiu2-best.txt",
RUN_SCHEDULER=False,
)
respx.get("https://example.test/xiu2-all.txt").mock(
return_value=httpx.Response(200, text=FIXTURE_XIU2_ALL)
)
respx.get("https://example.test/xiu2-best.txt").mock(
return_value=httpx.Response(200, text=FIXTURE_XIU2_BEST)
)
collector = Xiu2Collector()
async with httpx.AsyncClient() as client:
stats = await collector.import_once(session, client, settings)
assert stats.valid == 2
assert stats.new == 2
assert stats.error is None
sources = (await session.execute(select(TrackerSource))).scalars().all()
best = [s for s in sources if s.is_best_at_source]
assert len(best) == 1
@pytest.mark.asyncio
@respx.mock
async def test_xiu2_best_failure_does_not_fail_import(session, settings: Settings):
settings = Settings(
APP_NAME="RastroTest",
DATABASE_URL=settings.database_url,
XIU2_ALL_URL="https://example.test/xiu2-all.txt",
XIU2_BEST_URL="https://example.test/xiu2-best.txt",
RUN_SCHEDULER=False,
)
respx.get("https://example.test/xiu2-all.txt").mock(
return_value=httpx.Response(200, text=FIXTURE_XIU2_ALL)
)
respx.get("https://example.test/xiu2-best.txt").mock(return_value=httpx.Response(503))
collector = Xiu2Collector()
async with httpx.AsyncClient() as client:
stats = await collector.import_once(session, client, settings)
assert stats.error is None
assert stats.valid == 2
sources = (await session.execute(select(TrackerSource))).scalars().all()
assert all(not s.is_best_at_source for s in sources)
|