video-image-production-build/tests/contracts/test_production_progress.py

from __future__ import annotations

import unittest

from pydantic import ValidationError

from production_build_agents.contracts.production_planning_evaluation import (
    evaluate_adapt_planning_package,
    evaluate_adapt_transition,
    evaluate_production_progress_decision,
    evaluate_production_progress_package,
)
from production_build_agents.contracts.production_planning_models import (
    ProductionPlanningPackage,
    ProductionProgressDecision,
    ProductionProgressPackage,
)
from tests.support.production_fixtures import (
    immutable_ref,
    output_profile,
    production_plan,
    segment_bundle,
    shared_anchor,
)


class ProductionProgressContractTest(unittest.TestCase):
    def setUp(self) -> None:
        self.segment1 = segment_bundle(1)
        self.segment2 = segment_bundle(2)
        self.plan = production_plan(
            [self.segment1[0], self.segment2[0]]
        )
        self.planning = ProductionPlanningPackage(
            run_id="production-run",
            mode="INITIAL",
            plan_id=self.plan.plan_id,
            target_plan_version=1,
            production_brief_ref=immutable_ref("brief"),
            global_data_delivery_ref=immutable_ref("global-delivery"),
            production_input_catalog_ref=immutable_ref("catalog"),
            output_profile=output_profile(),
            shared_visual_anchor=shared_anchor(),
        )
        segment_package, _, delivery, _, report, accepted = self.segment1
        self.segment_package_ref = accepted.segment_package_ref.model_copy(
            update={
                "uri": delivery.segment_package_uri,
                "document_sha256": delivery.segment_package_sha256,
            }
        )
        self.delivery_ref = accepted.segment_delivery_ref
        self.report_ref = accepted.validation_report_ref
        self.accepted = accepted.model_copy(
            update={"segment_package_ref": self.segment_package_ref}
        )
        self.progress_ref = immutable_ref("progress-Segment1-v1")
        self.progress = ProductionProgressPackage(
            run_id="production-run",
            plan_id=self.plan.plan_id,
            plan_version=1,
            segment_id="Segment1",
            planning_package_ref=self.plan.planning_package_ref,
            production_plan_ref=immutable_ref("production-plan"),
            segment_package_ref=self.segment_package_ref,
            segment_delivery_ref=self.delivery_ref,
            segment_validation_report_ref=self.report_ref,
            accepted_segments=[self.accepted],
            remaining_plan_revisions=5,
        )

    @staticmethod
    def _codes(issues) -> set[str]:
        return {item.code for item in issues}

    def _package_issues(self, package=None):
        segment_package, _, delivery, _, report, _ = self.segment1
        return evaluate_production_progress_package(
            package or self.progress,
            planning_package=self.planning,
            planning_package_ref=self.plan.planning_package_ref,
            production_plan=self.plan,
            production_plan_ref=immutable_ref("production-plan"),
            segment_package=segment_package,
            segment_package_ref=self.segment_package_ref,
            segment_delivery=delivery,
            segment_delivery_ref=self.delivery_ref,
            segment_validation_report=report,
            segment_validation_report_ref=self.report_ref,
            accepted_segments=[self.accepted],
            remaining_plan_revisions=5,
        )

    def _decision(self, action: str, affected=None):
        return ProductionProgressDecision(
            run_id=self.progress.run_id,
            plan_id=self.progress.plan_id,
            plan_version=self.progress.plan_version,
            segment_id=self.progress.segment_id,
            progress_package_ref=self.progress_ref,
            action=action,
            affected_segment_ids=affected or [],
            reason="正式进度审阅",
        )

    def _decision_codes(self, decision, *, report=None, package=None):
        return self._codes(
            evaluate_production_progress_decision(
                package or self.progress,
                decision,
                progress_package_ref=self.progress_ref,
                production_plan=self.plan,
                segment_validation_report=report or self.segment1[4],
            )
        )

    def test_progress_package_binds_formal_pass_chain(self) -> None:
        self.assertEqual(self._package_issues(), [])
        wrong = self.progress.model_copy(
            update={"production_plan_ref": immutable_ref("wrong-plan")}
        )
        self.assertIn(
            "production_progress_document_ref_mismatch",
            self._codes(self._package_issues(wrong)),
        )

    def test_progress_models_reject_duplicate_and_invalid_action_shapes(
        self,
    ) -> None:
        with self.assertRaises(ValidationError):
            ProductionProgressPackage(
                **self.progress.model_dump(exclude={"accepted_segments"}),
                accepted_segments=[self.accepted, self.accepted],
            )
        with self.assertRaises(ValidationError):
            self._decision("ADAPT")
        with self.assertRaises(ValidationError):
            self._decision("CONTINUE", ["Segment2"])

    def test_pass_allows_continue_adapt_stop_but_not_replan(self) -> None:
        for action in ("CONTINUE", "STOP"):
            with self.subTest(action=action):
                self.assertEqual(
                    self._decision_codes(self._decision(action)),
                    set(),
                )
        self.assertEqual(
            self._decision_codes(
                self._decision("ADAPT", ["Segment2"])
            ),
            set(),
        )
        self.assertIn(
            "production_progress_action_verdict_mismatch",
            self._decision_codes(self._decision("REPLAN")),
        )

    def test_fail_allows_replan_stop_and_rejects_continue(self) -> None:
        report = self.segment1[4].model_copy(update={"verdict": "FAIL"})
        failed_progress = self.progress.model_copy(
            update={"accepted_segments": []}
        )
        self.assertEqual(
            self._decision_codes(
                self._decision("REPLAN"),
                report=report,
                package=failed_progress,
            ),
            set(),
        )
        self.assertIn(
            "production_progress_action_verdict_mismatch",
            self._decision_codes(
                self._decision("CONTINUE"),
                report=report,
                package=failed_progress,
            ),
        )

    def test_adapt_requires_pending_segment_and_revision_budget(self) -> None:
        self.assertIn(
            "production_progress_adapt_segment_not_pending",
            self._decision_codes(
                self._decision("ADAPT", ["Segment1"])
            ),
        )
        exhausted = self.progress.model_copy(
            update={"remaining_plan_revisions": 0}
        )
        self.assertIn(
            "production_progress_revision_budget_exhausted",
            self._decision_codes(
                self._decision("ADAPT", ["Segment2"]),
                package=exhausted,
            ),
        )

    def test_adapt_package_and_transition_are_future_only(self) -> None:
        decision_ref = immutable_ref("adapt-decision")
        package = ProductionPlanningPackage(
            **self.planning.model_dump(
                exclude={
                    "mode",
                    "target_plan_version",
                    "previous_plan_ref",
                    "failure_report_ref",
                    "progress_decision_ref",
                    "accepted_segments",
                    "authorized_revision_segment_ids",
                }
            ),
            mode="ADAPT",
            target_plan_version=2,
            previous_plan_ref=immutable_ref("production-plan"),
            progress_decision_ref=decision_ref,
            accepted_segments=[self.accepted],
            authorized_revision_segment_ids=["Segment2"],
        )
        changed_segment = self.plan.segments[1].model_copy(
            update={"objective": "根据最新进度调整未来段落"}
        )
        current = self.plan.model_copy(
            update={
                "plan_version": 2,
                "planning_package_ref": immutable_ref("planning-v2"),
                "segments": [self.plan.segments[0], changed_segment],
                "summary": "未来规划已调整",
            }
        )
        decision = self._decision("ADAPT", ["Segment2"])
        self.assertEqual(
            evaluate_adapt_planning_package(
                self.planning,
                package,
                progress_package=self.progress,
                progress_package_ref=self.progress_ref,
                progress_decision=decision,
                progress_decision_ref=decision_ref,
            ),
            [],
        )
        widened = package.model_copy(
            update={
                "authorized_revision_segment_ids": [
                    "Segment1",
                    "Segment2",
                ]
            }
        )
        self.assertIn(
            "adapt_authorized_segments_mismatch",
            self._codes(
                evaluate_adapt_planning_package(
                    self.planning,
                    widened,
                    progress_package=self.progress,
                    progress_package_ref=self.progress_ref,
                    progress_decision=decision,
                    progress_decision_ref=decision_ref,
                )
            ),
        )
        stale_snapshot = package.model_copy(
            update={"accepted_segments": []}
        )
        self.assertIn(
            "adapt_accepted_snapshot_mismatch",
            self._codes(
                evaluate_adapt_planning_package(
                    self.planning,
                    stale_snapshot,
                    progress_package=self.progress,
                    progress_package_ref=self.progress_ref,
                    progress_decision=decision,
                    progress_decision_ref=decision_ref,
                )
            ),
        )
        self.assertEqual(
            evaluate_adapt_transition(self.plan, current, package),
            [],
        )
        no_effect = current.model_copy(
            update={"segments": self.plan.segments}
        )
        self.assertIn(
            "adapt_has_no_effect",
            self._codes(
                evaluate_adapt_transition(
                    self.plan,
                    no_effect,
                    package,
                )
            ),
        )
        accepted_changed = current.model_copy(
            update={
                "segments": [
                    self.plan.segments[0].model_copy(
                        update={"objective": "越权修改"}
                    ),
                    changed_segment,
                ]
            }
        )
        self.assertIn(
            "adapt_protected_segment_changed",
            self._codes(
                evaluate_adapt_transition(
                    self.plan,
                    accepted_changed,
                    package,
                )
            ),
        )


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