video-image-production-build/tests/production/test_runtime_boundary.py

from __future__ import annotations

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

from production_build_agents.production.profile import default_output_profile
from production_build_agents.production.runtime import (
    production_request_sha256,
    validate_production_run_directory_boundary,
)
from run_production import _build_parser, main


class ProductionRunDirectoryBoundaryTest(unittest.TestCase):
    def test_request_digest_uses_current_protocol_domain(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            delivery = root / "delivery.json"
            anchor = root / "anchor.png"
            delivery.write_bytes(b"delivery")
            anchor.write_bytes(b"anchor")

            old_digest = hashlib.sha256()
            old_digest.update(b"production-protocol-0.6\0")
            old_digest.update(delivery.read_bytes())
            old_digest.update(b"\0shared-visual-anchor\0")
            old_digest.update(anchor.read_bytes())
            old_digest.update(b"\0output-profile\0")
            old_digest.update(
                json.dumps(
                    default_output_profile().model_dump(mode="json"),
                    ensure_ascii=False,
                    sort_keys=True,
                    separators=(",", ":"),
                ).encode("utf-8")
            )
            actual = production_request_sha256(
                delivery,
                shared_visual_anchor_path=anchor,
            )

            self.assertNotEqual(actual, old_digest.hexdigest())

    def test_cli_exposes_only_shared_plan_revision_budget(self) -> None:
        option_strings = {
            option
            for action in _build_parser()._actions
            for option in action.option_strings
        }
        self.assertIn("--max-plan-revisions", option_strings)
        self.assertNotIn("--max-replans", option_strings)

    def test_cli_exposes_progress_review_pause(self) -> None:
        parser = _build_parser()
        option_strings = {
            option
            for action in parser._actions
            for option in action.option_strings
        }
        self.assertIn("--pause-after-progress-review", option_strings)
        args = parser.parse_args(
            [
                "delivery.json",
                "--shared-visual-anchor",
                "anchor.png",
                "--pause-after-progress-review",
            ]
        )
        self.assertTrue(args.pause_after_progress_review)

    def test_cli_treats_requested_running_pause_as_success(self) -> None:
        argv = [
            "run_production.py",
            "delivery.json",
            "--shared-visual-anchor",
            "anchor.png",
            "--pause-after-progress-review",
        ]
        with (
            patch("sys.argv", argv),
            patch(
                "run_production.run_full_production",
                return_value={
                    "status": "RUNNING",
                    "phase": "PREPARE_SEGMENT",
                },
            ) as run,
        ):
            main()

        self.assertTrue(
            run.call_args.kwargs["pause_after_progress_review"]
        )

    def test_cli_still_fails_for_unrequested_running_result(self) -> None:
        argv = [
            "run_production.py",
            "delivery.json",
            "--shared-visual-anchor",
            "anchor.png",
        ]
        with (
            patch("sys.argv", argv),
            patch(
                "run_production.run_full_production",
                return_value={"status": "RUNNING"},
            ),
            self.assertRaises(SystemExit) as raised,
        ):
            main()

        self.assertEqual(raised.exception.code, 1)

    def test_new_and_empty_directories_are_valid(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            validate_production_run_directory_boundary(root / "missing")
            empty = root / "empty"
            empty.mkdir()
            validate_production_run_directory_boundary(empty)

    def test_non_production_evidence_is_outside_scan_scope(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            (root / "global_data_stage_delivery.json").write_text(
                '{"schema_version":"0.3"}',
                encoding="utf-8",
            )
            (root / "production_brief.json").write_text(
                '{"brief":true}',
                encoding="utf-8",
            )
            for directory in ("agent_runs", "tool_operations"):
                target = root / directory
                target.mkdir()
                (target / "evidence.json").write_text(
                    '{"schema_version":"0.4"}',
                    encoding="utf-8",
                )
            validate_production_run_directory_boundary(root)

    def test_formal_records_are_objects_with_current_schema(self) -> None:
        invalid_payloads = (
            [],
            {"plan_id": "missing-version"},
            {"schema_version": "0.4"},
            {"schema_version": "0.5"},
        )
        for index, payload in enumerate(invalid_payloads):
            with self.subTest(payload=payload), tempfile.TemporaryDirectory() as temp_dir:
                root = Path(temp_dir)
                plans = root / "production_plans"
                plans.mkdir()
                path = plans / f"production_dag.v{index + 1}.json"
                path.write_text(
                    json.dumps(payload),
                    encoding="utf-8",
                )
                before = path.read_bytes()
                with self.assertRaisesRegex(ValueError, "Production"):
                    validate_production_run_directory_boundary(root)
                self.assertEqual(path.read_bytes(), before)
                self.assertEqual(
                    [item.relative_to(root) for item in root.rglob("*")],
                    [
                        Path("production_plans"),
                        path.relative_to(root),
                    ],
                )

        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            plans = root / "production_plans"
            plans.mkdir()
            (plans / "production_dag.v1.json").write_text(
                '{"schema_version":"0.7"}',
                encoding="utf-8",
            )
            validate_production_run_directory_boundary(root)

    def test_mixed_run_is_rejected_without_writes(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            plans = root / "production_plans"
            segments = root / "segment_packages"
            plans.mkdir()
            segments.mkdir()
            (plans / "production_dag.v1.json").write_text(
                '{"schema_version":"0.7"}',
                encoding="utf-8",
            )
            old_segment = segments / "Segment1.v1.json"
            old_segment.write_text(
                '{"schema_version":"0.6"}',
                encoding="utf-8",
            )
            before = {
                path.relative_to(root): path.read_bytes()
                for path in root.rglob("*")
                if path.is_file()
            }
            with self.assertRaisesRegex(ValueError, "非当前协议"):
                validate_production_run_directory_boundary(root)
            after = {
                path.relative_to(root): path.read_bytes()
                for path in root.rglob("*")
                if path.is_file()
            }
            self.assertEqual(after, before)

    def test_legacy_run_is_rejected_without_writes(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            plans = root / "production_plans"
            plans.mkdir()
            legacy = plans / "production_dag.v1.json"
            legacy.write_text(
                '{"schema_version":"0.6"}',
                encoding="utf-8",
            )
            before = legacy.read_bytes()

            with self.assertRaisesRegex(ValueError, "非当前协议"):
                validate_production_run_directory_boundary(root)

            self.assertEqual(legacy.read_bytes(), before)
            self.assertEqual(
                [path.relative_to(root) for path in root.rglob("*")],
                [
                    Path("production_plans"),
                    Path("production_plans/production_dag.v1.json"),
                ],
            )

    def test_progress_records_use_current_schema_and_segment_version_name(
        self,
    ) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            for directory in (
                "production_progress_packages",
                "production_progress_decisions",
            ):
                target = root / directory
                target.mkdir()
                (target / "Segment1.v2.json").write_text(
                    '{"schema_version":"0.7"}',
                    encoding="utf-8",
                )
            validate_production_run_directory_boundary(root)

    def test_obsolete_unversioned_path_is_rejected(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            old_package = root / "production_package.json"
            old_package.write_text(
                '{"schema_version":"0.5"}',
                encoding="utf-8",
            )
            with self.assertRaisesRegex(ValueError, "旧版正式路径"):
                validate_production_run_directory_boundary(root)


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