feat: artifact storage module with per-session CRUD
This commit is contained in:
@@ -0,0 +1,48 @@
|
|||||||
|
"""Artifact storage module — per-session CRUD operations on the filesystem."""
|
||||||
|
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
ARTIFACTS_DIR = Path("/app/artifacts")
|
||||||
|
|
||||||
|
|
||||||
|
def list_artifacts(session_id: str) -> list[dict]:
|
||||||
|
"""Return sorted list of artifact dicts for a session.
|
||||||
|
|
||||||
|
Each dict contains: name, session_id, size.
|
||||||
|
Returns [] if the session directory does not exist.
|
||||||
|
"""
|
||||||
|
session_dir = ARTIFACTS_DIR / session_id
|
||||||
|
if not session_dir.exists():
|
||||||
|
return []
|
||||||
|
return sorted(
|
||||||
|
[
|
||||||
|
{
|
||||||
|
"name": f.name,
|
||||||
|
"session_id": session_id,
|
||||||
|
"size": f.stat().st_size,
|
||||||
|
}
|
||||||
|
for f in session_dir.iterdir()
|
||||||
|
if f.is_file()
|
||||||
|
],
|
||||||
|
key=lambda d: d["name"],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def get_artifact(session_id: str, name: str) -> str | None:
|
||||||
|
"""Return file content (utf-8) or None if the artifact does not exist."""
|
||||||
|
artifact_path = ARTIFACTS_DIR / session_id / name
|
||||||
|
if not artifact_path.exists():
|
||||||
|
return None
|
||||||
|
return artifact_path.read_text(encoding="utf-8")
|
||||||
|
|
||||||
|
|
||||||
|
def save_artifact(session_id: str, name: str, content: str) -> Path:
|
||||||
|
"""Write content to the artifact file, creating parent dirs as needed.
|
||||||
|
|
||||||
|
Returns the Path of the written file.
|
||||||
|
"""
|
||||||
|
session_dir = ARTIFACTS_DIR / session_id
|
||||||
|
session_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
artifact_path = session_dir / name
|
||||||
|
artifact_path.write_text(content, encoding="utf-8")
|
||||||
|
return artifact_path
|
||||||
@@ -0,0 +1,71 @@
|
|||||||
|
"""Tests for artifacts module — per-session CRUD storage."""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
import os
|
||||||
|
|
||||||
|
# Add backend to sys.path so we can import artifacts
|
||||||
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
import artifacts
|
||||||
|
from artifacts import list_artifacts, get_artifact, save_artifact
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def temp_artifacts(tmp_path, monkeypatch):
|
||||||
|
"""Redirect ARTIFACTS_DIR to a temp path for test isolation."""
|
||||||
|
monkeypatch.setattr(artifacts, "ARTIFACTS_DIR", tmp_path)
|
||||||
|
|
||||||
|
|
||||||
|
def test_list_empty_session():
|
||||||
|
"""Returns empty list for an unknown session (no directory)."""
|
||||||
|
result = list_artifacts("session-unknown")
|
||||||
|
assert result == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_and_list():
|
||||||
|
"""Save an artifact then list returns 1 artifact with correct fields."""
|
||||||
|
save_artifact("session-1", "guide.md", "# Guide content")
|
||||||
|
result = list_artifacts("session-1")
|
||||||
|
assert len(result) == 1
|
||||||
|
assert result[0]["name"] == "guide.md"
|
||||||
|
assert result[0]["session_id"] == "session-1"
|
||||||
|
|
||||||
|
|
||||||
|
def test_save_and_get():
|
||||||
|
"""Save an artifact then get returns the content."""
|
||||||
|
save_artifact("session-1", "notes.txt", "hello world")
|
||||||
|
content = get_artifact("session-1", "notes.txt")
|
||||||
|
assert content == "hello world"
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_nonexistent():
|
||||||
|
"""get_artifact returns None if file doesn't exist."""
|
||||||
|
result = get_artifact("session-1", "nonexistent.md")
|
||||||
|
assert result is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_multiple_artifacts():
|
||||||
|
"""2 artifacts for s1, 1 for s2 — list lengths match."""
|
||||||
|
save_artifact("s1", "a.md", "content a")
|
||||||
|
save_artifact("s1", "b.md", "content b")
|
||||||
|
save_artifact("s2", "c.md", "content c")
|
||||||
|
|
||||||
|
s1_list = list_artifacts("s1")
|
||||||
|
s2_list = list_artifacts("s2")
|
||||||
|
|
||||||
|
assert len(s1_list) == 2
|
||||||
|
assert len(s2_list) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_overwrite_artifact():
|
||||||
|
"""Save twice with same name → only 1 artifact, content is the second."""
|
||||||
|
save_artifact("session-1", "doc.md", "first content")
|
||||||
|
save_artifact("session-1", "doc.md", "second content")
|
||||||
|
|
||||||
|
result = list_artifacts("session-1")
|
||||||
|
assert len(result) == 1
|
||||||
|
|
||||||
|
content = get_artifact("session-1", "doc.md")
|
||||||
|
assert content == "second content"
|
||||||
Reference in New Issue
Block a user