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

from __future__ import annotations

import unittest

from production_build_agents.contracts.models import (
    ExpectationEvaluation,
    RequirementEvaluation,
)
from production_build_agents.contracts.production_assembly_evaluation import (
    evaluate_production_assembly_delivery,
)
from production_build_agents.contracts.production_coverage import (
    build_production_coverage,
    segment_artifact_ref_key,
)
from tests.support.production_fixtures import (
    ProductionScenario,
    immutable_ref,
)


class ProductionCoverageContractTest(unittest.TestCase):
    def setUp(self) -> None:
        self.scenario = ProductionScenario()
        self.__dict__.update(vars(self.scenario))

    def test_missing_segment_package_returns_contract_issue(self) -> None:
        codes = {
            item.code
            for item in evaluate_production_assembly_delivery(
                self.planning,
                self.plan,
                self.global_delivery,
                self.production_package,
                self.assembly,
                {"Segment1": self.package_map["Segment1"]},
                {"Segment1": self.delivery_map["Segment1"]},
                self.assembly_delivery,
                production_package_ref=immutable_ref("production-package"),
                assembly_package_ref=immutable_ref("assembly-package"),
                segment_package_refs={
                    "Segment1": self.segment_package_refs["Segment1"]
                },
            )
        }
        self.assertIn("assembly_segment_package_set_mismatch", codes)
        self.assertIn("production_coverage_missing_segment_package", codes)

    def test_unconsumed_satisfied_global_requirement_does_not_fail(
        self,
    ) -> None:
        omitted = self.global_delivery.model_copy(
            update={
                "requirement_evaluations": [
                    *self.global_delivery.requirement_evaluations,
                    RequirementEvaluation(
                        requirement_id="Requirement3",
                        expectation_evaluations=[
                            ExpectationEvaluation(
                                expectation_id=(
                                    "Requirement3-Expectation1"
                                ),
                                expected_count=1,
                                actual_count=1,
                                artifact_ids=["Task3-v1-artifact-1"],
                                satisfied=True,
                            )
                        ],
                    ),
                ]
            }
        )
        coverage = build_production_coverage(
            self.assembly,
            self.package_map,
            self.delivery_map,
            {
                segment_artifact_ref_key(artifact)
                for item in self.assembly.accepted_segments
                for artifact in (
                    item.primary_artifact,
                    item.subtitle_artifact,
                )
            },
            omitted,
        )
        self.assertTrue(coverage.satisfied)
        self.assertNotIn(
            "Requirement3-Expectation1",
            coverage.covered_expectation_ids,
        )

    def test_unsatisfied_global_requirement_cannot_vacuously_pass(
        self,
    ) -> None:
        unsatisfied = self.global_delivery.model_copy(
            update={
                "requirement_evaluations": [
                    *self.global_delivery.requirement_evaluations,
                    RequirementEvaluation(
                        requirement_id="Requirement3",
                        expectation_evaluations=[
                            ExpectationEvaluation(
                                expectation_id=(
                                    "Requirement3-Expectation1"
                                ),
                                expected_count=1,
                                actual_count=0,
                                artifact_ids=[],
                                satisfied=False,
                            )
                        ],
                    ),
                ]
            }
        )
        coverage = build_production_coverage(
            self.assembly,
            self.package_map,
            self.delivery_map,
            {
                segment_artifact_ref_key(artifact)
                for item in self.assembly.accepted_segments
                for artifact in (
                    item.primary_artifact,
                    item.subtitle_artifact,
                )
            },
            unsatisfied,
        )
        self.assertFalse(coverage.satisfied)
        self.assertNotIn(
            "Requirement3",
            coverage.covered_requirement_ids,
        )
        self.assertIn(
            "Requirement3-Expectation1",
            coverage.required_expectation_ids,
        )

    def test_planned_but_unused_artifact_is_not_reported_covered(
        self,
    ) -> None:
        package = self.package_map["Segment1"]
        delivery = self.delivery_map["Segment1"]
        source_id = package.artifact_inputs[0].artifact_id
        anchor_id = package.shared_visual_anchor.anchor_id
        lineages = [
            lineage.model_copy(
                update={
                    "input_artifact_ids": [
                        anchor_id if item == source_id else item
                        for item in lineage.input_artifact_ids
                    ]
                }
            )
            for lineage in delivery.artifact_graph.lineages
        ]
        changed = delivery.model_copy(
            update={
                "artifact_graph": delivery.artifact_graph.model_copy(
                    update={"lineages": lineages}
                )
            }
        )
        coverage = build_production_coverage(
            self.assembly,
            self.package_map,
            {**self.delivery_map, "Segment1": changed},
            {
                segment_artifact_ref_key(artifact)
                for item in self.assembly.accepted_segments
                for artifact in (
                    item.primary_artifact,
                    item.subtitle_artifact,
                )
            },
            self.global_delivery,
        )

        segment = coverage.segment_results[0]
        self.assertIn(source_id, segment.required_global_artifact_ids)
        self.assertNotIn(source_id, segment.covered_global_artifact_ids)
        self.assertTrue(segment.satisfied)
        self.assertTrue(coverage.satisfied)

    def test_unplanned_but_actually_used_artifact_is_reported_covered(
        self,
    ) -> None:
        package = self.package_map["Segment1"]
        delivery = self.delivery_map["Segment1"]
        planned_id = package.artifact_inputs[0].artifact_id
        extra_id = "Task3-v1-artifact-1"
        extra_artifact = self.global_delivery.active_artifacts[0].model_copy(
            update={
                "artifact_id": extra_id,
                "uri": "/runs/global/artifacts/task-3.json",
                "content_sha256": "3" * 64,
                "description": "未由 Planner 选择但实际使用的素材",
            }
        )
        changed_global = self.global_delivery.model_copy(
            update={
                "active_artifacts": [
                    *self.global_delivery.active_artifacts,
                    extra_artifact,
                ],
                "requirement_evaluations": [
                    *self.global_delivery.requirement_evaluations,
                    RequirementEvaluation(
                        requirement_id="Requirement3",
                        expectation_evaluations=[
                            ExpectationEvaluation(
                                expectation_id=(
                                    "Requirement3-Expectation1"
                                ),
                                expected_count=1,
                                actual_count=1,
                                artifact_ids=[extra_id],
                                satisfied=True,
                            )
                        ],
                        actual_artifact_ids=[extra_id],
                    ),
                ],
            }
        )
        lineages = [
            lineage.model_copy(
                update={
                    "input_artifact_ids": [
                        extra_id if item == planned_id else item
                        for item in lineage.input_artifact_ids
                    ]
                }
            )
            for lineage in delivery.artifact_graph.lineages
        ]
        changed_delivery = delivery.model_copy(
            update={
                "artifact_graph": delivery.artifact_graph.model_copy(
                    update={"lineages": lineages}
                )
            }
        )
        coverage = build_production_coverage(
            self.assembly,
            self.package_map,
            {**self.delivery_map, "Segment1": changed_delivery},
            {
                segment_artifact_ref_key(artifact)
                for item in self.assembly.accepted_segments
                for artifact in (
                    item.primary_artifact,
                    item.subtitle_artifact,
                )
            },
            changed_global,
        )

        segment = coverage.segment_results[0]
        self.assertIn(planned_id, segment.required_global_artifact_ids)
        self.assertNotIn(planned_id, segment.covered_global_artifact_ids)
        self.assertIn(extra_id, segment.covered_global_artifact_ids)
        self.assertIn(
            "Requirement3-Expectation1",
            segment.covered_expectation_ids,
        )
        self.assertTrue(coverage.satisfied)

    def test_missing_production_input_still_fails_coverage(self) -> None:
        delivery = self.delivery_map["Segment1"]
        changed_delivery = delivery.model_copy(
            update={
                "artifact_graph": delivery.artifact_graph.model_copy(
                    update={
                        "lineages": [
                            lineage.model_copy(
                                update={"production_input_ids": []}
                            )
                            for lineage in (
                                delivery.artifact_graph.lineages
                            )
                        ]
                    }
                )
            }
        )
        coverage = build_production_coverage(
            self.assembly,
            self.package_map,
            {**self.delivery_map, "Segment1": changed_delivery},
            {
                segment_artifact_ref_key(artifact)
                for item in self.assembly.accepted_segments
                for artifact in (
                    item.primary_artifact,
                    item.subtitle_artifact,
                )
            },
            self.global_delivery,
        )

        segment = coverage.segment_results[0]
        self.assertEqual(segment.covered_production_input_ids, [])
        self.assertFalse(segment.satisfied)
        self.assertIn("Segment1", coverage.reachable_segment_ids)
        self.assertFalse(coverage.satisfied)

    def test_unassigned_package_input_is_not_required_coverage(self) -> None:
        package = self.package_map["Segment1"]
        unassigned = package.production_inputs[0].model_copy(
            update={
                "input_id": "ProductionInput-" + "f" * 64,
                "source_path": "$.制作表.关系结果[0]",
                "content_sha256": "e" * 64,
                "value": {"关系": "由整片而非当前镜头验收"},
            }
        )
        changed_package = package.model_copy(
            update={
                "production_inputs": [
                    *package.production_inputs,
                    unassigned,
                ]
            }
        )
        coverage = build_production_coverage(
            self.assembly,
            {**self.package_map, "Segment1": changed_package},
            self.delivery_map,
            {
                segment_artifact_ref_key(artifact)
                for item in self.assembly.accepted_segments
                for artifact in (
                    item.primary_artifact,
                    item.subtitle_artifact,
                )
            },
            self.global_delivery,
        )

        segment = coverage.segment_results[0]
        self.assertEqual(
            segment.required_production_input_ids,
            package.shots[0].production_input_ids,
        )
        self.assertNotIn(
            unassigned.input_id,
            segment.covered_production_input_ids,
        )
        self.assertTrue(segment.satisfied)
        self.assertTrue(coverage.satisfied)



if __name__ == "__main__":
    import unittest

    unittest.main()