xiaoxia
560856cf22
Deploy / Build Production Runtime Images (push) Has been skipped
Deploy / Deploy Production (push) Has been skipped
Deploy / Production Browser E2E (push) Has been skipped
Deploy / Deploy Staging (push) Failing after 210h26m26s
CI/CD Pipeline / Frontend Lint (push) Failing after 210h27m0s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 210h27m7s
108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
from packages.adapters.in_memory import (
|
|
InMemoryAssetLibraryRepository,
|
|
InMemoryAssetRepository,
|
|
InMemoryIngestJobRepository,
|
|
InMemoryProjectRepository,
|
|
)
|
|
from packages.application import (
|
|
CreateAssetCommand,
|
|
CreateAssetLibraryCommand,
|
|
CreateAssetLibraryUseCase,
|
|
CreateAssetUseCase,
|
|
CreateProjectCommand,
|
|
CreateProjectUseCase,
|
|
GetProjectUseCase,
|
|
ListAssetLibrariesUseCase,
|
|
ListAssetsUseCase,
|
|
ListProjectsUseCase,
|
|
SubmitIngestJobCommand,
|
|
SubmitIngestJobUseCase,
|
|
)
|
|
from packages.domain import AssetLibraryKind, IngestJobStatus
|
|
|
|
|
|
def test_create_and_list_projects():
|
|
repository = InMemoryProjectRepository()
|
|
create_use_case = CreateProjectUseCase(repository)
|
|
list_use_case = ListProjectsUseCase(repository)
|
|
|
|
assert project.name == "Demo Project" # noqa: F821
|
|
|
|
items = list_use_case.execute("ws-1")
|
|
assert len(items) == 1
|
|
assert items[0].id == project.id # noqa: F821
|
|
|
|
|
|
def test_get_project_by_id_restores_workspace_context():
|
|
repository = InMemoryProjectRepository()
|
|
create_use_case = CreateProjectUseCase(repository)
|
|
get_use_case = GetProjectUseCase(repository)
|
|
|
|
retrieved = get_use_case.execute(project.id) # noqa: F821
|
|
assert retrieved is not None
|
|
assert retrieved.id == project.id # noqa: F821
|
|
|
|
|
|
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(
|
|
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(
|
|
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(
|
|
project_id="proj-1",
|
|
library_id="lib-1",
|
|
storage_key="uploads/raw-video.mp4",
|
|
)
|
|
)
|
|
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
|