Adriano
9e80a20063
- TemplateStore: CRUD filesystem + asset dir, frontmatter YAML roundtrip, path traversal rejection - OpenRouterClient: async httpx con retry backoff esponenziale (5xx, 429, timeout), no-retry su 4xx, parse usage/cost - GenerationStore: SQLite aiosqlite con schema generations + ephemeral_assets, cleanup TTL, stats aggregate Root pyproject aggiornato con respx + pytest-cov dev deps. 19 + 11 + 9 + 6 = 45 test totali, tutti passed. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
116 lines
3.0 KiB
Python
116 lines
3.0 KiB
Python
import pytest
|
|
|
|
from mcp_docugen.generation_store import (
|
|
EphemeralAssetRecord,
|
|
GenerationRecord,
|
|
GenerationStore,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
async def store(tmp_path):
|
|
s = GenerationStore(
|
|
db_path=tmp_path / "gen.db", generated_dir=tmp_path / "generated"
|
|
)
|
|
await s.init()
|
|
return s
|
|
|
|
|
|
async def test_record_success_generation(store):
|
|
await store.record_generation(
|
|
GenerationRecord(
|
|
id="g-1",
|
|
template_name="fattura",
|
|
model="m",
|
|
tokens_in=100,
|
|
tokens_out=200,
|
|
cost_usd=0.01,
|
|
success=True,
|
|
error_msg=None,
|
|
)
|
|
)
|
|
stats = await store.get_stats()
|
|
assert stats["total"] == 1
|
|
assert stats["success"] == 1
|
|
assert stats["failed"] == 0
|
|
|
|
|
|
async def test_record_failed_generation(store):
|
|
await store.record_generation(
|
|
GenerationRecord(
|
|
id="g-2",
|
|
template_name="fattura",
|
|
model="m",
|
|
tokens_in=0,
|
|
tokens_out=0,
|
|
cost_usd=0.0,
|
|
success=False,
|
|
error_msg="LLMTimeout",
|
|
)
|
|
)
|
|
stats = await store.get_stats()
|
|
assert stats["failed"] == 1
|
|
|
|
|
|
async def test_register_ephemeral_asset(store, tmp_path):
|
|
asset_file = tmp_path / "generated" / "g-1" / "foto.png"
|
|
asset_file.parent.mkdir(parents=True)
|
|
asset_file.write_bytes(b"png-bytes")
|
|
|
|
await store.register_ephemeral_asset(
|
|
EphemeralAssetRecord(
|
|
generation_id="g-1",
|
|
var_name="foto",
|
|
file_path=str(asset_file),
|
|
mime="image/png",
|
|
ttl_days=30,
|
|
)
|
|
)
|
|
asset = await store.get_ephemeral_asset("g-1", "foto.png")
|
|
assert asset is not None
|
|
assert asset.mime == "image/png"
|
|
|
|
|
|
async def test_get_ephemeral_asset_returns_none_if_missing(store):
|
|
asset = await store.get_ephemeral_asset("nope", "foo.png")
|
|
assert asset is None
|
|
|
|
|
|
async def test_cleanup_expired_removes_records_and_files(store, tmp_path):
|
|
asset_file = tmp_path / "generated" / "g-old" / "foto.png"
|
|
asset_file.parent.mkdir(parents=True)
|
|
asset_file.write_bytes(b"bytes")
|
|
|
|
await store.register_ephemeral_asset(
|
|
EphemeralAssetRecord(
|
|
generation_id="g-old",
|
|
var_name="foto",
|
|
file_path=str(asset_file),
|
|
mime="image/png",
|
|
ttl_days=-1,
|
|
)
|
|
)
|
|
removed = await store.cleanup_expired()
|
|
assert removed == 1
|
|
assert not asset_file.exists()
|
|
assert await store.get_ephemeral_asset("g-old", "foto.png") is None
|
|
|
|
|
|
async def test_ephemeral_asset_expired_flag(store, tmp_path):
|
|
f = tmp_path / "generated" / "g-e" / "img.png"
|
|
f.parent.mkdir(parents=True)
|
|
f.write_bytes(b"x")
|
|
|
|
await store.register_ephemeral_asset(
|
|
EphemeralAssetRecord(
|
|
generation_id="g-e",
|
|
var_name="img",
|
|
file_path=str(f),
|
|
mime="image/png",
|
|
ttl_days=-1,
|
|
)
|
|
)
|
|
asset = await store.get_ephemeral_asset("g-e", "img.png")
|
|
assert asset is not None
|
|
assert asset.is_expired is True
|