tests/integration/test_api.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 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
"""API / HTML integration tests (offline)."""
from __future__ import annotations
from datetime import UTC, datetime
import pytest
from app.config import get_settings
from app.db import reset_engine
from app.db.models import Base, Tracker
from app.main import create_app
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
@pytest.fixture
async def app_client(tmp_path, monkeypatch):
db = tmp_path / "api.db"
url = f"sqlite+aiosqlite:///{db.as_posix()}"
monkeypatch.setenv("DATABASE_URL", url)
monkeypatch.setenv("RUN_SCHEDULER", "false")
monkeypatch.setenv("APP_NAME", "RastroTest")
monkeypatch.setenv("ENABLE_IPV6", "false")
get_settings.cache_clear()
reset_engine()
engine = create_async_engine(url)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
factory = async_sessionmaker(engine, expire_on_commit=False)
async with factory() as session:
session.add(
Tracker(
canonical_url="udp://demo.example:6969/announce",
scheme="udp",
hostname="demo.example",
port=6969,
path="/announce",
current_status="up",
first_seen_at=datetime.now(UTC),
last_seen_at=datetime.now(UTC),
last_checked_at=datetime.now(UTC),
last_latency_ms=42,
consecutive_failures=0,
infrastructure_fingerprint="abc",
supports_ipv4=True,
supports_ipv6="not_tested",
)
)
await session.commit()
await engine.dispose()
# Patch lifespan import to skip network bootstrap
application = create_app()
transport = ASGITransport(app=application)
async with AsyncClient(transport=transport, base_url="http://test") as client:
yield client
get_settings.cache_clear()
reset_engine()
@pytest.mark.asyncio
async def test_health_meta_and_pages(app_client: AsyncClient):
r = await app_client.get("/healthz")
assert r.status_code == 200
assert r.text == "ok"
meta = await app_client.get("/api/v1/meta")
assert meta.status_code == 200
body = meta.json()
assert body["name"] == "RastroTest"
assert "version" in body
home = await app_client.get("/")
assert home.status_code == 200
assert "demo.example" in home.text
assert "RastroTest" in home.text
trackers = await app_client.get("/api/v1/trackers")
assert trackers.status_code == 200
assert trackers.json()["total"] >= 1
detail = await app_client.get("/api/v1/trackers/1")
assert detail.status_code == 200
lists = await app_client.get("/lists")
assert lists.status_code == 200
@pytest.mark.asyncio
async def test_pages_follow_language(app_client: AsyncClient):
home_en = await app_client.get("/", headers={"Accept-Language": "en-US,en;q=0.9"})
assert home_en.status_code == 200
assert 'lang="en"' in home_en.text
assert "watched" in home_en.text
home_pt = await app_client.get(
"/",
headers={"Accept-Language": "en"},
cookies={"lang": "pt"},
)
assert 'lang="pt-BR"' in home_pt.text
assert "monitorados" in home_pt.text
missing_en = await app_client.get("/trackers/99999", headers={"Accept-Language": "en"})
assert missing_en.status_code == 404
assert "not found" in missing_en.text.lower()
missing_pt = await app_client.get("/trackers/99999", cookies={"lang": "pt"})
assert "não encontrado" in missing_pt.text
@pytest.mark.asyncio
async def test_lang_switch_sets_cookie(app_client: AsyncClient):
r = await app_client.get("/lang/en", follow_redirects=False)
assert r.status_code == 303
assert "lang=en" in r.headers.get("set-cookie", "")
@pytest.mark.asyncio
async def test_smart_list_etag(app_client: AsyncClient):
first = await app_client.get("/api/v1/lists/smart.txt")
assert first.status_code == 200
etag = first.headers.get("etag")
assert etag
second = await app_client.get("/api/v1/lists/smart.txt", headers={"If-None-Match": etag})
assert second.status_code == 304
live = await app_client.get("/api/v1/lists/live.txt")
assert live.status_code == 200
assert "demo.example" in live.text
|