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

from __future__ import annotations

import unittest

from production_build_agents.contracts.production_validation_evaluation import (
    evaluate_final_production_delivery,
    evaluate_production_readiness_report,
    evaluate_production_validation_report,
    evaluate_production_validator_candidate,
    compute_production_readiness_checks,
)
from production_build_agents.contracts.production_validation_models import (
    ProductionMediaInspection,
    ProductionValidationReport,
)
from tests.support.production_fixtures import (
    ProductionScenario,
    final_delivery,
    immutable_ref,
    readiness_records,
    text_sha256,
    validation_records,
)


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

    def test_readiness_and_validation_verdicts_are_code_owned(self) -> None:
        inspection, readiness = readiness_records(
            self.assembly_delivery,
            self.assembly,
        )
        self.assertEqual(
            evaluate_production_readiness_report(
                self.assembly_delivery,
                self.assembly,
                inspection,
                readiness,
                assembly_delivery_ref=immutable_ref("assembly-delivery"),
                media_inspection_ref=immutable_ref("media-inspection"),
            ),
            [],
        )
        failed_check = readiness.check_results[0].model_copy(
            update={"verdict": "FAIL", "reason": "失败"}
        )
        forged_readiness = readiness.model_copy(
            update={
                "check_results": [
                    failed_check,
                    *readiness.check_results[1:],
                ],
                "verdict": "PASS",
            }
        )
        codes = {
            item.code
            for item in evaluate_production_readiness_report(
                self.assembly_delivery,
                self.assembly,
                inspection,
                forged_readiness,
                assembly_delivery_ref=immutable_ref("assembly-delivery"),
                media_inspection_ref=immutable_ref("media-inspection"),
            )
        }
        self.assertIn("production_preflight_verdict_mismatch", codes)

        forged_inspection = inspection.model_copy(
            update={"audio_stream_count": 0}
        )
        codes = {
            item.code
            for item in evaluate_production_readiness_report(
                self.assembly_delivery,
                self.assembly,
                forged_inspection,
                readiness,
                assembly_delivery_ref=immutable_ref("assembly-delivery"),
                media_inspection_ref=immutable_ref("media-inspection"),
            )
        }
        self.assertIn("production_preflight_check_mismatch", codes)

        candidate, report = validation_records(self.assembly_delivery)
        self.assertEqual(
            evaluate_production_validation_report(
                self.assembly_delivery,
                self.assembly,
                inspection,
                readiness,
                candidate,
                report,
                assembly_delivery_ref=immutable_ref("assembly-delivery"),
                preflight_report_ref=immutable_ref("readiness-report"),
                media_inspection_ref=immutable_ref("media-inspection"),
                validator_candidate_ref=immutable_ref(
                    "validator-candidate"
                ),
            ),
            [],
        )

        no_boundaries = candidate.model_copy(
            update={"boundary_results": []}
        )
        forged_report = ProductionValidationReport(
            run_id=no_boundaries.run_id,
            plan_id=no_boundaries.plan_id,
            plan_version=no_boundaries.plan_version,
            assembly_delivery_ref=no_boundaries.assembly_delivery_ref,
            preflight_report_ref=no_boundaries.preflight_report_ref,
            validator_candidate_ref=report.validator_candidate_ref,
            validator_run_id=report.validator_run_id,
            verdict="PASS",
            summary=no_boundaries.summary,
        )
        codes = {
            item.code
            for item in evaluate_production_validation_report(
                self.assembly_delivery,
                self.assembly,
                inspection,
                readiness,
                no_boundaries,
                forged_report,
                assembly_delivery_ref=immutable_ref("assembly-delivery"),
                preflight_report_ref=immutable_ref("readiness-report"),
                media_inspection_ref=immutable_ref("media-inspection"),
                validator_candidate_ref=immutable_ref(
                    "validator-candidate"
                ),
            )
        }
        self.assertIn(
            "production_validation_boundaries_mismatch",
            codes,
        )
        self.assertIn("production_validation_verdict_mismatch", codes)

    def test_python_renames_preserve_current_json_fields(self) -> None:
        _, readiness = readiness_records(
            self.assembly_delivery,
            self.assembly,
        )
        delivery = final_delivery(self.assembly_delivery)

        readiness_payload = readiness.model_dump(mode="json")
        delivery_payload = delivery.model_dump(mode="json")

        self.assertEqual(readiness_payload["schema_version"], "0.7")
        self.assertIn("coverage_evaluation", readiness_payload)
        self.assertNotIn("production_coverage", readiness_payload)
        self.assertEqual(delivery_payload["schema_version"], "0.7")
        self.assertIn("preflight_report_ref", delivery_payload)
        self.assertNotIn("readiness_report_ref", delivery_payload)

    def test_candidate_rules_are_the_formal_report_rules(self) -> None:
        inspection, readiness = readiness_records(
            self.assembly_delivery,
            self.assembly,
        )
        candidate, report = validation_records(self.assembly_delivery)
        invalid = candidate.model_copy(
            update={
                "criterion_results": candidate.criterion_results[:-1],
                "boundary_results": [],
            }
        )
        candidate_codes = {
            item.code
            for item in evaluate_production_validator_candidate(
                self.assembly_delivery,
                invalid,
                assembly_delivery_ref=immutable_ref("assembly-delivery"),
                preflight_report_ref=immutable_ref("readiness-report"),
            )
        }
        report_codes = {
            item.code
            for item in evaluate_production_validation_report(
                self.assembly_delivery,
                self.assembly,
                inspection,
                readiness,
                invalid,
                report,
                assembly_delivery_ref=immutable_ref("assembly-delivery"),
                preflight_report_ref=immutable_ref("readiness-report"),
                media_inspection_ref=immutable_ref("media-inspection"),
                validator_candidate_ref=immutable_ref(
                    "validator-candidate"
                ),
            )
        }
        self.assertEqual(
            candidate_codes,
            {
                "production_validation_criteria_mismatch",
                "production_validation_boundaries_mismatch",
            },
        )
        self.assertTrue(candidate_codes.issubset(report_codes))

    def test_media_inspection_can_record_bad_media_as_fail(self) -> None:
        inspection, _ = readiness_records(
            self.assembly_delivery,
            self.assembly,
        )
        wrong_container = ProductionMediaInspection.model_validate(
            {
                **inspection.model_dump(mode="python"),
                "actual_output_profile": {
                    **inspection.actual_output_profile.model_dump(
                        mode="python"
                    ),
                    "container": "matroska",
                },
            }
        )
        self.assertFalse(
            compute_production_readiness_checks(
                self.assembly_delivery,
                self.assembly,
                wrong_container,
            )["output_profile"]
        )

        no_subtitles = ProductionMediaInspection.model_validate(
            {
                **inspection.model_dump(mode="python"),
                "subtitle_cue_count": 0,
                "subtitle_first_start_ms": None,
                "subtitle_last_end_ms": None,
            }
        )
        self.assertFalse(
            compute_production_readiness_checks(
                self.assembly_delivery,
                self.assembly,
                no_subtitles,
            )["subtitle_integrity"]
        )

        zero_frame_rate = inspection.model_copy(
            update={
                "actual_output_profile": (
                    inspection.actual_output_profile.model_copy(
                        update={
                            "frame_rate_numerator": 0,
                            "frame_rate_denominator": 0,
                        }
                    )
                )
            }
        )
        self.assertFalse(
            compute_production_readiness_checks(
                self.assembly_delivery,
                self.assembly,
                zero_frame_rate,
            )["output_profile"]
        )

    def test_media_inspection_binds_audio_master_content(self) -> None:
        inspection, _ = readiness_records(
            self.assembly_delivery,
            self.assembly,
        )
        artifacts = [
            (
                item.model_copy(
                    update={"content_sha256": text_sha256("tampered-audio")}
                )
                if item.artifact_id
                == self.assembly_delivery.audio_master_artifact_id
                else item
            )
            for item in self.assembly_delivery.artifacts
        ]
        tampered = self.assembly_delivery.model_copy(
            update={"artifacts": artifacts}
        )
        self.assertFalse(
            compute_production_readiness_checks(
                tampered,
                self.assembly,
                inspection,
            )["artifact_integrity"]
        )

    def test_final_delivery_requires_both_validation_layers_to_pass(
        self,
    ) -> None:
        inspection, readiness = readiness_records(
            self.assembly_delivery,
            self.assembly,
        )
        candidate, report = validation_records(self.assembly_delivery)
        final = final_delivery(self.assembly_delivery)
        self.assertEqual(self.scenario.final_issues(), [])
        failed_report = report.model_copy(update={"verdict": "FAIL"})
        codes = {
            item.code
            for item in evaluate_final_production_delivery(
                self.assembly_delivery,
                self.assembly,
                inspection,
                readiness,
                candidate,
                failed_report,
                final,
                production_package_ref=(
                    self.assembly_delivery.production_package_ref
                ),
                assembly_delivery_ref=immutable_ref("assembly-delivery"),
                preflight_report_ref=immutable_ref("readiness-report"),
                media_inspection_ref=immutable_ref("media-inspection"),
                validator_candidate_ref=immutable_ref(
                    "validator-candidate"
                ),
                validation_report_ref=immutable_ref("validation-report"),
            )
        }
        self.assertIn("final_production_not_passed", codes)

        forged_refs = final.model_copy(
            update={
                "assembly_delivery_ref": immutable_ref("forged-assembly"),
                "validation_report_ref": immutable_ref("forged-validation"),
            }
        )
        codes = {
            item.code
            for item in evaluate_final_production_delivery(
                self.assembly_delivery,
                self.assembly,
                inspection,
                readiness,
                candidate,
                report,
                forged_refs,
                production_package_ref=(
                    self.assembly_delivery.production_package_ref
                ),
                assembly_delivery_ref=immutable_ref("assembly-delivery"),
                preflight_report_ref=immutable_ref("readiness-report"),
                media_inspection_ref=immutable_ref("media-inspection"),
                validator_candidate_ref=immutable_ref(
                    "validator-candidate"
                ),
                validation_report_ref=immutable_ref("validation-report"),
            )
        }
        self.assertIn("final_production_document_ref_mismatch", codes)

    def test_validation_report_references_candidate_without_copying_it(
        self,
    ) -> None:
        _, report = validation_records(self.assembly_delivery)
        self.assertNotIn(
            "criterion_results",
            ProductionValidationReport.model_fields,
        )
        self.assertNotIn(
            "boundary_results",
            ProductionValidationReport.model_fields,
        )
        self.assertEqual(
            report.validator_candidate_ref,
            immutable_ref("validator-candidate"),
        )



if __name__ == "__main__":
    import unittest

    unittest.main()