content-find-agent-new/tests/test_run_service_post_archive.py

from __future__ import annotations

from content_agent import run_service
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from content_agent.run_service import RunService
from content_agent.schemas import RunStartRequest


class EventRecordingRuntime(LocalRuntimeFileStore):
    def append_run_event_records(self, run_id, policy_run_id, rows):
        prepared = [
            {**row, "run_id": run_id, "policy_run_id": row.get("policy_run_id", policy_run_id)}
            for row in rows
        ]
        self.append_jsonl(run_id, "run_events.jsonl", prepared)


def test_post_run_oss_archive_triggers_for_real_pending_records(monkeypatch, tmp_path):
    runtime = EventRecordingRuntime(tmp_path)
    runtime.prepare_run("run_001")
    runtime.append_jsonl(
        "run_001",
        "content_media_records.jsonl",
        [
            {
                "run_id": "run_001",
                "policy_run_id": "policy_001",
                "platform": "kuaishou",
                "platform_content_id": "content_001",
                "content_media_status": "oss_upload_pending",
                "play_url": "https://source.example/video.mp4",
                "raw_payload": {},
            }
        ],
    )
    service = object.__new__(RunService)
    service.runtime = runtime
    service._start_background_thread = lambda target, name: target()
    calls: list[str] = []

    def fake_archive(runtime_arg, run_id):
        calls.append(run_id)
        return [
            {
                "platform_content_id": "content_001",
                "content_media_status": "oss_uploaded",
            }
        ]

    monkeypatch.setattr(run_service.oss_archive, "archive_pending_for_run", fake_archive)

    service._trigger_post_run_oss_archive(
        {"run_id": "run_001", "policy_run_id": "policy_001"},
        RunStartRequest(platform="kuaishou", platform_mode="real"),
    )

    events = runtime.read_jsonl("run_001", "run_events.jsonl")
    assert calls == ["run_001"]
    assert [event["event_id"] for event in events] == [
        "oss_archive_post_run_started",
        "oss_archive_post_run_completed",
    ]
    assert events[0]["raw_payload"]["pending_due_count"] == 1
    assert events[1]["raw_payload"]["content_media_status_counts"] == {"oss_uploaded": 1}


def test_post_run_oss_archive_failure_is_recorded_without_raising(monkeypatch, tmp_path):
    runtime = EventRecordingRuntime(tmp_path)
    runtime.prepare_run("run_001")
    runtime.append_jsonl(
        "run_001",
        "content_media_records.jsonl",
        [
            {
                "run_id": "run_001",
                "policy_run_id": "policy_001",
                "platform": "shipinhao",
                "platform_content_id": "content_001",
                "content_media_status": "oss_upload_pending",
                "play_url": "https://source.example/video.mp4",
                "raw_payload": {},
            }
        ],
    )
    service = object.__new__(RunService)
    service.runtime = runtime
    service._start_background_thread = lambda target, name: target()

    def fail_archive(runtime_arg, run_id):
        raise TimeoutError("archive stuck")

    monkeypatch.setattr(run_service.oss_archive, "archive_pending_for_run", fail_archive)

    service._trigger_post_run_oss_archive(
        {"run_id": "run_001", "policy_run_id": "policy_001"},
        RunStartRequest(platform="shipinhao", platform_mode="real"),
    )

    events = runtime.read_jsonl("run_001", "run_events.jsonl")
    assert [event["event_id"] for event in events] == [
        "oss_archive_post_run_started",
        "oss_archive_post_run_failed",
    ]
    assert events[1]["error_code"] == "OSS_ARCHIVE_POST_RUN_FAILED"
    assert events[1]["raw_payload"]["exception_type"] == "TimeoutError"


def test_post_run_oss_archive_skips_mock_and_non_pending_records(monkeypatch, tmp_path):
    runtime = EventRecordingRuntime(tmp_path)
    runtime.prepare_run("run_001")
    runtime.append_jsonl(
        "run_001",
        "content_media_records.jsonl",
        [
            {
                "run_id": "run_001",
                "policy_run_id": "policy_001",
                "platform_content_id": "content_uploaded",
                "content_media_status": "oss_uploaded",
                "play_url": "https://source.example/video.mp4",
            }
        ],
    )
    service = object.__new__(RunService)
    service.runtime = runtime
    service._start_background_thread = lambda target, name: target()
    calls: list[str] = []
    monkeypatch.setattr(
        run_service.oss_archive,
        "archive_pending_for_run",
        lambda runtime_arg, run_id: calls.append(run_id),
    )

    service._trigger_post_run_oss_archive(
        {"run_id": "run_001", "policy_run_id": "policy_001"},
        RunStartRequest(platform="douyin", platform_mode="real"),
    )
    service._trigger_post_run_oss_archive(
        {"run_id": "run_001", "policy_run_id": "policy_001"},
        RunStartRequest(platform="douyin", platform_mode="mock"),
    )

    assert calls == []
    assert runtime.read_jsonl("run_001", "run_events.jsonl") == []