SupplyAgent/tests/supply_infra/scheduler/test_aigc_publish_scheduler.py

from __future__ import annotations

from unittest.mock import patch

from supply_infra.scheduler.app import create_scheduler, submit_aigc_publish
from supply_infra.scheduler.constants import (
    AIGC_PUBLISH_JOB_ID,
    SUPPLY_PIPELINE_JOB_ID,
)


def test_create_scheduler_registers_pipeline_and_aigc_jobs() -> None:
    scheduler = create_scheduler()
    job_ids = {job.id for job in scheduler.get_jobs()}
    assert SUPPLY_PIPELINE_JOB_ID in job_ids
    assert AIGC_PUBLISH_JOB_ID in job_ids
    aigc_job = scheduler.get_job(AIGC_PUBLISH_JOB_ID)
    assert aigc_job is not None
    assert aigc_job.trigger.interval.total_seconds() == 600


@patch(
    "supply_infra.scheduler.jobs.publish_videos_from_discovery.publish_videos_from_discovery"
)
def test_submit_aigc_publish_scans_all_unpublished(mock_publish) -> None:
    mock_publish.return_value = {
        "success": True,
        "candidate_count": 0,
        "batch_count": 0,
        "failed_batch_count": 0,
    }

    result = submit_aigc_publish()

    assert result["success"] is True
    mock_publish.assert_called_once_with(skip_published=True, any_biz_dt=True)