Artur Mukhamadiev
e65fbf4b67
FastAPI app, upstream Confluence/model adapters, authoritative history, rootless container lifecycle, artifact storage and downloads, fake peers under backend/dev, tests under tests/backend. Root pytest.ini deselects the live marker by default; requirements gain the backend dependencies.
368 lines
14 KiB
Python
368 lines
14 KiB
Python
"""Unit and integration tests for FastAPI application, HTTP contracts, and security."""
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
import pytest
|
|
import httpx
|
|
from httpx import ASGITransport
|
|
|
|
from backend.app import create_app, SESSION_COOKIE_NAME, CSP_POLICY
|
|
from backend.artifacts import ArtifactStore
|
|
from backend.confluence import ConfluenceClient
|
|
from backend.containers import FakeContainerManager
|
|
from backend.dev.fake_peer import ScriptedContainerPeer
|
|
from backend.model import FakeModelAdapter
|
|
from backend.settings import Settings
|
|
|
|
|
|
from tests.backend.conftest import make_test_confluence_client_factory
|
|
|
|
|
|
@pytest.fixture
|
|
def test_app_env(tmp_path: Path):
|
|
settings = Settings(
|
|
approved_confluence_origins=["https://approved.example.com"],
|
|
query_timeout_seconds=30.0,
|
|
cleanup_timeout_seconds=5.0,
|
|
)
|
|
store = ArtifactStore(tmp_path / "artifacts")
|
|
container_mgr = FakeContainerManager(lambda: ScriptedContainerPeer(scenario="standard"))
|
|
model_adapter = FakeModelAdapter()
|
|
conf_factory = make_test_confluence_client_factory()
|
|
|
|
app = create_app(
|
|
settings=settings,
|
|
container_manager=container_mgr,
|
|
artifact_store=store,
|
|
model_adapter=model_adapter,
|
|
confluence_client_factory=conf_factory,
|
|
)
|
|
return app, store, container_mgr
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_root_endpoint(test_app_env):
|
|
app, _, _ = test_app_env
|
|
transport = ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
resp = await client.get("/")
|
|
assert resp.status_code == 200
|
|
assert SESSION_COOKIE_NAME in resp.cookies
|
|
cookie = resp.cookies[SESSION_COOKIE_NAME]
|
|
assert len(cookie) >= 16
|
|
# Security headers
|
|
assert resp.headers.get("Content-Security-Policy") == CSP_POLICY
|
|
assert resp.headers.get("Referrer-Policy") == "no-referrer"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_verify_origin_enforcement(test_app_env):
|
|
app, _, _ = test_app_env
|
|
transport = ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
# 1. Missing Origin header -> 403 origin_denied
|
|
resp = await client.post(
|
|
"/api/v1/auth/verify",
|
|
json={"url": "https://approved.example.com", "pat": "valid-pat"},
|
|
)
|
|
assert resp.status_code == 403
|
|
data = resp.json()
|
|
assert data["error"]["code"] == "origin_denied"
|
|
|
|
# 2. Mismatched Origin header -> 403 origin_denied
|
|
resp = await client.post(
|
|
"/api/v1/auth/verify",
|
|
json={"url": "https://approved.example.com", "pat": "valid-pat"},
|
|
headers={"Origin": "https://evil.com"},
|
|
)
|
|
assert resp.status_code == 403
|
|
data = resp.json()
|
|
assert data["error"]["code"] == "origin_denied"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_verify_success_and_failures(test_app_env):
|
|
app, _, _ = test_app_env
|
|
transport = ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
headers = {"Origin": "http://testserver"}
|
|
|
|
# 1. Valid credentials
|
|
resp = await client.post(
|
|
"/api/v1/auth/verify",
|
|
json={"url": "https://approved.example.com", "pat": "valid-pat"},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"valid": True}
|
|
assert resp.headers.get("Cache-Control") == "no-store"
|
|
assert SESSION_COOKIE_NAME in resp.cookies
|
|
|
|
# 2. Invalid PAT -> 403 confluence_auth_failed
|
|
resp = await client.post(
|
|
"/api/v1/auth/verify",
|
|
json={"url": "https://approved.example.com", "pat": "wrong-pat"},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 403
|
|
assert resp.json()["error"]["code"] == "confluence_auth_failed"
|
|
|
|
# 3. Disapproved Confluence URL -> 403 destination_denied
|
|
resp = await client.post(
|
|
"/api/v1/auth/verify",
|
|
json={"url": "https://unapproved.example.com", "pat": "valid-pat"},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 403
|
|
assert resp.json()["error"]["code"] == "destination_denied"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_and_artifact_download(test_app_env):
|
|
app, store, container_mgr = test_app_env
|
|
transport = ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
headers = {"Origin": "http://testserver"}
|
|
|
|
# 1. Execute query
|
|
payload = {
|
|
"prompt": "How do I deploy service X?",
|
|
"credentials": {
|
|
"url": "https://approved.example.com",
|
|
"pat": "valid-pat",
|
|
},
|
|
}
|
|
resp = await client.post("/api/v1/query", json=payload, headers=headers)
|
|
assert resp.status_code == 200
|
|
assert resp.headers.get("Cache-Control") == "no-store"
|
|
assert SESSION_COOKIE_NAME in client.cookies
|
|
|
|
res_data = resp.json()
|
|
assert res_data["session_id"]
|
|
assert "Deployment Guide" in res_data["markdown"]
|
|
assert len(res_data["pages_accessed"]) == 1
|
|
assert len(res_data["artifacts"]) == 1
|
|
|
|
artifact_meta = res_data["artifacts"][0]
|
|
aid = artifact_meta["id"]
|
|
assert artifact_meta["name"] == "checklist.md"
|
|
assert artifact_meta["size_bytes"] == 32
|
|
|
|
# 2. Download artifact with same session cookie
|
|
dl_resp = await client.get(f"/api/v1/artifacts/{aid}")
|
|
assert dl_resp.status_code == 200
|
|
assert dl_resp.headers.get("Cache-Control") == "no-store"
|
|
assert dl_resp.headers.get("X-Content-Type-Options") == "nosniff"
|
|
assert "attachment" in dl_resp.headers.get("Content-Disposition", "")
|
|
assert dl_resp.content == b"# Checklist\n\n- Deploy service X\n"
|
|
|
|
# 3. Download with wrong session cookie -> 404
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as other_client:
|
|
# New client has no or different session cookie
|
|
other_resp = await other_client.get(f"/api/v1/artifacts/{aid}")
|
|
assert other_resp.status_code == 404
|
|
assert other_resp.json()["error"]["code"] == "artifact_not_found"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_framework_validation_error_sanitization(test_app_env):
|
|
app, _, _ = test_app_env
|
|
transport = ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
headers = {"Origin": "http://testserver"}
|
|
# Send payload missing required 'prompt' field and containing a fake PAT
|
|
bad_payload = {"credentials": {"url": "https://approved.example.com", "pat": "secret-secret-token"}}
|
|
resp = await client.post("/api/v1/query", json=bad_payload, headers=headers)
|
|
assert resp.status_code == 400
|
|
data = resp.json()
|
|
assert data["error"]["code"] == "invalid_input"
|
|
# Secret token must NEVER be echoed in error response!
|
|
assert "secret-secret-token" not in str(data)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_query_origin_enforcement(test_app_env):
|
|
app, _, _ = test_app_env
|
|
transport = ASGITransport(app=app)
|
|
payload = {
|
|
"prompt": "Test query",
|
|
"credentials": {"url": "https://approved.example.com", "pat": "valid-pat"},
|
|
}
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
# 1. Missing Origin header
|
|
resp = await client.post("/api/v1/query", json=payload)
|
|
assert resp.status_code == 403
|
|
assert resp.json()["error"]["code"] == "origin_denied"
|
|
assert resp.headers.get("Cache-Control") == "no-store"
|
|
|
|
# 2. Mismatched Origin header
|
|
resp = await client.post(
|
|
"/api/v1/query",
|
|
json=payload,
|
|
headers={"Origin": "https://malicious.org"},
|
|
)
|
|
assert resp.status_code == 403
|
|
assert resp.json()["error"]["code"] == "origin_denied"
|
|
assert resp.headers.get("Cache-Control") == "no-store"
|
|
|
|
# 3. Scheme mismatch (https Origin for http request)
|
|
resp = await client.post(
|
|
"/api/v1/query",
|
|
json=payload,
|
|
headers={"Origin": "https://testserver"},
|
|
)
|
|
assert resp.status_code == 403
|
|
assert resp.json()["error"]["code"] == "origin_denied"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_body_size_limits_and_chunked_streaming(test_app_env):
|
|
app, _, _ = test_app_env
|
|
transport = ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
# 1. Malformed Content-Length -> 400 invalid_input
|
|
resp = await client.post(
|
|
"/api/v1/auth/verify",
|
|
content=b"{}",
|
|
headers={"Origin": "http://testserver", "Content-Length": "not-an-int"},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert resp.json()["error"]["code"] == "invalid_input"
|
|
assert resp.headers.get("Cache-Control") == "no-store"
|
|
|
|
# 2. Declared Content-Length exceeding 128 KiB -> 413 request_too_large
|
|
resp = await client.post(
|
|
"/api/v1/auth/verify",
|
|
content=b"{}",
|
|
headers={"Origin": "http://testserver", "Content-Length": str(129 * 1024)},
|
|
)
|
|
assert resp.status_code == 413
|
|
assert resp.json()["error"]["code"] == "request_too_large"
|
|
assert resp.headers.get("Cache-Control") == "no-store"
|
|
|
|
# 3. Chunked/streamed body exceeding limit -> 413 request_too_large
|
|
async def oversized_stream():
|
|
chunk = b"x" * 1024
|
|
for _ in range(129): # 129 KiB > 128 KiB
|
|
yield chunk
|
|
|
|
resp = await client.post(
|
|
"/api/v1/auth/verify",
|
|
content=oversized_stream(),
|
|
headers={"Origin": "http://testserver", "Content-Type": "application/json"},
|
|
)
|
|
assert resp.status_code == 413
|
|
assert resp.json()["error"]["code"] == "request_too_large"
|
|
assert resp.headers.get("Cache-Control") == "no-store"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unexpected_fields_rejection(test_app_env):
|
|
app, _, _ = test_app_env
|
|
transport = ASGITransport(app=app)
|
|
headers = {"Origin": "http://testserver"}
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
# 1. Extra field in auth verify payload
|
|
resp = await client.post(
|
|
"/api/v1/auth/verify",
|
|
json={"url": "https://approved.example.com", "pat": "valid-pat", "unexpected": "bad"},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 400
|
|
assert resp.json()["error"]["code"] == "invalid_input"
|
|
|
|
# 2. Extra field in query payload
|
|
resp = await client.post(
|
|
"/api/v1/query",
|
|
json={
|
|
"prompt": "test",
|
|
"credentials": {"url": "https://approved.example.com", "pat": "valid-pat"},
|
|
"extra_key": 123,
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 400
|
|
assert resp.json()["error"]["code"] == "invalid_input"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_route_404_and_envelope_codes(test_app_env):
|
|
app, _, _ = test_app_env
|
|
transport = ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
# Unknown API route maps to invalid_input (not artifact_not_found)
|
|
resp = await client.get("/api/v1/nonexistent")
|
|
assert resp.status_code == 404
|
|
assert resp.json()["error"]["code"] == "invalid_input"
|
|
assert resp.headers.get("Cache-Control") == "no-store"
|
|
|
|
# Unknown artifact maps to artifact_not_found
|
|
resp = await client.get("/api/v1/artifacts/missing-id")
|
|
assert resp.status_code == 404
|
|
assert resp.json()["error"]["code"] == "artifact_not_found"
|
|
assert resp.headers.get("Cache-Control") == "no-store"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rfc6266_non_ascii_filename_download(test_app_env):
|
|
app, store, _ = test_app_env
|
|
transport = ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
# 1. Obtain a server-issued session
|
|
root_resp = await client.get("/")
|
|
assert root_resp.status_code == 200
|
|
session_id = root_resp.cookies[SESSION_COOKIE_NAME]
|
|
|
|
# 2. Stage and commit an artifact with non-ASCII Unicode characters (e.g. 'clé')
|
|
import base64
|
|
staging = store.create_staging_session("q-unicode")
|
|
decision, _ = staging.handle_begin("t-1", "rapport_clé_2026.pdf", 14)
|
|
assert decision == "accept"
|
|
b64_data = base64.b64encode(b"PDF Content OK").decode("utf-8")
|
|
staging.handle_chunk("t-1", 0, b64_data)
|
|
staging.handle_end("t-1", 14, 1)
|
|
committed = staging.commit(session_id=session_id)
|
|
assert len(committed) == 1
|
|
art_id = committed[0]["id"]
|
|
|
|
# 3. Download the artifact and verify RFC 6266 Content-Disposition header
|
|
resp = await client.get(f"/api/v1/artifacts/{art_id}")
|
|
assert resp.status_code == 200
|
|
assert resp.headers.get("Cache-Control") == "no-store"
|
|
assert resp.headers.get("X-Content-Type-Options") == "nosniff"
|
|
|
|
cd = resp.headers.get("Content-Disposition", "")
|
|
# Must contain ASCII fallback
|
|
assert 'filename="rapport_cl__2026.pdf"' in cd
|
|
# Must contain UTF-8 percent-encoded filename
|
|
assert "filename*=UTF-8''rapport_cl%C3%A9_2026.pdf" in cd
|
|
assert resp.content == b"PDF Content OK"
|
|
|
|
|
|
def test_dev_mode_wires_fake_container_manager(tmp_path: Path):
|
|
settings = Settings(dev_mode=True, artifact_storage_dir=tmp_path)
|
|
app = create_app(settings=settings)
|
|
assert app is not None
|
|
|
|
|
|
def test_production_model_validation_failures(tmp_path: Path):
|
|
# 1. Unknown provider
|
|
settings_bad_provider = Settings(
|
|
dev_mode=False,
|
|
model_provider="unsupported_provider",
|
|
artifact_storage_dir=tmp_path,
|
|
)
|
|
with pytest.raises(RuntimeError, match="Unknown model_provider"):
|
|
create_app(settings=settings_bad_provider)
|
|
|
|
# 2. Missing OpenAI key in production mode
|
|
settings_no_key = Settings(
|
|
dev_mode=False,
|
|
model_provider="openai",
|
|
model_api_key="",
|
|
artifact_storage_dir=tmp_path,
|
|
)
|
|
with pytest.raises(RuntimeError, match="Missing model_api_key"):
|
|
create_app(settings=settings_no_key)
|
|
|