Timothy Farrell
f856c92394
- Deleted test_auth.py (auth no longer exists) - Rewrote TestOrderDelayRoute -> TestRootRedirectWithOrderDelay using query params - Updated TestHashPageWithRefresh to use ?order=...&delay=... URLs - Added play button query param assertion in TestHashPage - Removed password=None from test_navigation.py seeded_indexers fixture - Formatted with black, all 59 tests passing
211 lines
8.7 KiB
Python
211 lines
8.7 KiB
Python
"""Tests for FastAPI endpoints."""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
import main
|
|
|
|
|
|
class TestHealthCheck:
|
|
"""Tests for GET /api/health."""
|
|
|
|
async def test_returns_200(self, client_dir: AsyncClient) -> None:
|
|
"""Health endpoint returns 200 OK."""
|
|
response = await client_dir.get("/api/health")
|
|
assert response.status_code == 200
|
|
|
|
async def test_returns_file_count(self, client_dir: AsyncClient) -> None:
|
|
"""Health endpoint reports correct file count."""
|
|
response = await client_dir.get("/api/health")
|
|
data = response.json()
|
|
assert data["status"] == "healthy"
|
|
assert data["file_count"] == len(main.file_mapping)
|
|
|
|
async def test_returns_zero_when_empty(self) -> None:
|
|
"""Health endpoint returns 0 when no files are indexed."""
|
|
main.file_mapping.clear()
|
|
from httpx import ASGITransport
|
|
|
|
transport = ASGITransport(app=main.app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
response = await ac.get("/api/health")
|
|
data = response.json()
|
|
assert data["file_count"] == 0
|
|
|
|
|
|
class TestGetFileData:
|
|
"""Tests for GET /api/{file_hash}/data."""
|
|
|
|
async def test_returns_file_content(self, client_dir: AsyncClient) -> None:
|
|
"""Returns the content of a valid file."""
|
|
file_hash = list(main.file_mapping.keys())[0]
|
|
response = await client_dir.get(f"/api/{file_hash}/data")
|
|
assert response.status_code == 200
|
|
assert len(response.content) > 0
|
|
|
|
async def test_returns_correct_content_type_for_text(
|
|
self, client_dir: AsyncClient
|
|
) -> None:
|
|
"""Text files get text/plain content type."""
|
|
# Find a .txt file hash
|
|
for file_hash, filepath in main.file_mapping.items():
|
|
if filepath.endswith(".txt"):
|
|
response = await client_dir.get(f"/api/{file_hash}/data")
|
|
assert response.status_code == 200
|
|
assert "text/plain" in response.headers["content-type"]
|
|
break
|
|
|
|
async def test_returns_correct_content_type_for_image(
|
|
self, client_dir: AsyncClient
|
|
) -> None:
|
|
"""Image files get appropriate content type."""
|
|
for file_hash, filepath in main.file_mapping.items():
|
|
if filepath.endswith(".jpg"):
|
|
response = await client_dir.get(f"/api/{file_hash}/data")
|
|
assert response.status_code == 200
|
|
assert "image/jpeg" in response.headers["content-type"]
|
|
break
|
|
|
|
async def test_returns_404_for_invalid_hash(self, client_dir: AsyncClient) -> None:
|
|
"""Returns 404 for a hash that doesn't exist."""
|
|
response = await client_dir.get("/api/nonexistent-hash/data")
|
|
assert response.status_code == 404
|
|
|
|
async def test_has_content_disposition_header(
|
|
self, client_dir: AsyncClient
|
|
) -> None:
|
|
"""Response includes Content-Disposition header."""
|
|
file_hash = list(main.file_mapping.keys())[0]
|
|
response = await client_dir.get(f"/api/{file_hash}/data")
|
|
assert "content-disposition" in response.headers
|
|
assert "inline" in response.headers["content-disposition"]
|
|
|
|
async def test_zip_file_content(self, client_zip: AsyncClient) -> None:
|
|
"""Files from a zip archive are served correctly."""
|
|
file_hash = list(main.file_mapping.keys())[0]
|
|
response = await client_zip.get(f"/api/{file_hash}/data")
|
|
assert response.status_code == 200
|
|
assert len(response.content) > 0
|
|
|
|
|
|
class TestRootRedirect:
|
|
"""Tests for GET /."""
|
|
|
|
async def test_redirects_to_random_hash(self, client_dir: AsyncClient) -> None:
|
|
"""Root redirects (307) to a random file hash page."""
|
|
response = await client_dir.get("/", follow_redirects=False)
|
|
assert response.status_code in (307, 302, 301)
|
|
location = response.headers["location"]
|
|
# Location should be a hash in the mapping
|
|
hash_from_url = location.lstrip("/")
|
|
assert hash_from_url in main.file_mapping
|
|
|
|
|
|
class TestRootRedirectWithOrderDelay:
|
|
"""Tests for GET / with order/delay query parameters."""
|
|
|
|
async def test_next_order_redirects(self, client_dir: AsyncClient) -> None:
|
|
"""/?order=next&delay=5 redirects to /{hash}?order=next&delay=5."""
|
|
response = await client_dir.get(
|
|
"/", params={"order": "next", "delay": 5}, follow_redirects=False
|
|
)
|
|
assert response.status_code in (307, 302, 301)
|
|
location = response.headers["location"]
|
|
assert "order=next" in location
|
|
assert "delay=5" in location
|
|
|
|
async def test_random_order_redirects(self, client_dir: AsyncClient) -> None:
|
|
"""/?order=random&delay=3 redirects to /{hash}?order=random&delay=3."""
|
|
response = await client_dir.get(
|
|
"/", params={"order": "random", "delay": 3}, follow_redirects=False
|
|
)
|
|
assert response.status_code in (307, 302, 301)
|
|
location = response.headers["location"]
|
|
assert "order=random" in location
|
|
assert "delay=3" in location
|
|
|
|
|
|
class TestHashPage:
|
|
"""Tests for GET /{file_hash}."""
|
|
|
|
async def test_returns_html_page(self, client_dir: AsyncClient) -> None:
|
|
"""Returns an HTML page for a valid file hash."""
|
|
file_hash = list(main.file_mapping.keys())[0]
|
|
response = await client_dir.get(f"/{file_hash}")
|
|
assert response.status_code == 200
|
|
assert "text/html" in response.headers["content-type"]
|
|
|
|
async def test_page_contains_image_url(self, client_dir: AsyncClient) -> None:
|
|
"""HTML page contains the image URL."""
|
|
file_hash = list(main.file_mapping.keys())[0]
|
|
response = await client_dir.get(f"/{file_hash}")
|
|
assert f"/api/{file_hash}/data" in response.text
|
|
|
|
async def test_page_contains_prev_next_buttons(
|
|
self, client_dir: AsyncClient
|
|
) -> None:
|
|
"""HTML page contains prev and next navigation."""
|
|
file_hash = list(main.file_mapping.keys())[0]
|
|
response = await client_dir.get(f"/{file_hash}")
|
|
assert 'class="chevron left"' in response.text
|
|
assert 'class="chevron right"' in response.text
|
|
|
|
async def test_page_contains_play_button(self, client_dir: AsyncClient) -> None:
|
|
"""HTML page contains play button with query param URL."""
|
|
file_hash = list(main.file_mapping.keys())[0]
|
|
response = await client_dir.get(f"/{file_hash}")
|
|
assert "?order=next&delay=5" in response.text
|
|
|
|
async def test_returns_404_for_invalid_hash(self, client_dir: AsyncClient) -> None:
|
|
"""Returns 404 for a hash that doesn't exist."""
|
|
response = await client_dir.get("/nonexistent-hash")
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestHashPageWithRefresh:
|
|
"""Tests for GET /{file_hash}?order=...&delay=... (auto-refresh mode)."""
|
|
|
|
async def test_next_order_returns_html(self, client_dir: AsyncClient) -> None:
|
|
"""Next order returns HTML with refresh meta tag."""
|
|
file_hash = list(main.file_mapping.keys())[0]
|
|
response = await client_dir.get(
|
|
f"/{file_hash}", params={"order": "next", "delay": 5}
|
|
)
|
|
assert response.status_code == 200
|
|
assert 'http-equiv="refresh"' in response.text
|
|
|
|
async def test_random_order_returns_html(self, client_dir: AsyncClient) -> None:
|
|
"""Random order returns HTML with refresh meta tag."""
|
|
file_hash = list(main.file_mapping.keys())[0]
|
|
response = await client_dir.get(
|
|
f"/{file_hash}", params={"order": "random", "delay": 3}
|
|
)
|
|
assert response.status_code == 200
|
|
assert 'http-equiv="refresh"' in response.text
|
|
|
|
async def test_invalid_order_returns_400(self, client_dir: AsyncClient) -> None:
|
|
"""Invalid order parameter returns 400."""
|
|
file_hash = list(main.file_mapping.keys())[0]
|
|
response = await client_dir.get(
|
|
f"/{file_hash}", params={"order": "shuffle", "delay": 5}
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
async def test_returns_404_for_invalid_hash(self, client_dir: AsyncClient) -> None:
|
|
"""Returns 404 for a hash that doesn't exist."""
|
|
response = await client_dir.get(
|
|
"/nonexistent-hash", params={"order": "next", "delay": 5}
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
async def test_refresh_url_points_to_next_file(
|
|
self, client_dir: AsyncClient
|
|
) -> None:
|
|
"""Refresh meta tag points to the next file in sequence with query params."""
|
|
file_hash = list(main.file_mapping.keys())[0]
|
|
response = await client_dir.get(
|
|
f"/{file_hash}", params={"order": "next", "delay": 5}
|
|
)
|
|
assert "order=next" in response.text
|
|
assert "delay=5" in response.text
|