Adriano
e8705dcd0b
- Renderer: orchestratore generate() — validazione strict variabili,
materializzazione image vars come asset effimeri su disco + URL rewrite,
asset paths template da ./assets/X -> {PUBLIC_BASE_URL}/assets/<t>/X,
integrazione LLM error -> record success=0
- FastAPI sub-app: GET /health (no auth), /assets/{t}/{f} (auth+traversal check),
/generated/{gen_id}/{f} (410 su scaduto o mancante)
- FastMCP server con 6 tool: template_create/update/delete/list/get,
document_generate. Tools esposti anche via mcp.tools dict per test.
- main.build_app() compone http_app + FastMCP mount su /mcp + auth middleware
+ lifespan cleanup task TTL (24h). run() entry point per script console.
68 test passed. Build Docker arca-mcp-docugen:dev verificata,
/health endpoint risponde correttamente nel container.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
123 lines
3.6 KiB
Python
123 lines
3.6 KiB
Python
import base64
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from mcp_docugen.auth import ApiKeyAuthMiddleware
|
|
from mcp_docugen.generation_store import EphemeralAssetRecord, GenerationStore
|
|
from mcp_docugen.http_routes import build_http_app
|
|
from mcp_docugen.models import TemplateFrontmatter
|
|
from mcp_docugen.template_store import TemplateStore
|
|
|
|
|
|
@pytest.fixture
|
|
async def env(tmp_path):
|
|
template_store = TemplateStore(base_dir=tmp_path / "templates")
|
|
generation_store = GenerationStore(
|
|
db_path=tmp_path / "gen.db",
|
|
generated_dir=tmp_path / "generated",
|
|
)
|
|
await generation_store.init()
|
|
|
|
inner = build_http_app(template_store, generation_store)
|
|
app = FastAPI()
|
|
app.add_middleware(
|
|
ApiKeyAuthMiddleware, api_key="test-key", exempt_paths={"/health"}
|
|
)
|
|
app.mount("/", inner)
|
|
return TestClient(app), template_store, generation_store, tmp_path
|
|
|
|
|
|
def _headers():
|
|
return {"Authorization": "Bearer test-key"}
|
|
|
|
|
|
def test_health_no_auth(env):
|
|
tc, *_ = env
|
|
r = tc.get("/health")
|
|
assert r.status_code == 200
|
|
assert r.json() == {"status": "ok"}
|
|
|
|
|
|
async def test_assets_serve_existing_file(env):
|
|
tc, template_store, _, _ = env
|
|
fm = TemplateFrontmatter(name="brand", description="x")
|
|
assets = [
|
|
{
|
|
"filename": "logo.png",
|
|
"data_b64": base64.b64encode(b"\x89PNG").decode(),
|
|
"mime": "image/png",
|
|
}
|
|
]
|
|
await template_store.create(
|
|
name="brand", frontmatter=fm, body="b", assets=assets
|
|
)
|
|
|
|
r = tc.get("/assets/brand/logo.png", headers=_headers())
|
|
assert r.status_code == 200
|
|
assert r.content == b"\x89PNG"
|
|
|
|
|
|
def test_assets_require_auth(env):
|
|
tc, *_ = env
|
|
r = tc.get("/assets/brand/logo.png")
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_assets_path_traversal_rejected(env):
|
|
tc, *_ = env
|
|
# Starlette decodifica %2F in / prima del routing, quindi la path
|
|
# non matcha /assets/{template}/{filename} e torna 404. 400/404 entrambi safe.
|
|
r = tc.get("/assets/brand/..%2Fevil.png", headers=_headers())
|
|
assert r.status_code in (400, 404)
|
|
|
|
|
|
def test_assets_missing_returns_404(env):
|
|
tc, *_ = env
|
|
r = tc.get("/assets/brand/missing.png", headers=_headers())
|
|
assert r.status_code == 404
|
|
|
|
|
|
async def test_generated_fresh_returns_file(env):
|
|
tc, _, generation_store, tmp_path = env
|
|
gen_dir = tmp_path / "generated" / "g-1"
|
|
gen_dir.mkdir(parents=True)
|
|
(gen_dir / "foto.png").write_bytes(b"image-bytes")
|
|
await generation_store.register_ephemeral_asset(
|
|
EphemeralAssetRecord(
|
|
generation_id="g-1",
|
|
var_name="foto",
|
|
file_path=str(gen_dir / "foto.png"),
|
|
mime="image/png",
|
|
ttl_days=30,
|
|
)
|
|
)
|
|
r = tc.get("/generated/g-1/foto.png", headers=_headers())
|
|
assert r.status_code == 200
|
|
assert r.content == b"image-bytes"
|
|
|
|
|
|
async def test_generated_expired_returns_410(env):
|
|
tc, _, generation_store, tmp_path = env
|
|
gen_dir = tmp_path / "generated" / "g-old"
|
|
gen_dir.mkdir(parents=True)
|
|
(gen_dir / "foto.png").write_bytes(b"x")
|
|
await generation_store.register_ephemeral_asset(
|
|
EphemeralAssetRecord(
|
|
generation_id="g-old",
|
|
var_name="foto",
|
|
file_path=str(gen_dir / "foto.png"),
|
|
mime="image/png",
|
|
ttl_days=-1,
|
|
)
|
|
)
|
|
r = tc.get("/generated/g-old/foto.png", headers=_headers())
|
|
assert r.status_code == 410
|
|
|
|
|
|
def test_generated_unknown_returns_410(env):
|
|
tc, *_ = env
|
|
r = tc.get("/generated/unknown-id/file.png", headers=_headers())
|
|
assert r.status_code == 410
|