all repos — rastro @ main

BitTorrent tracker!

tests/unit/test_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
 
"""Collector parsing and failure isolation tests."""

from __future__ import annotations

import httpx
import pytest
import respx
from app.collectors.base import parse_and_normalize
from app.collectors.trackerslist import TrackerslistCollector
from app.config import Settings
from app.db.models import Tracker, TrackerSource
from sqlalchemy import select

FIXTURE_ALL = """
udp://tracker.example:6969/announce

http://tracker.example:80/announce

udp://tracker.example:6969/announce

ws://ignored.example/announce
"""

FIXTURE_BEST = """
udp://tracker.example:6969/announce
"""


def test_parse_and_normalize_counts():
    parsed = parse_and_normalize(FIXTURE_ALL)
    assert parsed.received == 3
    assert len(parsed.valid) == 2
    assert len(parsed.unsupported) == 1


@pytest.mark.asyncio
@respx.mock
async def test_trackerslist_import_and_empty_does_not_wipe(session, settings: Settings):
    settings = Settings(
        APP_NAME="RastroTest",
        DATABASE_URL=settings.database_url,
        TRACKERSLIST_ALL_URL="https://example.test/all.txt",
        TRACKERSLIST_BEST_URL="https://example.test/best.txt",
        RUN_SCHEDULER=False,
    )
    respx.get("https://example.test/all.txt").mock(
        return_value=httpx.Response(200, text=FIXTURE_ALL)
    )
    respx.get("https://example.test/best.txt").mock(
        return_value=httpx.Response(200, text=FIXTURE_BEST)
    )
    collector = TrackerslistCollector()
    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 any(s.is_best_at_source for s in sources)

    # Failed empty import must preserve data
    respx.get("https://example.test/all.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_source_failure_preserves_previous(session, settings: Settings):
    settings = Settings(
        APP_NAME="RastroTest",
        DATABASE_URL=settings.database_url,
        TRACKERSLIST_ALL_URL="https://example.test/all.txt",
        TRACKERSLIST_BEST_URL="https://example.test/best.txt",
        RUN_SCHEDULER=False,
    )
    respx.get("https://example.test/all.txt").mock(
        return_value=httpx.Response(200, text="udp://ok.example:80/announce\n")
    )
    respx.get("https://example.test/best.txt").mock(return_value=httpx.Response(200, text=""))
    collector = TrackerslistCollector()
    async with httpx.AsyncClient() as client:
        await collector.import_once(session, client, settings)

    respx.get("https://example.test/all.txt").mock(return_value=httpx.Response(503))
    async with httpx.AsyncClient() as client:
        stats = await collector.import_once(session, client, settings)
    assert stats.error
    count = len((await session.execute(select(Tracker))).scalars().all())
    assert count == 1