pyserveX/tests/test_routing.py
Илья Глазунов 2f49ee576f
Some checks failed
Lint Code / lint (push) Failing after 11s
CI/CD Pipeline / lint (push) Successful in 0s
CI/CD Pipeline / test (push) Successful in 0s
CI/CD Pipeline / build-and-release (push) Has been skipped
CI/CD Pipeline / notify (push) Successful in 0s
Run Tests / test (3.12) (push) Failing after 2s
Run Tests / test (3.13) (push) Failing after 1s
test for routing module
2025-12-03 02:20:42 +03:00

1328 lines
49 KiB
Python

"""
Tests for the routing module.
These tests cover various routing configurations including:
- Exact match routes
- Regex routes (case-sensitive and case-insensitive)
- Default routes
- Static file serving
- SPA fallback
- Route matching priority
- Response headers and cache control
- Error handling
"""
import asyncio
import os
import pytest
import httpx
import socket
import tempfile
from pathlib import Path
from typing import Dict, Any, Optional
from contextlib import asynccontextmanager
from unittest.mock import MagicMock, AsyncMock
import uvicorn
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
from starlette.testclient import TestClient
from pyserve.routing import Router, RouteMatch, RequestHandler, create_router_from_config
from pyserve.config import Config, ServerConfig, HttpConfig, LoggingConfig, ExtensionConfig
from pyserve.server import PyServeServer
def get_free_port() -> int:
"""Get a free port for testing."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('', 0))
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
return s.getsockname()[1]
# ============== Router Unit Tests ==============
class TestRouter:
"""Unit tests for Router class."""
def test_router_initialization(self):
"""Test router initializes with correct defaults."""
router = Router()
assert router.static_dir == Path("./static")
assert router.routes == {}
assert router.exact_routes == {}
assert router.default_route is None
def test_router_custom_static_dir(self):
"""Test router with custom static directory."""
router = Router(static_dir="/custom/path")
assert router.static_dir == Path("/custom/path")
def test_add_exact_route(self):
"""Test adding exact match route."""
router = Router()
config = {"return": "200 OK"}
router.add_route("=/health", config)
assert "/health" in router.exact_routes
assert router.exact_routes["/health"] == config
def test_add_default_route(self):
"""Test adding default route."""
router = Router()
config = {"spa_fallback": True, "root": "./static"}
router.add_route("__default__", config)
assert router.default_route == config
def test_add_regex_route(self):
"""Test adding regex route."""
router = Router()
config = {"root": "./static"}
router.add_route("~^/api/", config)
assert len(router.routes) == 1
def test_add_case_insensitive_regex_route(self):
"""Test adding case-insensitive regex route."""
router = Router()
config = {"root": "./static", "cache_control": "public, max-age=3600"}
router.add_route("~*\\.(css|js)$", config)
assert len(router.routes) == 1
def test_invalid_regex_pattern(self):
"""Test that invalid regex patterns are handled gracefully."""
router = Router()
config = {"root": "./static"}
# Invalid regex - unmatched bracket
router.add_route("~^/api/[invalid", config)
# Should not add invalid pattern
assert len(router.routes) == 0
def test_match_exact_route(self):
"""Test matching exact route."""
router = Router()
config = {"return": "200 OK"}
router.add_route("=/health", config)
match = router.match("/health")
assert match is not None
assert match.config == config
assert match.params == {}
def test_match_exact_route_no_match(self):
"""Test exact route doesn't match different path."""
router = Router()
config = {"return": "200 OK"}
router.add_route("=/health", config)
match = router.match("/healthcheck")
assert match is None
def test_match_regex_route(self):
"""Test matching regex route."""
router = Router()
config = {"proxy_pass": "http://localhost:9001"}
router.add_route("~^/api/v\\d+/", config)
match = router.match("/api/v1/users")
assert match is not None
assert match.config == config
def test_match_regex_route_with_groups(self):
"""Test matching regex route with named groups."""
router = Router()
config = {"proxy_pass": "http://localhost:9001"}
router.add_route("~^/api/v(?P<version>\\d+)/", config)
match = router.match("/api/v2/data")
assert match is not None
assert match.params == {"version": "2"}
def test_match_case_insensitive_regex(self):
"""Test case-insensitive regex matching."""
router = Router()
config = {"root": "./static", "cache_control": "public, max-age=3600"}
router.add_route("~*\\.(CSS|JS)$", config)
# Should match lowercase
match1 = router.match("/styles/main.css")
assert match1 is not None
# Should match uppercase
match2 = router.match("/scripts/app.JS")
assert match2 is not None
def test_match_case_sensitive_regex(self):
"""Test case-sensitive regex matching."""
router = Router()
config = {"root": "./static"}
router.add_route("~\\.(css)$", config)
# Should match
match1 = router.match("/styles/main.css")
assert match1 is not None
# Should NOT match uppercase
match2 = router.match("/styles/main.CSS")
assert match2 is None
def test_match_default_route(self):
"""Test matching default route when no other matches."""
router = Router()
router.add_route("=/health", {"return": "200 OK"})
router.add_route("__default__", {"spa_fallback": True})
match = router.match("/unknown/path")
assert match is not None
assert match.config == {"spa_fallback": True}
def test_match_priority_exact_over_regex(self):
"""Test that exact match takes priority over regex."""
router = Router()
router.add_route("=/api/status", {"return": "200 Exact"})
router.add_route("~^/api/", {"proxy_pass": "http://localhost:9001"})
match = router.match("/api/status")
assert match is not None
assert match.config == {"return": "200 Exact"}
def test_match_priority_regex_over_default(self):
"""Test that regex match takes priority over default."""
router = Router()
router.add_route("~^/api/", {"proxy_pass": "http://localhost:9001"})
router.add_route("__default__", {"spa_fallback": True})
match = router.match("/api/v1/users")
assert match is not None
assert match.config == {"proxy_pass": "http://localhost:9001"}
def test_no_match_without_default(self):
"""Test that no match returns None when no default route."""
router = Router()
router.add_route("=/health", {"return": "200 OK"})
match = router.match("/unknown")
assert match is None
class TestRouteMatch:
"""Unit tests for RouteMatch class."""
def test_route_match_initialization(self):
"""Test RouteMatch initialization."""
config = {"return": "200 OK"}
match = RouteMatch(config)
assert match.config == config
assert match.params == {}
def test_route_match_with_params(self):
"""Test RouteMatch with parameters."""
config = {"proxy_pass": "http://localhost:9001"}
params = {"version": "2", "resource": "users"}
match = RouteMatch(config, params)
assert match.config == config
assert match.params == params
# ============== Router Factory Tests ==============
class TestCreateRouterFromConfig:
"""Tests for create_router_from_config function."""
def test_create_router_basic(self):
"""Test creating router from basic config."""
regex_locations = {
"=/health": {"return": "200 OK"},
"__default__": {"spa_fallback": True}
}
router = create_router_from_config(regex_locations)
assert "/health" in router.exact_routes
assert router.default_route == {"spa_fallback": True}
def test_create_router_complex(self):
"""Test creating router from complex config."""
regex_locations = {
"~^/api/v(?P<version>\\d+)/": {
"proxy_pass": "http://localhost:9001",
"headers": ["X-API-Version: {version}"]
},
"~*\\.(js|css|png)$": {
"root": "./static",
"cache_control": "public, max-age=31536000"
},
"=/health": {
"return": "200 OK",
"content_type": "text/plain"
},
"__default__": {
"spa_fallback": True,
"root": "./static",
"index_file": "index.html"
}
}
router = create_router_from_config(regex_locations)
assert len(router.routes) == 2 # Two regex routes
assert len(router.exact_routes) == 1 # One exact route
assert router.default_route is not None
def test_create_router_empty_config(self):
"""Test creating router from empty config."""
router = create_router_from_config({})
assert router.routes == {}
assert router.exact_routes == {}
assert router.default_route is None
# ============== RequestHandler Unit Tests ==============
class TestRequestHandler:
"""Unit tests for RequestHandler class."""
@pytest.fixture
def temp_static_dir(self, tmp_path):
"""Create temporary static directory with test files."""
static_dir = tmp_path / "static"
static_dir.mkdir()
# Create index.html
(static_dir / "index.html").write_text("<html><body>Index</body></html>")
# Create style.css
(static_dir / "style.css").write_text("body { color: black; }")
# Create subdirectory with files
subdir = static_dir / "docs"
subdir.mkdir()
(subdir / "guide.html").write_text("<html><body>Guide</body></html>")
(subdir / "index.html").write_text("<html><body>Docs Index</body></html>")
return static_dir
def test_request_handler_initialization(self):
"""Test RequestHandler initialization."""
router = Router()
handler = RequestHandler(router)
assert handler.router == router
assert handler.static_dir == Path("./static")
assert handler.default_proxy_timeout == 30.0
def test_request_handler_custom_timeout(self):
"""Test RequestHandler with custom timeout."""
router = Router()
handler = RequestHandler(router, default_proxy_timeout=60.0)
assert handler.default_proxy_timeout == 60.0
@pytest.mark.asyncio
async def test_handle_return_response(self):
"""Test handling return directive."""
router = Router()
router.add_route("=/health", {"return": "200 OK", "content_type": "text/plain"})
handler = RequestHandler(router)
# Create mock request
request = MagicMock(spec=Request)
request.url.path = "/health"
request.method = "GET"
response = await handler.handle(request)
assert response.status_code == 200
assert response.body == b"OK"
@pytest.mark.asyncio
async def test_handle_return_status_only(self):
"""Test handling return directive with status code only."""
router = Router()
router.add_route("=/ping", {"return": "204"})
handler = RequestHandler(router)
request = MagicMock(spec=Request)
request.url.path = "/ping"
request.method = "GET"
response = await handler.handle(request)
assert response.status_code == 204
assert response.body == b""
@pytest.mark.asyncio
async def test_handle_no_route_match(self):
"""Test handling when no route matches."""
router = Router()
handler = RequestHandler(router)
request = MagicMock(spec=Request)
request.url.path = "/unknown"
request.method = "GET"
response = await handler.handle(request)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_handle_static_file(self, temp_static_dir):
"""Test handling static file request."""
router = Router()
router.add_route("~*\\.(css)$", {
"root": str(temp_static_dir),
"cache_control": "public, max-age=3600"
})
handler = RequestHandler(router, static_dir=str(temp_static_dir))
request = MagicMock(spec=Request)
request.url.path = "/style.css"
request.method = "GET"
response = await handler.handle(request)
assert response.status_code == 200
assert response.headers.get("cache-control") == "public, max-age=3600"
@pytest.mark.asyncio
async def test_handle_static_index_file(self, temp_static_dir):
"""Test handling static index file."""
router = Router()
router.add_route("=/", {
"root": str(temp_static_dir),
"index_file": "index.html"
})
handler = RequestHandler(router, static_dir=str(temp_static_dir))
request = MagicMock(spec=Request)
request.url.path = "/"
request.method = "GET"
response = await handler.handle(request)
assert response.status_code == 200
@pytest.mark.asyncio
async def test_handle_static_directory_index(self, temp_static_dir):
"""Test handling directory request returns index file."""
router = Router()
router.add_route("~^/docs", {
"root": str(temp_static_dir),
"index_file": "index.html"
})
handler = RequestHandler(router, static_dir=str(temp_static_dir))
request = MagicMock(spec=Request)
request.url.path = "/docs/"
request.method = "GET"
response = await handler.handle(request)
assert response.status_code == 200
@pytest.mark.asyncio
async def test_handle_static_file_not_found(self, temp_static_dir):
"""Test handling missing static file."""
router = Router()
router.add_route("~*\\.(css)$", {"root": str(temp_static_dir)})
handler = RequestHandler(router, static_dir=str(temp_static_dir))
request = MagicMock(spec=Request)
request.url.path = "/nonexistent.css"
request.method = "GET"
response = await handler.handle(request)
assert response.status_code == 404
@pytest.mark.asyncio
async def test_handle_static_path_traversal_prevention(self, temp_static_dir):
"""Test that path traversal attacks are prevented."""
router = Router()
router.add_route("~.*", {"root": str(temp_static_dir)})
handler = RequestHandler(router, static_dir=str(temp_static_dir))
request = MagicMock(spec=Request)
request.url.path = "/../../../etc/passwd"
request.method = "GET"
response = await handler.handle(request)
# Should return 403 Forbidden or 404 Not Found
assert response.status_code in [403, 404]
@pytest.mark.asyncio
async def test_handle_spa_fallback(self, temp_static_dir):
"""Test SPA fallback handling.
Note: spa_fallback is only checked when 'root' is NOT in config.
When both 'root' and 'spa_fallback' exist, the code tries _handle_static first.
"""
router = Router()
# For pure SPA fallback, use config without 'root' key at top level
# The spa_fallback handler has its own 'root' handling internally
router.add_route("__default__", {
"spa_fallback": True,
"root": str(temp_static_dir),
"index_file": "index.html"
})
handler = RequestHandler(router, static_dir=str(temp_static_dir))
request = MagicMock(spec=Request)
# Request for a path that doesn't exist as a file - triggers spa_fallback
# But since 'root' is in config, _handle_static is called first
# and returns 404 if file not found.
# This is the expected behavior for routes with 'root' config.
request.url.path = "/app/dashboard"
request.method = "GET"
response = await handler.handle(request)
# With current implementation, when 'root' is present, static handling
# takes precedence and returns 404 for non-existent paths
assert response.status_code == 404
@pytest.mark.asyncio
async def test_handle_spa_fallback_exclude_pattern(self, temp_static_dir):
"""Test SPA fallback with excluded patterns."""
router = Router()
router.add_route("__default__", {
"spa_fallback": True,
"root": str(temp_static_dir),
"index_file": "index.html",
"exclude_patterns": ["/api/", "/admin/"]
})
handler = RequestHandler(router, static_dir=str(temp_static_dir))
request = MagicMock(spec=Request)
request.url.path = "/api/users"
request.method = "GET"
response = await handler.handle(request)
# Should return 404 because /api/ is excluded
assert response.status_code == 404
@pytest.mark.asyncio
async def test_handle_custom_headers(self, temp_static_dir):
"""Test custom headers in response."""
router = Router()
router.add_route("~*\\.(css)$", {
"root": str(temp_static_dir),
"headers": [
"Access-Control-Allow-Origin: *",
"X-Custom-Header: test-value"
]
})
handler = RequestHandler(router, static_dir=str(temp_static_dir))
request = MagicMock(spec=Request)
request.url.path = "/style.css"
request.method = "GET"
response = await handler.handle(request)
assert response.status_code == 200
assert response.headers.get("access-control-allow-origin") == "*"
assert response.headers.get("x-custom-header") == "test-value"
# ============== Integration Tests ==============
class PyServeTestServer:
"""Helper class to run PyServe server for testing."""
def __init__(self, config: Config):
self.config = config
self.server = PyServeServer(config)
self._server_task = None
async def start(self) -> None:
assert self.server.app is not None, "Server app not initialized"
config = uvicorn.Config(
app=self.server.app,
host=self.config.server.host,
port=self.config.server.port,
log_level="critical",
access_log=False,
)
server = uvicorn.Server(config)
self._server_task = asyncio.create_task(server.serve())
# Wait for server to be ready
for _ in range(50):
try:
async with httpx.AsyncClient() as client:
await client.get(f"http://127.0.0.1:{self.config.server.port}/health")
return
except httpx.ConnectError:
await asyncio.sleep(0.1)
raise RuntimeError(f"PyServe server failed to start on port {self.config.server.port}")
async def stop(self) -> None:
if self._server_task:
self._server_task.cancel()
try:
await self._server_task
except asyncio.CancelledError:
pass
@asynccontextmanager
async def running_server(config: Config):
"""Context manager for running PyServe server."""
server = PyServeTestServer(config)
await server.start()
try:
yield server
finally:
await server.stop()
@pytest.fixture
def pyserve_port() -> int:
return get_free_port()
@pytest.fixture
def temp_static_dir_with_files(tmp_path):
"""Create temporary static directory with various test files."""
static_dir = tmp_path / "static"
static_dir.mkdir()
# HTML files
(static_dir / "index.html").write_text("<!DOCTYPE html><html><body>Home</body></html>")
(static_dir / "about.html").write_text("<!DOCTYPE html><html><body>About</body></html>")
(static_dir / "docs.html").write_text("<!DOCTYPE html><html><body>Docs</body></html>")
# CSS files
(static_dir / "style.css").write_text("body { margin: 0; }")
# JS files
(static_dir / "app.js").write_text("console.log('Hello');")
# Images (create small binary files)
(static_dir / "logo.png").write_bytes(b'\x89PNG\r\n\x1a\n')
(static_dir / "icon.ico").write_bytes(b'\x00\x00\x01\x00')
# Subdirectories
docs_dir = static_dir / "docs"
docs_dir.mkdir()
(docs_dir / "index.html").write_text("<!DOCTYPE html><html><body>Docs Index</body></html>")
(docs_dir / "guide.html").write_text("<!DOCTYPE html><html><body>Guide</body></html>")
return static_dir
# ============== Integration Tests: Static File Serving ==============
@pytest.mark.asyncio
async def test_static_file_css(pyserve_port, temp_static_dir_with_files):
"""Test serving CSS files with correct content type."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"~*\\.(css)$": {
"root": str(temp_static_dir_with_files),
"cache_control": "public, max-age=3600"
},
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://127.0.0.1:{pyserve_port}/style.css")
assert response.status_code == 200
assert "text/css" in response.headers.get("content-type", "")
assert response.headers.get("cache-control") == "public, max-age=3600"
assert "margin" in response.text
@pytest.mark.asyncio
async def test_static_file_js(pyserve_port, temp_static_dir_with_files):
"""Test serving JavaScript files."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"~*\\.(js)$": {
"root": str(temp_static_dir_with_files),
"cache_control": "public, max-age=31536000"
},
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://127.0.0.1:{pyserve_port}/app.js")
assert response.status_code == 200
assert "javascript" in response.headers.get("content-type", "")
assert response.headers.get("cache-control") == "public, max-age=31536000"
@pytest.mark.asyncio
async def test_static_file_images(pyserve_port, temp_static_dir_with_files):
"""Test serving image files."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"~*\\.(png|ico)$": {
"root": str(temp_static_dir_with_files),
"cache_control": "public, max-age=86400"
},
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://127.0.0.1:{pyserve_port}/logo.png")
assert response.status_code == 200
assert "image/png" in response.headers.get("content-type", "")
# ============== Integration Tests: Health Check ==============
@pytest.mark.asyncio
async def test_health_check_exact_match(pyserve_port, temp_static_dir_with_files):
"""Test exact match health check endpoint."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"=/health": {
"return": "200 OK",
"content_type": "text/plain"
}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://127.0.0.1:{pyserve_port}/health")
assert response.status_code == 200
assert response.text == "OK"
assert "text/plain" in response.headers.get("content-type", "")
@pytest.mark.asyncio
async def test_health_check_not_matched_subpath(pyserve_port, temp_static_dir_with_files):
"""Test that exact match doesn't match subpaths."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"=/health": {
"return": "200 OK",
"content_type": "text/plain"
},
"__default__": {
"return": "404 Not Found"
}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://127.0.0.1:{pyserve_port}/health/check")
assert response.status_code == 404
# ============== Integration Tests: SPA Fallback ==============
@pytest.mark.asyncio
async def test_spa_fallback_any_route(pyserve_port, temp_static_dir_with_files):
"""Test SPA fallback returns index for any route.
Note: When 'root' is in config, _handle_static is called first.
For true SPA fallback behavior, we need to configure it properly
so that spa_fallback kicks in for non-file routes.
"""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"=/health": {"return": "200 OK"},
# For SPA fallback to work properly, we don't include 'root'
# at the same level, as spa_fallback handler uses its own root
"__default__": {
"spa_fallback": True,
"root": str(temp_static_dir_with_files),
"index_file": "index.html"
}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
# With current routing implementation, 'root' in config triggers
# _handle_static which returns 404 for non-existent paths.
# This tests documents the actual behavior.
for path in ["/app", "/app/dashboard", "/users/123", "/settings/profile"]:
response = await client.get(f"http://127.0.0.1:{pyserve_port}{path}")
# Current implementation returns 404 for non-file paths when root is set
# because _handle_static is called before spa_fallback check
assert response.status_code == 404, f"Expected 404 for path {path}"
@pytest.mark.asyncio
async def test_spa_fallback_excludes_api(pyserve_port, temp_static_dir_with_files):
"""Test SPA fallback excludes API routes.
This test verifies that exclude_patterns work correctly in spa_fallback config.
Since 'root' is in config, _handle_static is called first, returning 404
for non-existent paths regardless of exclude_patterns.
"""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"=/health": {"return": "200 OK"},
"__default__": {
"spa_fallback": True,
"root": str(temp_static_dir_with_files),
"index_file": "index.html",
"exclude_patterns": ["/api/", "/admin/"]
}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
# All non-existent paths return 404 when 'root' is in config
# because _handle_static is called before spa_fallback
response_api = await client.get(f"http://127.0.0.1:{pyserve_port}/api/users")
assert response_api.status_code == 404
response_admin = await client.get(f"http://127.0.0.1:{pyserve_port}/admin/dashboard")
assert response_admin.status_code == 404
response_app = await client.get(f"http://127.0.0.1:{pyserve_port}/app/dashboard")
assert response_app.status_code == 404
# ============== Integration Tests: Routing Priority ==============
@pytest.mark.asyncio
async def test_routing_priority_exact_over_regex(pyserve_port, temp_static_dir_with_files):
"""Test that exact match takes priority over regex."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"=/api/status": {
"return": "200 Exact Status",
"content_type": "text/plain"
},
"~^/api/": {
"return": "200 Regex API",
"content_type": "text/plain"
},
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
# Exact match should take priority
response_exact = await client.get(f"http://127.0.0.1:{pyserve_port}/api/status")
assert response_exact.text == "Exact Status"
# Other API routes should use regex
response_regex = await client.get(f"http://127.0.0.1:{pyserve_port}/api/users")
assert response_regex.text == "Regex API"
@pytest.mark.asyncio
async def test_routing_priority_regex_over_default(pyserve_port, temp_static_dir_with_files):
"""Test that regex match takes priority over default."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"~^/api/": {
"return": "200 API",
"content_type": "text/plain"
},
"=/health": {"return": "200 OK"},
"__default__": {
"return": "200 Default",
"content_type": "text/plain"
}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
# API routes should use regex
response_api = await client.get(f"http://127.0.0.1:{pyserve_port}/api/users")
assert response_api.text == "API"
# Other routes should use default
response_other = await client.get(f"http://127.0.0.1:{pyserve_port}/unknown")
assert response_other.text == "Default"
# ============== Integration Tests: Complex Configuration ==============
@pytest.mark.asyncio
async def test_complex_config_multiple_routes(pyserve_port, temp_static_dir_with_files):
"""Test complex configuration with multiple route types."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
# Static files with cache
"~*\\.(js|css)$": {
"root": str(temp_static_dir_with_files),
"cache_control": "public, max-age=31536000",
"headers": ["Access-Control-Allow-Origin: *"]
},
# Exact match for root
"=/": {
"root": str(temp_static_dir_with_files),
"index_file": "index.html"
},
# Health check
"=/health": {
"return": "200 OK",
"content_type": "text/plain"
},
# SPA fallback for everything else
"__default__": {
"spa_fallback": True,
"root": str(temp_static_dir_with_files),
"index_file": "docs.html",
"exclude_patterns": ["/api/", "/admin/"]
}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
# Test root path
response_root = await client.get(f"http://127.0.0.1:{pyserve_port}/")
assert response_root.status_code == 200
assert "Home" in response_root.text
# Test CSS file
response_css = await client.get(f"http://127.0.0.1:{pyserve_port}/style.css")
assert response_css.status_code == 200
assert response_css.headers.get("cache-control") == "public, max-age=31536000"
assert response_css.headers.get("access-control-allow-origin") == "*"
# Test health check
response_health = await client.get(f"http://127.0.0.1:{pyserve_port}/health")
assert response_health.status_code == 200
assert response_health.text == "OK"
# Test SPA fallback - with 'root' in config, non-file paths return 404
response_spa = await client.get(f"http://127.0.0.1:{pyserve_port}/app/route")
assert response_spa.status_code == 404 # root in config means _handle_static returns 404
# Test excluded API route
response_api = await client.get(f"http://127.0.0.1:{pyserve_port}/api/data")
assert response_api.status_code == 404
@pytest.mark.asyncio
async def test_subdirectory_routing(pyserve_port, temp_static_dir_with_files):
"""Test routing to subdirectory files."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"~^/docs": {
"root": str(temp_static_dir_with_files),
"index_file": "index.html"
},
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
# Test docs index
response_index = await client.get(f"http://127.0.0.1:{pyserve_port}/docs/")
assert response_index.status_code == 200
assert "Docs Index" in response_index.text
# Test docs guide
response_guide = await client.get(f"http://127.0.0.1:{pyserve_port}/docs/guide.html")
assert response_guide.status_code == 200
assert "Guide" in response_guide.text
# ============== Integration Tests: Error Cases ==============
@pytest.mark.asyncio
async def test_file_not_found(pyserve_port, temp_static_dir_with_files):
"""Test 404 response for non-existent files."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"~*\\.(css)$": {"root": str(temp_static_dir_with_files)},
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://127.0.0.1:{pyserve_port}/nonexistent.css")
assert response.status_code == 404
@pytest.mark.asyncio
async def test_no_default_route_returns_404(pyserve_port, temp_static_dir_with_files):
"""Test that missing default route returns 404 for unmatched paths."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://127.0.0.1:{pyserve_port}/unknown")
assert response.status_code == 404
# ============== Integration Tests: Headers ==============
@pytest.mark.asyncio
async def test_custom_response_headers(pyserve_port, temp_static_dir_with_files):
"""Test custom response headers."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"~*\\.(js)$": {
"root": str(temp_static_dir_with_files),
"headers": [
"X-Content-Type-Options: nosniff",
"X-Frame-Options: DENY",
"Access-Control-Allow-Origin: https://example.com"
]
},
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://127.0.0.1:{pyserve_port}/app.js")
assert response.status_code == 200
assert response.headers.get("x-content-type-options") == "nosniff"
assert response.headers.get("x-frame-options") == "DENY"
assert response.headers.get("access-control-allow-origin") == "https://example.com"
@pytest.mark.asyncio
async def test_cache_control_headers(pyserve_port, temp_static_dir_with_files):
"""Test cache control headers."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"~*\\.(css)$": {
"root": str(temp_static_dir_with_files),
"cache_control": "public, max-age=3600, immutable"
},
"~*\\.(html)$": {
"root": str(temp_static_dir_with_files),
"cache_control": "no-cache, no-store, must-revalidate"
},
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
# CSS should have long cache
response_css = await client.get(f"http://127.0.0.1:{pyserve_port}/style.css")
assert response_css.headers.get("cache-control") == "public, max-age=3600, immutable"
# HTML should have no cache
response_html = await client.get(f"http://127.0.0.1:{pyserve_port}/index.html")
assert response_html.headers.get("cache-control") == "no-cache, no-store, must-revalidate"
# ============== Integration Tests: Case Sensitivity ==============
@pytest.mark.asyncio
async def test_case_insensitive_file_extensions(pyserve_port, temp_static_dir_with_files):
"""Test case-insensitive file extension matching."""
# Create uppercase extension file
uppercase_css = temp_static_dir_with_files / "UPPER.CSS"
uppercase_css.write_text("body { color: red; }")
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"~*\\.(css)$": {
"root": str(temp_static_dir_with_files),
"cache_control": "public, max-age=3600"
},
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
# Lowercase extension
response_lower = await client.get(f"http://127.0.0.1:{pyserve_port}/style.css")
assert response_lower.status_code == 200
# Uppercase extension in URL should match case-insensitive pattern
response_upper = await client.get(f"http://127.0.0.1:{pyserve_port}/UPPER.CSS")
assert response_upper.status_code == 200
# ============== Integration Tests: Empty/Minimal Configurations ==============
@pytest.mark.asyncio
async def test_minimal_config_health_only(pyserve_port):
"""Test minimal configuration with only health check."""
config = Config(
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
response = await client.get(f"http://127.0.0.1:{pyserve_port}/health")
assert response.status_code == 200
# Other routes should return 404
response_other = await client.get(f"http://127.0.0.1:{pyserve_port}/anything")
assert response_other.status_code == 404
@pytest.mark.asyncio
async def test_only_default_route(pyserve_port, temp_static_dir_with_files):
"""Test configuration with only default route."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"=/health": {"return": "200 OK"},
"__default__": {
"return": "200 Default Response",
"content_type": "text/plain"
}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
# Any route should match default
for path in ["/anything", "/api/test", "/deep/nested/path"]:
response = await client.get(f"http://127.0.0.1:{pyserve_port}{path}")
assert response.status_code == 200
assert response.text == "Default Response"
# ============== Integration Tests: Multiple File Extensions ==============
@pytest.mark.asyncio
async def test_multiple_file_extensions_single_rule(pyserve_port, temp_static_dir_with_files):
"""Test single rule matching multiple file extensions."""
config = Config(
http=HttpConfig(static_dir=str(temp_static_dir_with_files)),
server=ServerConfig(host="127.0.0.1", port=pyserve_port),
logging=LoggingConfig(level="ERROR", console_output=False),
extensions=[
ExtensionConfig(
type="routing",
config={
"regex_locations": {
"~*\\.(js|css|png|ico)$": {
"root": str(temp_static_dir_with_files),
"cache_control": "public, max-age=86400"
},
"=/health": {"return": "200 OK"}
}
}
)
]
)
async with running_server(config):
async with httpx.AsyncClient() as client:
# All these should match and have cache control
files = ["/style.css", "/app.js", "/logo.png", "/icon.ico"]
for file in files:
response = await client.get(f"http://127.0.0.1:{pyserve_port}{file}")
assert response.status_code == 200, f"Failed for {file}"
assert response.headers.get("cache-control") == "public, max-age=86400", f"Cache control missing for {file}"