feat(mcp-docugen): Task 1-3 config, models, auth middleware
- Settings con Pydantic Settings, validazione env obbligatori - Shared models: TemplateVariable/Frontmatter, ImageVariable, TokenUsage, GenerationResult, TemplateSummary, TemplateAsset - ApiKeyAuthMiddleware Bearer token con exempt paths 19 test, tutti passed. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,34 @@
|
|||||||
|
from starlette.middleware.base import BaseHTTPMiddleware
|
||||||
|
from starlette.requests import Request
|
||||||
|
from starlette.responses import JSONResponse
|
||||||
|
from starlette.types import ASGIApp
|
||||||
|
|
||||||
|
|
||||||
|
class ApiKeyAuthMiddleware(BaseHTTPMiddleware):
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
app: ASGIApp,
|
||||||
|
api_key: str,
|
||||||
|
exempt_paths: set[str] | None = None,
|
||||||
|
) -> None:
|
||||||
|
super().__init__(app)
|
||||||
|
self.api_key = api_key
|
||||||
|
self.exempt_paths = exempt_paths or set()
|
||||||
|
|
||||||
|
async def dispatch(self, request: Request, call_next):
|
||||||
|
if request.url.path in self.exempt_paths:
|
||||||
|
return await call_next(request)
|
||||||
|
|
||||||
|
header = request.headers.get("Authorization", "")
|
||||||
|
if not header.startswith("Bearer "):
|
||||||
|
return JSONResponse(
|
||||||
|
status_code=401, content={"error": "invalid_api_key"}
|
||||||
|
)
|
||||||
|
|
||||||
|
provided = header.removeprefix("Bearer ").strip()
|
||||||
|
if provided != self.api_key:
|
||||||
|
return JSONResponse(
|
||||||
|
status_code=401, content={"error": "invalid_api_key"}
|
||||||
|
)
|
||||||
|
|
||||||
|
return await call_next(request)
|
||||||
@@ -0,0 +1,25 @@
|
|||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from pydantic import Field, field_validator
|
||||||
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||||
|
|
||||||
|
|
||||||
|
class Settings(BaseSettings):
|
||||||
|
model_config = SettingsConfigDict(
|
||||||
|
env_file=".env", env_file_encoding="utf-8", extra="ignore"
|
||||||
|
)
|
||||||
|
|
||||||
|
api_key: str = Field(..., min_length=8)
|
||||||
|
openrouter_api_key: str = Field(..., min_length=8)
|
||||||
|
openrouter_base_url: str = "https://openrouter.ai/api/v1"
|
||||||
|
llm_model_default: str = "anthropic/claude-sonnet-4"
|
||||||
|
public_base_url: str = Field(...)
|
||||||
|
data_dir: Path = Path("/data")
|
||||||
|
asset_ttl_days: int = 30
|
||||||
|
max_image_size_mb: int = 10
|
||||||
|
llm_timeout_seconds: int = 60
|
||||||
|
|
||||||
|
@field_validator("public_base_url")
|
||||||
|
@classmethod
|
||||||
|
def strip_trailing_slash(cls, v: str) -> str:
|
||||||
|
return v.rstrip("/")
|
||||||
@@ -0,0 +1,72 @@
|
|||||||
|
import re
|
||||||
|
from datetime import datetime
|
||||||
|
from typing import Literal
|
||||||
|
|
||||||
|
from pydantic import BaseModel, Field, field_validator
|
||||||
|
|
||||||
|
_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
|
||||||
|
_SUPPORTED_MIMES = {"image/png", "image/jpeg", "image/webp"}
|
||||||
|
|
||||||
|
|
||||||
|
class TemplateVariable(BaseModel):
|
||||||
|
name: str = Field(..., min_length=1)
|
||||||
|
type: Literal["string", "image"]
|
||||||
|
|
||||||
|
|
||||||
|
class TemplateFrontmatter(BaseModel):
|
||||||
|
name: str
|
||||||
|
description: str = ""
|
||||||
|
model: str | None = None
|
||||||
|
required_variables: list[TemplateVariable] = Field(default_factory=list)
|
||||||
|
instructions_hint: str | None = None
|
||||||
|
|
||||||
|
@field_validator("name")
|
||||||
|
@classmethod
|
||||||
|
def validate_slug(cls, v: str) -> str:
|
||||||
|
if not _SLUG_RE.match(v):
|
||||||
|
raise ValueError(
|
||||||
|
f"invalid template name '{v}': must match {_SLUG_RE.pattern}"
|
||||||
|
)
|
||||||
|
return v
|
||||||
|
|
||||||
|
|
||||||
|
class ImageVariable(BaseModel):
|
||||||
|
kind: Literal["image"]
|
||||||
|
data_b64: str
|
||||||
|
mime: str
|
||||||
|
|
||||||
|
@field_validator("mime")
|
||||||
|
@classmethod
|
||||||
|
def validate_mime(cls, v: str) -> str:
|
||||||
|
if v not in _SUPPORTED_MIMES:
|
||||||
|
raise ValueError(
|
||||||
|
f"unsupported mime '{v}'; allowed: {sorted(_SUPPORTED_MIMES)}"
|
||||||
|
)
|
||||||
|
return v
|
||||||
|
|
||||||
|
|
||||||
|
class TokenUsage(BaseModel):
|
||||||
|
input: int = 0
|
||||||
|
output: int = 0
|
||||||
|
|
||||||
|
|
||||||
|
class GenerationResult(BaseModel):
|
||||||
|
generation_id: str
|
||||||
|
markdown: str
|
||||||
|
model: str
|
||||||
|
tokens: TokenUsage
|
||||||
|
cost_usd: float
|
||||||
|
ephemeral_assets_urls: list[str] = Field(default_factory=list)
|
||||||
|
ephemeral_expires_at: datetime | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class TemplateSummary(BaseModel):
|
||||||
|
name: str
|
||||||
|
description: str
|
||||||
|
updated_at: datetime
|
||||||
|
|
||||||
|
|
||||||
|
class TemplateAsset(BaseModel):
|
||||||
|
filename: str
|
||||||
|
mime: str
|
||||||
|
size_bytes: int
|
||||||
@@ -0,0 +1,57 @@
|
|||||||
|
import pytest
|
||||||
|
from fastapi import FastAPI
|
||||||
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
|
from mcp_docugen.auth import ApiKeyAuthMiddleware
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def app_with_auth():
|
||||||
|
app = FastAPI()
|
||||||
|
app.add_middleware(
|
||||||
|
ApiKeyAuthMiddleware, api_key="secret", exempt_paths={"/health"}
|
||||||
|
)
|
||||||
|
|
||||||
|
@app.get("/protected")
|
||||||
|
async def protected():
|
||||||
|
return {"ok": True}
|
||||||
|
|
||||||
|
@app.get("/health")
|
||||||
|
async def health():
|
||||||
|
return {"status": "ok"}
|
||||||
|
|
||||||
|
return app
|
||||||
|
|
||||||
|
|
||||||
|
def test_protected_without_header_returns_401(app_with_auth):
|
||||||
|
client = TestClient(app_with_auth)
|
||||||
|
response = client.get("/protected")
|
||||||
|
assert response.status_code == 401
|
||||||
|
assert response.json() == {"error": "invalid_api_key"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_protected_with_wrong_key_returns_401(app_with_auth):
|
||||||
|
client = TestClient(app_with_auth)
|
||||||
|
response = client.get("/protected", headers={"Authorization": "Bearer wrong"})
|
||||||
|
assert response.status_code == 401
|
||||||
|
|
||||||
|
|
||||||
|
def test_protected_with_correct_key_passes(app_with_auth):
|
||||||
|
client = TestClient(app_with_auth)
|
||||||
|
response = client.get("/protected", headers={"Authorization": "Bearer secret"})
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert response.json() == {"ok": True}
|
||||||
|
|
||||||
|
|
||||||
|
def test_health_bypasses_auth(app_with_auth):
|
||||||
|
client = TestClient(app_with_auth)
|
||||||
|
response = client.get("/health")
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
|
||||||
|
def test_malformed_auth_header_returns_401(app_with_auth):
|
||||||
|
client = TestClient(app_with_auth)
|
||||||
|
response = client.get("/protected", headers={"Authorization": "secret"})
|
||||||
|
assert response.status_code == 401
|
||||||
|
response = client.get("/protected", headers={"Authorization": "Basic secret"})
|
||||||
|
assert response.status_code == 401
|
||||||
@@ -0,0 +1,44 @@
|
|||||||
|
import pytest
|
||||||
|
from pydantic import ValidationError
|
||||||
|
|
||||||
|
from mcp_docugen.config import Settings
|
||||||
|
|
||||||
|
|
||||||
|
def test_settings_loads_from_env(monkeypatch, tmp_path):
|
||||||
|
monkeypatch.setenv("API_KEY", "test-api-key")
|
||||||
|
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-test")
|
||||||
|
monkeypatch.setenv("PUBLIC_BASE_URL", "https://mcp.example.com")
|
||||||
|
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
||||||
|
|
||||||
|
settings = Settings()
|
||||||
|
|
||||||
|
assert settings.api_key == "test-api-key"
|
||||||
|
assert settings.openrouter_api_key == "sk-or-test"
|
||||||
|
assert settings.public_base_url == "https://mcp.example.com"
|
||||||
|
assert settings.data_dir == tmp_path
|
||||||
|
assert settings.llm_model_default == "anthropic/claude-sonnet-4"
|
||||||
|
assert settings.asset_ttl_days == 30
|
||||||
|
assert settings.max_image_size_mb == 10
|
||||||
|
assert settings.llm_timeout_seconds == 60
|
||||||
|
assert settings.openrouter_base_url == "https://openrouter.ai/api/v1"
|
||||||
|
|
||||||
|
|
||||||
|
def test_settings_missing_required_fails(monkeypatch, tmp_path):
|
||||||
|
monkeypatch.delenv("API_KEY", raising=False)
|
||||||
|
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
|
||||||
|
monkeypatch.delenv("PUBLIC_BASE_URL", raising=False)
|
||||||
|
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
||||||
|
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
Settings(_env_file=None)
|
||||||
|
|
||||||
|
|
||||||
|
def test_public_base_url_strips_trailing_slash(monkeypatch, tmp_path):
|
||||||
|
monkeypatch.setenv("API_KEY", "k-with-long-enough")
|
||||||
|
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-long-enough")
|
||||||
|
monkeypatch.setenv("PUBLIC_BASE_URL", "https://mcp.example.com/")
|
||||||
|
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
||||||
|
|
||||||
|
settings = Settings()
|
||||||
|
|
||||||
|
assert settings.public_base_url == "https://mcp.example.com"
|
||||||
@@ -0,0 +1,92 @@
|
|||||||
|
import pytest
|
||||||
|
from pydantic import ValidationError
|
||||||
|
|
||||||
|
from mcp_docugen.models import (
|
||||||
|
GenerationResult,
|
||||||
|
ImageVariable,
|
||||||
|
TemplateFrontmatter,
|
||||||
|
TemplateVariable,
|
||||||
|
TokenUsage,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_template_variable_string():
|
||||||
|
v = TemplateVariable(name="cliente", type="string")
|
||||||
|
assert v.name == "cliente"
|
||||||
|
assert v.type == "string"
|
||||||
|
|
||||||
|
|
||||||
|
def test_template_variable_image():
|
||||||
|
v = TemplateVariable(name="foto", type="image")
|
||||||
|
assert v.type == "image"
|
||||||
|
|
||||||
|
|
||||||
|
def test_template_variable_rejects_unknown_type():
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
TemplateVariable(name="x", type="number")
|
||||||
|
|
||||||
|
|
||||||
|
def test_template_variable_rejects_empty_name():
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
TemplateVariable(name="", type="string")
|
||||||
|
|
||||||
|
|
||||||
|
def test_frontmatter_minimal():
|
||||||
|
fm = TemplateFrontmatter(name="fattura", description="Fattura commerciale")
|
||||||
|
assert fm.name == "fattura"
|
||||||
|
assert fm.required_variables == []
|
||||||
|
assert fm.model is None
|
||||||
|
assert fm.instructions_hint is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_frontmatter_with_variables():
|
||||||
|
fm = TemplateFrontmatter(
|
||||||
|
name="fattura",
|
||||||
|
description="...",
|
||||||
|
model="anthropic/claude-sonnet-4",
|
||||||
|
required_variables=[
|
||||||
|
{"name": "cliente", "type": "string"},
|
||||||
|
{"name": "foto", "type": "image"},
|
||||||
|
],
|
||||||
|
instructions_hint="tono formale",
|
||||||
|
)
|
||||||
|
assert len(fm.required_variables) == 2
|
||||||
|
assert fm.required_variables[1].type == "image"
|
||||||
|
|
||||||
|
|
||||||
|
def test_frontmatter_name_slug_rules():
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
TemplateFrontmatter(name="Fattura Commerciale", description="x")
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
TemplateFrontmatter(name="../escape", description="x")
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
TemplateFrontmatter(name="-starts-with-dash", description="x")
|
||||||
|
|
||||||
|
|
||||||
|
def test_image_variable():
|
||||||
|
img = ImageVariable(kind="image", data_b64="aGVsbG8=", mime="image/png")
|
||||||
|
assert img.mime == "image/png"
|
||||||
|
|
||||||
|
|
||||||
|
def test_image_variable_rejects_unsupported_mime():
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
ImageVariable(kind="image", data_b64="aGVsbG8=", mime="image/bmp")
|
||||||
|
|
||||||
|
|
||||||
|
def test_image_variable_rejects_wrong_kind():
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
ImageVariable(kind="file", data_b64="aGVsbG8=", mime="image/png")
|
||||||
|
|
||||||
|
|
||||||
|
def test_generation_result_shape():
|
||||||
|
result = GenerationResult(
|
||||||
|
generation_id="abc-123",
|
||||||
|
markdown="# hello",
|
||||||
|
model="anthropic/claude-sonnet-4",
|
||||||
|
tokens=TokenUsage(input=100, output=200),
|
||||||
|
cost_usd=0.01,
|
||||||
|
ephemeral_assets_urls=[],
|
||||||
|
ephemeral_expires_at=None,
|
||||||
|
)
|
||||||
|
assert result.generation_id == "abc-123"
|
||||||
|
assert result.tokens.input == 100
|
||||||
Reference in New Issue
Block a user