Xiaoxia AI
124 lines
3.8 KiB
Python
124 lines
3.8 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from packages.application import CreateGenerationTaskCommand, CreateGenerationTaskUseCase
|
|
from packages.domain import GeneratedVideo, GenerationTaskStatus
|
|
|
|
|
|
class DummyGenerationTaskRepository:
|
|
def __init__(self):
|
|
self.items = {}
|
|
|
|
def create(self, task):
|
|
self.items[task.id] = task
|
|
return task
|
|
|
|
def get(self, task_id):
|
|
return self.items.get(task_id)
|
|
|
|
def list_by_project(self, project_id):
|
|
return [task for task in self.items.values() if task.project_id == project_id]
|
|
|
|
def update(self, task):
|
|
self.items[task.id] = task
|
|
return task
|
|
|
|
|
|
class DummyGeneratedVideoRepository:
|
|
def __init__(self):
|
|
self.items = {}
|
|
|
|
def create(self, video):
|
|
self.items[video.id] = video
|
|
return video
|
|
|
|
def get(self, video_id):
|
|
return self.items.get(video_id)
|
|
|
|
def list_by_project(self, project_id):
|
|
return [video for video in self.items.values() if video.project_id == project_id]
|
|
|
|
def list_by_generation_task(self, generation_task_id):
|
|
return [video for video in self.items.values() if video.generation_task_id == generation_task_id]
|
|
|
|
|
|
def simulate_generate_video(task_id: str, task_repo: DummyGenerationTaskRepository, video_repo: DummyGeneratedVideoRepository) -> dict:
|
|
task = task_repo.get(task_id)
|
|
if task is None:
|
|
return {"status": "failed", "error": "task not found"}
|
|
|
|
task.status = GenerationTaskStatus.RUNNING
|
|
task.progress = 20.0
|
|
task.started_at = task.started_at or datetime.now(timezone.utc)
|
|
task_repo.update(task)
|
|
|
|
video = GeneratedVideo.create(
|
|
workspace_id=task.workspace_id,
|
|
project_id=task.project_id,
|
|
generation_task_id=task.id,
|
|
name=f"{task.id}.mp4",
|
|
file_url=f"https://example.invalid/generated/{task.id}.mp4",
|
|
file_size=2048,
|
|
duration=12.5,
|
|
width=1920,
|
|
height=1080,
|
|
fps=25.0,
|
|
)
|
|
video_repo.create(video)
|
|
|
|
task.status = GenerationTaskStatus.COMPLETED
|
|
task.progress = 100.0
|
|
task.result_count = 1
|
|
task.completed_at = datetime.now(timezone.utc)
|
|
task_repo.update(task)
|
|
|
|
return {"status": "completed", "task_id": task.id, "video_id": video.id}
|
|
|
|
|
|
def test_create_generation_task_smoke():
|
|
repo = DummyGenerationTaskRepository()
|
|
use_case = CreateGenerationTaskUseCase(repo)
|
|
task = use_case.execute(
|
|
CreateGenerationTaskCommand(
|
|
workspace_id="ws-1",
|
|
project_id="proj-1",
|
|
asset_library_id="lib-1",
|
|
strategy_id="str-1",
|
|
voice_library_id="voice-1",
|
|
created_by_user_id="user-1",
|
|
)
|
|
)
|
|
assert task.workspace_id == "ws-1"
|
|
assert task.project_id == "proj-1"
|
|
assert task.asset_library_id == "lib-1"
|
|
assert task.status == GenerationTaskStatus.PENDING
|
|
assert repo.get(task.id) is not None
|
|
|
|
|
|
def test_generation_pipeline_smoke():
|
|
task_repo = DummyGenerationTaskRepository()
|
|
video_repo = DummyGeneratedVideoRepository()
|
|
use_case = CreateGenerationTaskUseCase(task_repo)
|
|
task = use_case.execute(
|
|
CreateGenerationTaskCommand(
|
|
workspace_id="ws-1",
|
|
project_id="proj-1",
|
|
asset_library_id="lib-1",
|
|
strategy_id="str-1",
|
|
voice_library_id="voice-1",
|
|
created_by_user_id="user-1",
|
|
)
|
|
)
|
|
|
|
result = simulate_generate_video(task.id, task_repo, video_repo)
|
|
|
|
assert result["status"] == "completed"
|
|
updated_task = task_repo.get(task.id)
|
|
assert updated_task is not None
|
|
assert updated_task.status == GenerationTaskStatus.COMPLETED
|
|
assert updated_task.result_count == 1
|
|
|
|
videos = video_repo.list_by_generation_task(task.id)
|
|
assert len(videos) == 1
|
|
assert videos[0].project_id == "proj-1"
|
|
assert videos[0].generation_task_id == task.id
|