xiaoxia
cdcc919853
Deploy / Deploy Staging (push) Waiting to run
Deploy / Build Production Runtime Images (push) Has been skipped
Deploy / Deploy Production (push) Has been skipped
Deploy / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Failing after 142h2m16s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 142h2m23s
- tests.yml: 添加集成测试步骤（单元测试后执行 pytest tests/integration） - ci-cd.yml: 添加集成测试步骤 - deploy.yml: 添加部署后冒烟测试（健康检查 + 登录API验证 + /docs端点检查） - requirements-dev.txt: 添加 pytest-timeout 依赖 - test_auth.py: 修复预期状态码（201→200），需要PG的测试添加skipif标记 - test_api.py: 修复预期状态码和错误消息格式，添加skipif标记 - test_projects.py: 修复 ListAssetLibrariesUseCase.execute() 调用签名 - InMemoryAssetRepository: 添加 find_by_library 别名匹配端口接口 集成测试结果: 101 passed, 19 skipped (需PG的认证测试) 无外部依赖的测试全部通过
124 lines
3.4 KiB
Python
Executable File
124 lines
3.4 KiB
Python
Executable File
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)
|
|
|
|
project = create_use_case.execute(
|
|
CreateProjectCommand(
|
|
name="Demo Project",
|
|
description="Demo description",
|
|
),
|
|
owner_user_id="user-1",
|
|
)
|
|
|
|
assert project.name == "Demo Project"
|
|
|
|
items = list_use_case.execute("user-1")
|
|
assert len(items) == 1
|
|
assert items[0].id == project.id
|
|
|
|
|
|
def test_get_project_by_id():
|
|
repository = InMemoryProjectRepository()
|
|
create_use_case = CreateProjectUseCase(repository)
|
|
get_use_case = GetProjectUseCase(repository)
|
|
|
|
project = create_use_case.execute(
|
|
CreateProjectCommand(
|
|
name="Demo Project",
|
|
description="Demo description",
|
|
),
|
|
owner_user_id="user-1",
|
|
)
|
|
|
|
retrieved = get_use_case.execute(project.id)
|
|
assert retrieved is not None
|
|
assert retrieved.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(
|
|
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")
|
|
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
|