94 lines
3.8 KiB
Python
94 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from fastapi import HTTPException
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "backend"))
|
|
|
|
from app import main # noqa: E402
|
|
from app.services.storage import WorkspaceStore, write_json # noqa: E402
|
|
from app.settings import ProviderConfig, ProviderModel, Settings # noqa: E402
|
|
|
|
|
|
BACKEND = ROOT / "backend"
|
|
|
|
|
|
def settings(root: Path) -> Settings:
|
|
provider = ProviderConfig("test", "Test", "https://example.invalid/v1", "test-key", (ProviderModel("test-model"),))
|
|
return Settings(
|
|
task_root=root / "tasks",
|
|
conversation_root=root / "conversations",
|
|
library_root=BACKEND / "cdsl_library",
|
|
engine_root=BACKEND / "engine" / "cdsl_engine",
|
|
llm_base_url=provider.base_url,
|
|
llm_api_key=provider.api_key,
|
|
llm_model="test-model",
|
|
llm_timeout_s=1,
|
|
default_provider_id="test",
|
|
providers=(provider,),
|
|
)
|
|
|
|
|
|
class AutonomousArtifactTests(unittest.TestCase):
|
|
def test_checkpoint_exposes_only_its_preview_until_final_publication(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
store = WorkspaceStore(settings(Path(temporary)))
|
|
task = store.ensure_task(None, "artifact visibility")
|
|
task_id = str(task["task_id"])
|
|
revision_id, revision_dir = store.next_revision(task_id)
|
|
paths = {
|
|
"cdsl_path": f"revisions/{revision_id}/model.cdsl.json",
|
|
"step_path": f"revisions/{revision_id}/model.step",
|
|
"glb_path": f"revisions/{revision_id}/model.glb",
|
|
"report_path": f"revisions/{revision_id}/rebuild-report.json",
|
|
}
|
|
for relative in paths.values():
|
|
target = store.artifact_path(task_id, relative)
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(b"artifact")
|
|
candidate = store.task_dir(task_id) / "candidates" / "candidate_1234567890ab" / "fragment.json"
|
|
candidate.parent.mkdir(parents=True, exist_ok=True)
|
|
write_json(candidate, {"secret": "candidate-only"})
|
|
store.write_requirements_document(task_id, "# Frozen")
|
|
store.update_task(task_id, {
|
|
"revision_id": revision_id,
|
|
"status": "success",
|
|
**paths,
|
|
"visibility": "checkpoint",
|
|
"parent_revision_id": "",
|
|
"branch_id": "main",
|
|
})
|
|
store.start_generation(task_id, request="artifact visibility")
|
|
|
|
previous_store = main.store
|
|
main.store = store
|
|
try:
|
|
preview = asyncio.run(main.read_artifact(task_id, paths["glb_path"]))
|
|
self.assertEqual(preview.status_code, 200)
|
|
self.assertEqual(preview.headers["content-disposition"], "inline")
|
|
for forbidden in (paths["cdsl_path"], "requirements.md", "candidates/candidate_1234567890ab/fragment.json"):
|
|
with self.assertRaises(HTTPException) as error:
|
|
asyncio.run(main.read_artifact(task_id, forbidden))
|
|
self.assertEqual(error.exception.status_code, 403)
|
|
|
|
store.finish_generation(task_id, lifecycle="completed")
|
|
for released in paths.values():
|
|
delivery = asyncio.run(main.read_artifact(task_id, released))
|
|
self.assertEqual(delivery.status_code, 200)
|
|
with self.assertRaises(HTTPException) as error:
|
|
asyncio.run(main.read_artifact(task_id, "requirements.md"))
|
|
self.assertEqual(error.exception.status_code, 403)
|
|
finally:
|
|
main.store = previous_store
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|