xiaoxia-saas/tests/integration/test_generation_pipeline.py

from datetime import datetime, timezone

from packages.application import (
    CreateGenerationTaskCommand,
    CreateGenerationTaskUseCase,
    GetGeneratedVideoDownloadUrlUseCase,
)
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)

    file_url = f"/projects/{task.project_id}/generated/{task.id}/{task.id}.mp4"
    video = GeneratedVideo.create(
        project_id=task.project_id,
        generation_task_id=task.id,
        name=f"{task.id}.mp4",
        file_url=file_url,
        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,
        "file_url": file_url,
    }


def test_create_generation_task_smoke():
    repo = DummyGenerationTaskRepository()
    use_case = CreateGenerationTaskUseCase(repo)
    task = use_case.execute(
        CreateGenerationTaskCommand(
            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.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(
            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"
    assert "/projects/proj-1/generated/" in result["file_url"]
    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


def test_get_generated_video_download_url():
    video_repo = DummyGeneratedVideoRepository()
    video = GeneratedVideo.create(
        project_id="proj-1",
        generation_task_id="task-1",
        name="task-1.mp4",
        file_url="https://example.invalid/generated/task-1.mp4",
        file_size=2048,
        duration=12.5,
        width=1920,
        height=1080,
        fps=25.0,
    )
    video_repo.create(video)

    use_case = GetGeneratedVideoDownloadUrlUseCase(video_repo)
    assert use_case.execute(video.id) == "https://example.invalid/generated/task-1.mp4"
    assert use_case.execute("missing") is None


def test_generated_video_download_source_url_is_stable():
    video_repo = DummyGeneratedVideoRepository()
    video = GeneratedVideo.create(
        project_id="proj-1",
        generation_task_id="task-2",
        name="task-2.mp4",
        file_url="http://localhost:9000/xiaoxia-assets/generated/task-2.mp4",
        file_size=1024,
        duration=8.0,
        width=1280,
        height=720,
        fps=25.0,
    )
    video_repo.create(video)

    use_case = GetGeneratedVideoDownloadUrlUseCase(video_repo)
    assert use_case.execute(video.id).endswith("/generated/task-2.mp4")