"""Tests for ArtifactStorageManager."""
import pytest
from chuk_motion.models import (
NamespaceType,
ProjectMetadata,
ProviderType,
StorageScope,
)
from chuk_motion.storage import ArtifactStorageManager
@pytest.mark.asyncio
async def test_artifact_storage_initialization():
"""Test ArtifactStorageManager initialization."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
assert manager.store is not None
await manager.cleanup()
@pytest.mark.asyncio
async def test_create_project_user_scope():
"""Test creating a project with USER scope."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create project
project_info = await manager.create_project(
project_name="test_project",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
# Verify project info
assert project_info.namespace_info.namespace_type == NamespaceType.WORKSPACE
assert project_info.namespace_info.scope == StorageScope.USER
assert project_info.namespace_info.user_id == "alice"
assert project_info.metadata.project_name == "test_project"
assert project_info.metadata.theme == "tech"
assert project_info.metadata.fps == 30
assert project_info.metadata.width == 1920
assert project_info.metadata.height == 1080
assert project_info.checkpoints == []
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_create_project_session_scope():
"""Test creating a project with SESSION scope."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create project with SESSION scope and TTL
project_info = await manager.create_project(
project_name="temp_project",
theme="minimal",
fps=24,
width=1280,
height=720,
scope=StorageScope.SESSION,
ttl_hours=24,
)
# Verify project info
assert project_info.namespace_info.namespace_type == NamespaceType.WORKSPACE
assert project_info.namespace_info.scope == StorageScope.SESSION
# Note: ttl_hours might not be preserved by chuk-artifacts
assert project_info.metadata.project_name == "temp_project"
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_create_project_requires_user_id_for_user_scope():
"""Test that USER scope requires user_id."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
with pytest.raises(ValueError, match="user_id is required for USER scope"):
await manager.create_project(
project_name="test_project",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id=None, # Missing user_id
)
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_get_project():
"""Test getting project information."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create project
created_project = await manager.create_project(
project_name="get_test_project",
theme="finance",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="bob",
)
# Get project
retrieved_project = await manager.get_project(created_project.namespace_info.namespace_id)
# Verify
assert (
retrieved_project.namespace_info.namespace_id
== created_project.namespace_info.namespace_id
)
assert retrieved_project.metadata.project_name == "get_test_project"
assert retrieved_project.metadata.theme == "finance"
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_update_project_metadata():
"""Test updating project metadata."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create project
project = await manager.create_project(
project_name="update_test",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
# Update metadata
updated_metadata = ProjectMetadata(
project_name="update_test",
theme="tech",
fps=30,
width=1920,
height=1080,
total_duration_seconds=45.5,
component_count=5,
)
updated_project = await manager.update_project_metadata(
project.namespace_info.namespace_id, updated_metadata
)
# Verify updates
assert updated_project.metadata.total_duration_seconds == 45.5
assert updated_project.metadata.component_count == 5
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_list_projects():
"""Test listing projects."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create multiple projects
p1 = await manager.create_project(
project_name="project1",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
p2 = await manager.create_project(
project_name="project2",
theme="finance",
fps=24,
width=1280,
height=720,
scope=StorageScope.USER,
user_id="bob",
)
# Note: list_namespaces in chuk-artifacts has a caching issue
# So we verify we can retrieve individual projects instead
retrieved_p1 = await manager.get_project(p1.namespace_info.namespace_id)
assert retrieved_p1.metadata.project_name == "project1"
retrieved_p2 = await manager.get_project(p2.namespace_info.namespace_id)
assert retrieved_p2.metadata.project_name == "project2"
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_delete_project():
"""Test deleting a project."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create project
project = await manager.create_project(
project_name="delete_test",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
# Delete project
await manager.delete_project(project.namespace_info.namespace_id)
# Verify deletion - should raise error when trying to get
with pytest.raises(ValueError, match="Namespace not found"):
await manager.get_project(project.namespace_info.namespace_id)
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_store_and_retrieve_render():
"""Test storing and retrieving rendered video."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create project first
project = await manager.create_project(
project_name="render_test",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
# Store render
video_data = b"fake video data"
render_info = await manager.store_render(
project_namespace_id=project.namespace_info.namespace_id,
video_data=video_data,
format="mp4",
resolution="1920x1080",
fps=30,
duration_seconds=45.5,
scope=StorageScope.USER,
user_id="alice",
codec="h264",
bitrate_kbps=5000,
)
# Verify render info
assert render_info.namespace_info.namespace_type == NamespaceType.BLOB
assert render_info.metadata.format == "mp4"
assert render_info.metadata.duration_seconds == 45.5
assert render_info.metadata.file_size_bytes == len(video_data)
assert render_info.metadata.checksum is not None
# Retrieve render data
retrieved_data = await manager.read_render_data(render_info.namespace_info.namespace_id)
assert retrieved_data == video_data
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_store_and_retrieve_asset():
"""Test storing and retrieving media asset."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Store asset
asset_data = b"fake image data"
asset_info = await manager.store_asset(
asset_data=asset_data,
asset_type="image",
mime_type="image/png",
tags=["background", "tech"],
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
# Verify asset info
assert asset_info.namespace_info.namespace_type == NamespaceType.BLOB
assert asset_info.metadata.asset_type == "image"
assert asset_info.metadata.mime_type == "image/png"
assert asset_info.metadata.tags == ["background", "tech"]
assert asset_info.metadata.width == 1920
assert asset_info.metadata.height == 1080
# Retrieve asset data
retrieved_data = await manager.read_asset_data(asset_info.namespace_info.namespace_id)
assert retrieved_data == asset_data
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_create_and_list_checkpoints():
"""Test creating and listing checkpoints."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create project
project = await manager.create_project(
project_name="checkpoint_test",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
# Create checkpoints
await manager.create_checkpoint(
namespace_id=project.namespace_info.namespace_id,
name="v1.0",
description="First version",
)
await manager.create_checkpoint(
namespace_id=project.namespace_info.namespace_id,
name="v1.1",
description="Updated version",
)
# List checkpoints
checkpoints = await manager.list_checkpoints(project.namespace_info.namespace_id)
assert len(checkpoints) == 2
assert checkpoints[0].name == "v1.0"
assert checkpoints[1].name == "v1.1"
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_restore_checkpoint():
"""Test restoring from a checkpoint."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create project
project = await manager.create_project(
project_name="restore_test",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
# Create a checkpoint
cp = await manager.create_checkpoint(
namespace_id=project.namespace_info.namespace_id,
name="v1.0",
description="Before changes",
)
# Restore checkpoint - this should work without error
await manager.restore_checkpoint(
namespace_id=project.namespace_info.namespace_id,
checkpoint_id=cp.checkpoint_id,
)
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_store_render_requires_user_id_for_user_scope():
"""Test that store_render requires user_id for USER scope."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
with pytest.raises(ValueError, match="user_id is required for USER scope"):
await manager.store_render(
project_namespace_id="test_ns",
video_data=b"video",
format="mp4",
resolution="1920x1080",
fps=30,
duration_seconds=30.0,
scope=StorageScope.USER,
user_id=None,
)
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_store_asset_requires_user_id_for_user_scope():
"""Test that store_asset requires user_id for USER scope."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
with pytest.raises(ValueError, match="user_id is required for USER scope"):
await manager.store_asset(
asset_data=b"image",
asset_type="image",
mime_type="image/png",
scope=StorageScope.USER,
user_id=None,
)
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_store_not_initialized_raises_error():
"""Test that accessing store before initialization raises error."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
with pytest.raises(RuntimeError, match="not initialized"):
_ = manager.store
@pytest.mark.asyncio
async def test_double_initialize():
"""Test that double initialization is safe."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
# Second init should be a no-op
await manager.initialize()
assert manager.store is not None
await manager.cleanup()
@pytest.mark.asyncio
async def test_double_cleanup():
"""Test that double cleanup is safe."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
await manager.cleanup()
# Second cleanup should be a no-op
await manager.cleanup()
@pytest.mark.asyncio
@pytest.mark.skip(reason="BLOB namespace path-based read not supported in memory provider")
async def test_get_render():
"""Test getting render information by namespace ID."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create project first
project = await manager.create_project(
project_name="render_get_test",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
# Store render
video_data = b"test video data"
render_info = await manager.store_render(
project_namespace_id=project.namespace_info.namespace_id,
video_data=video_data,
format="mp4",
resolution="1920x1080",
fps=30,
duration_seconds=30.0,
scope=StorageScope.USER,
user_id="alice",
)
# Get render info
retrieved_render = await manager.get_render(render_info.namespace_info.namespace_id)
assert retrieved_render.metadata.format == "mp4"
assert retrieved_render.metadata.duration_seconds == 30.0
assert retrieved_render.namespace_info.namespace_type == NamespaceType.BLOB
finally:
await manager.cleanup()
@pytest.mark.asyncio
@pytest.mark.skip(reason="BLOB namespace path-based read not supported in memory provider")
async def test_get_asset():
"""Test getting asset information by namespace ID."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Store asset
asset_data = b"test image data"
asset_info = await manager.store_asset(
asset_data=asset_data,
asset_type="image",
mime_type="image/png",
tags=["test", "unit"],
scope=StorageScope.USER,
user_id="alice",
)
# Get asset info
retrieved_asset = await manager.get_asset(asset_info.namespace_info.namespace_id)
assert retrieved_asset.metadata.asset_type == "image"
assert retrieved_asset.metadata.mime_type == "image/png"
assert retrieved_asset.metadata.tags == ["test", "unit"]
assert retrieved_asset.namespace_info.namespace_type == NamespaceType.BLOB
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_list_projects_with_filtering():
"""Test listing projects with scope filtering."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create projects with different scopes
await manager.create_project(
project_name="user_project",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
await manager.create_project(
project_name="session_project",
theme="minimal",
fps=30,
width=1920,
height=1080,
scope=StorageScope.SESSION,
)
# List all projects - just verify it doesn't error
await manager.list_projects()
# List with scope filter - just verify it doesn't error
await manager.list_projects(scope=StorageScope.USER)
# List with user filter - just verify it doesn't error
await manager.list_projects(user_id="alice")
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_get_project_vfs():
"""Test getting VFS for a project."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
project = await manager.create_project(
project_name="vfs_test",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
# Get VFS
vfs = await manager.get_project_vfs(project.namespace_info.namespace_id)
assert vfs is not None
# Write a file using VFS
await vfs.write_file("/test.txt", b"hello world")
# Read it back
content = await vfs.read_file("/test.txt")
assert content == b"hello world"
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_create_project_session_scope_default_ttl():
"""Test creating session project with default TTL."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create project with SESSION scope without specifying TTL
project_info = await manager.create_project(
project_name="session_default_ttl",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.SESSION,
# ttl_hours not specified, should default to 24
)
assert project_info.namespace_info.scope == StorageScope.SESSION
assert project_info.metadata.project_name == "session_default_ttl"
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_store_asset_with_duration():
"""Test storing asset with optional duration parameter."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Store an audio asset with duration
asset_data = b"audio data"
asset_info = await manager.store_asset(
asset_data=asset_data,
asset_type="audio",
mime_type="audio/mp3",
duration_seconds=120.5,
scope=StorageScope.USER,
user_id="alice",
)
assert asset_info.metadata.asset_type == "audio"
assert asset_info.metadata.duration_seconds == 120.5
finally:
await manager.cleanup()
@pytest.mark.asyncio
async def test_store_asset_with_project_namespace_ids():
"""Test storing asset linked to projects."""
manager = ArtifactStorageManager(provider_type=ProviderType.MEMORY)
await manager.initialize()
try:
# Create a project
project = await manager.create_project(
project_name="asset_link_test",
theme="tech",
fps=30,
width=1920,
height=1080,
scope=StorageScope.USER,
user_id="alice",
)
# Store asset linked to project
asset_data = b"linked asset"
asset_info = await manager.store_asset(
asset_data=asset_data,
asset_type="image",
mime_type="image/png",
project_namespace_ids=[project.namespace_info.namespace_id],
scope=StorageScope.USER,
user_id="alice",
)
assert asset_info.metadata.project_namespace_ids == [project.namespace_info.namespace_id]
finally:
await manager.cleanup()
