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()