Robert Helewka
4a3c14d4af
Add comprehensive rule documentation for AI-assisted development covering authentication surfaces, outbound-call safety invariants, and other project conventions to guide Claude's understanding of critical system behaviors.
118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
"""
|
|
API surface tests — owner enforcement and route registration order.
|
|
|
|
The app is exercised without its lifespan: auth runs before any handler,
|
|
so a 503 ("Gateway not initialized") proves the caller was accepted as
|
|
owner. Auth internals (JWT/PAT resolution) are covered in test_auth.py;
|
|
here we assert the routers are gated and the routes register in the right
|
|
order. In dev-owner mode (SSO disabled) a tokenless request is the owner.
|
|
"""
|
|
|
|
import httpx
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
from sqlalchemy.pool import StaticPool
|
|
from starlette.routing import Match
|
|
|
|
import db.database as dbmod
|
|
import main
|
|
from config import get_settings
|
|
from db.database import Base
|
|
|
|
|
|
@pytest.fixture
|
|
async def mem_db(monkeypatch):
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
poolclass=StaticPool,
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
monkeypatch.setattr(dbmod, "_engine", engine)
|
|
monkeypatch.setattr(dbmod, "_session_factory", factory)
|
|
yield factory
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
def dev_owner(monkeypatch):
|
|
"""SSO disabled — every request resolves to the dev owner."""
|
|
monkeypatch.setattr(get_settings().casdoor, "enabled", False)
|
|
|
|
|
|
@pytest.fixture
|
|
def sso_enabled(monkeypatch):
|
|
"""SSO enabled with no credentials supplied → 401 on protected routes."""
|
|
monkeypatch.setattr(get_settings().casdoor, "enabled", True)
|
|
monkeypatch.setattr(get_settings().casdoor, "endpoint", "https://id.example.test")
|
|
monkeypatch.setattr(get_settings(), "owner_name", "owner@example.test")
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
class TestOwnerGate:
|
|
async def test_dev_owner_reaches_handler(self, dev_owner, mem_db, client):
|
|
resp = await client.get("/api/v1/calls/active")
|
|
assert resp.status_code == 503 # dev-owner accepted; handler 503s (no lifespan)
|
|
|
|
async def test_sso_missing_credentials_rejected(self, sso_enabled, mem_db, client):
|
|
resp = await client.get("/api/v1/calls/active")
|
|
assert resp.status_code == 401
|
|
assert resp.headers["www-authenticate"] == "Bearer"
|
|
|
|
async def test_all_api_routers_protected(self, sso_enabled, mem_db, client):
|
|
for path in (
|
|
"/api/v1/calls/active",
|
|
"/api/v1/call-flows/",
|
|
"/api/v1/devices/",
|
|
"/api/v1/routing/rules",
|
|
"/api/v1/calls/history",
|
|
"/api/v1/tokens",
|
|
):
|
|
resp = await client.get(path)
|
|
assert resp.status_code == 401, path
|
|
|
|
async def test_auth_routes_are_public(self, sso_enabled, mem_db, client):
|
|
"""The OIDC endpoints must be reachable without a token."""
|
|
# /auth/me with no token → 401 (not 403); /auth/login → redirect to Casdoor
|
|
resp = await client.get("/auth/login", follow_redirects=False)
|
|
assert resp.status_code in (302, 307)
|
|
|
|
|
|
class TestRouteOrder:
|
|
def _resolve(self, path: str):
|
|
scope = {
|
|
"type": "http",
|
|
"method": "GET",
|
|
"path": path,
|
|
"root_path": "",
|
|
"query_string": b"",
|
|
"headers": [],
|
|
}
|
|
for route in main.app.router.routes:
|
|
match, _ = route.matches(scope)
|
|
if match == Match.FULL:
|
|
return route
|
|
return None
|
|
|
|
def test_history_not_shadowed_by_call_id(self):
|
|
route = self._resolve("/api/v1/calls/history")
|
|
assert route is not None
|
|
assert route.endpoint.__name__ == "list_history"
|
|
|
|
def test_call_id_still_matches(self):
|
|
route = self._resolve("/api/v1/calls/call_abc123")
|
|
assert route is not None
|
|
assert route.endpoint.__name__ == "get_call"
|
|
|
|
def test_mcp_mounted(self):
|
|
mounted = [getattr(r, "path", "") for r in main.app.router.routes]
|
|
assert "/mcp" in mounted
|