video-image-production-build/tests/run/test_segment_entry.py

from __future__ import annotations

import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from production_build_agents.agents.segment.run import (
    run_segment_production,
)
from production_build_agents.run.records import VersionConflictError
from production_build_agents.run.registry import register_thread
from production_build_agents.production.runtime import (
    PRODUCTION_PROTOCOL_VERSION,
    validate_production_run_directory_boundary as validate_boundary,
)
from production_build_agents.run.layout import (
    run_directory,
    segment_package_path,
)
from production_build_agents.run.lock import (
    RunLockError,
    acquire_run_lock,
)
from run_segment import run_production_segment, segment_request_sha256
from tests.support.production_fixtures import segment_bundle


class SegmentEntryTest(unittest.TestCase):
    def test_thread_binds_exact_segment_package_bytes(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package, *_ = segment_bundle(1, run_id="segment-thread")
            package_path = root / "Segment1.v1.json"
            package_path.write_text(
                package.model_dump_json(),
                encoding="utf-8",
            )
            registry = root / "registry"
            run_dir = root / "run"
            first_hash = segment_request_sha256(package_path.read_bytes())
            register_thread(
                registry_dir=registry,
                thread_id="segment-thread",
                protocol_version=PRODUCTION_PROTOCOL_VERSION,
                input_path=package_path,
                input_sha256=first_hash,
                run_dir=run_dir,
            )

            package_path.write_bytes(package_path.read_bytes() + b"\n")
            with self.assertRaisesRegex(
                VersionConflictError,
                "input_sha256",
            ):
                register_thread(
                    registry_dir=registry,
                    thread_id="segment-thread",
                    protocol_version=PRODUCTION_PROTOCOL_VERSION,
                    input_path=package_path,
                    input_sha256=segment_request_sha256(
                        package_path.read_bytes()
                    ),
                    run_dir=run_dir,
                )

    def test_old_run_is_rejected_before_registry_or_lock_write(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package, *_ = segment_bundle(1, run_id="segment-thread")
            package_path = root / "Segment1.v1.json"
            package_path.write_text(
                package.model_dump_json(),
                encoding="utf-8",
            )
            output_root = root / "output"
            run_dir = run_directory(output_root, package.run_id)
            run_dir.mkdir(parents=True)
            old_record = run_dir / "production_package.json"
            old_record.write_text(
                '{"schema_version":"0.5"}',
                encoding="utf-8",
            )
            registry = root / "registry"

            with self.assertRaisesRegex(ValueError, "旧版正式路径"):
                run_production_segment(
                    package_path,
                    output_root=output_root,
                    registry_dir=registry,
                )

            self.assertFalse(registry.exists())
            self.assertEqual(
                sorted(
                    path.relative_to(run_dir)
                    for path in run_dir.rglob("*")
                ),
                [Path("production_package.json")],
            )

    def test_locked_recheck_rejects_concurrent_mixed_record(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package, *_ = segment_bundle(1, run_id="segment-thread")
            package_path = root / "Segment1.v1.json"
            package_path.write_text(
                package.model_dump_json(),
                encoding="utf-8",
            )
            output_root = root / "output"
            run_dir = run_directory(output_root, package.run_id)
            registry = root / "registry"
            injected = run_dir / "segment_packages" / "Injected.v1.json"
            call_count = 0

            def validate_then_inject(path: Path) -> None:
                nonlocal call_count
                validate_boundary(path)
                call_count += 1
                if call_count == 1:
                    injected.parent.mkdir(parents=True)
                    injected.write_text(
                        '{"schema_version":"0.4"}',
                        encoding="utf-8",
                    )

            with (
                patch(
                    "run_segment.validate_production_run_directory_boundary",
                    side_effect=validate_then_inject,
                ),
                self.assertRaisesRegex(ValueError, "非当前协议"),
            ):
                run_production_segment(
                    package_path,
                    output_root=output_root,
                    registry_dir=registry,
                )

            self.assertEqual(call_count, 1)
            self.assertFalse(registry.exists())
            self.assertEqual(
                injected.read_text(encoding="utf-8"),
                '{"schema_version":"0.4"}',
            )
            self.assertFalse(
                segment_package_path(
                    run_dir,
                    package.segment_id,
                    package.plan_version,
                ).exists()
            )

    def test_source_mutation_after_parse_is_rejected_before_registration(
        self,
    ) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package, *_ = segment_bundle(1, run_id="segment-thread")
            package_path = root / "Segment1.v1.json"
            package_path.write_text(
                package.model_dump_json(),
                encoding="utf-8",
            )
            original_bytes = package_path.read_bytes()
            output_root = root / "output"
            run_dir = run_directory(output_root, package.run_id)
            registry = root / "registry"
            call_count = 0

            def validate_then_mutate(path: Path) -> None:
                nonlocal call_count
                validate_boundary(path)
                call_count += 1
                if call_count == 1:
                    package_path.write_bytes(original_bytes + b"\n")

            with (
                patch(
                    "run_segment.validate_production_run_directory_boundary",
                    side_effect=validate_then_mutate,
                ),
                self.assertRaisesRegex(
                    VersionConflictError,
                    "输入文件在封印前已变化",
                ),
            ):
                run_production_segment(
                    package_path,
                    output_root=output_root,
                    registry_dir=registry,
                )

            self.assertEqual(call_count, 2)
            self.assertFalse(registry.exists())
            self.assertFalse(
                segment_package_path(
                    run_dir,
                    package.segment_id,
                    package.plan_version,
                ).exists()
            )

    def test_locked_preflight_runs_while_run_lock_is_held(self) -> None:
        class PreflightObserved(RuntimeError):
            pass

        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            package, *_ = segment_bundle(1, run_id="segment-thread")

            def observe_lock(run_dir: Path) -> None:
                with self.assertRaises(RunLockError):
                    with acquire_run_lock(run_dir):
                        pass
                raise PreflightObserved

            with self.assertRaises(PreflightObserved):
                run_segment_production(
                    package,
                    run_dir=root / package.run_id,
                    locked_preflight=observe_lock,
                )

            self.assertFalse(
                segment_package_path(
                    root / package.run_id,
                    package.segment_id,
                    package.plan_version,
                ).exists()
            )


if __name__ == "__main__":
    unittest.main()