Xiaoxia AI
b5a62ee9a3
- domain: User, Workspace, Project, AssetLibrary, Asset, IngestJob - ports: repository interfaces - application: use cases - adapters: in-memory - API: FastAPI 5 routes - worker: Celery ingest_asset - tests: 4 passing
103 lines
3.0 KiB
Python
103 lines
3.0 KiB
Python
from packages.application import (
|
|
CreateAssetCommand,
|
|
CreateAssetLibraryCommand,
|
|
CreateAssetLibraryUseCase,
|
|
CreateAssetUseCase,
|
|
CreateProjectCommand,
|
|
CreateProjectUseCase,
|
|
ListAssetLibrariesUseCase,
|
|
ListAssetsUseCase,
|
|
ListProjectsUseCase,
|
|
SubmitIngestJobCommand,
|
|
SubmitIngestJobUseCase,
|
|
)
|
|
from packages.adapters.in_memory import (
|
|
InMemoryAssetLibraryRepository,
|
|
InMemoryAssetRepository,
|
|
InMemoryIngestJobRepository,
|
|
InMemoryProjectRepository,
|
|
)
|
|
from packages.domain import AssetLibraryKind, IngestJobStatus
|
|
|
|
|
|
def test_create_and_list_projects():
|
|
repository = InMemoryProjectRepository()
|
|
create_use_case = CreateProjectUseCase(repository)
|
|
list_use_case = ListProjectsUseCase(repository)
|
|
|
|
project = create_use_case.execute(CreateProjectCommand(workspace_id="ws-1", name=" Demo Project "))
|
|
assert project.name == "Demo Project"
|
|
assert project.workspace_id == "ws-1"
|
|
|
|
items = list_use_case.execute("ws-1")
|
|
assert len(items) == 1
|
|
assert items[0].id == project.id
|
|
|
|
|
|
def test_create_and_list_asset_libraries():
|
|
repository = InMemoryAssetLibraryRepository()
|
|
create_use_case = CreateAssetLibraryUseCase(repository)
|
|
list_use_case = ListAssetLibrariesUseCase(repository)
|
|
|
|
library = create_use_case.execute(
|
|
CreateAssetLibraryCommand(
|
|
workspace_id="ws-1",
|
|
project_id="proj-1",
|
|
name=" 素材库 A ",
|
|
kind=AssetLibraryKind.VIDEO,
|
|
)
|
|
)
|
|
assert library.name == "素材库 A"
|
|
assert library.kind == AssetLibraryKind.VIDEO
|
|
|
|
items = list_use_case.execute("proj-1", kind=AssetLibraryKind.VIDEO)
|
|
assert len(items) == 1
|
|
assert items[0].id == library.id
|
|
|
|
|
|
def test_create_and_list_assets():
|
|
repository = InMemoryAssetRepository()
|
|
create_use_case = CreateAssetUseCase(repository)
|
|
list_use_case = ListAssetsUseCase(repository)
|
|
|
|
asset = create_use_case.execute(
|
|
CreateAssetCommand(
|
|
workspace_id="ws-1",
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
name=" demo.mp4 ",
|
|
storage_key="uploads/demo.mp4",
|
|
mime_type="video/mp4",
|
|
metadata={"duration": 12.5},
|
|
)
|
|
)
|
|
assert asset.name == "demo.mp4"
|
|
assert asset.storage_key == "uploads/demo.mp4"
|
|
assert asset.metadata["duration"] == 12.5
|
|
|
|
items = list_use_case.execute("lib-1")
|
|
assert len(items) == 1
|
|
assert items[0].id == asset.id
|
|
|
|
|
|
def test_submit_ingest_job():
|
|
repository = InMemoryIngestJobRepository()
|
|
use_case = SubmitIngestJobUseCase(repository)
|
|
|
|
job = use_case.execute(
|
|
SubmitIngestJobCommand(
|
|
workspace_id="ws-1",
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
storage_key="uploads/raw-video.mp4",
|
|
)
|
|
)
|
|
assert job.workspace_id == "ws-1"
|
|
assert job.storage_key == "uploads/raw-video.mp4"
|
|
assert job.status == IngestJobStatus.PENDING
|
|
assert job.result_asset_id == ""
|
|
|
|
retrieved = repository.get(job.id)
|
|
assert retrieved is not None
|
|
assert retrieved.id == job.id
|