import json import pytest from pydantic import ValidationError from src.atmos.models import ( AtmosValidationBuilder, AtmosValidationFindingCode, AtmosValidationMetadata, AtmosValidationRequest, AtmosValidationResult, ) def _valid_body() -> dict[str, str]: return { "task_token": "tok", "asset_upload_type": "atmos", "atmos_bucket": "atmos-bucket", "atmos_key": "k/atmos.wav", "stereo_reference_bucket": "stereo-bucket", "stereo_reference_key": "k/stereo.flac", } def _lookup_body() -> dict[str, str | bool]: return { "task_token": "tok", "asset_upload_type": "atmos", "atmos_bucket": "atmos-bucket", "atmos_key": "k/atmos.wav", "lookup_stereo_reference": True, } class TestAtmosValidationRequest: def test_parses_valid_body(self) -> None: atmos_validation_request = AtmosValidationRequest.model_validate_json(json.dumps(_valid_body())) assert atmos_validation_request.task_token == "tok" assert atmos_validation_request.atmos_bucket == "atmos-bucket" assert atmos_validation_request.atmos_key == "k/atmos.wav" assert atmos_validation_request.stereo_reference_bucket == "stereo-bucket" assert atmos_validation_request.stereo_reference_key == "k/stereo.flac" assert atmos_validation_request.lookup_stereo_reference is False assert atmos_validation_request.report_validation_result is False @pytest.mark.parametrize( "missing_field", [ "task_token", "asset_upload_type", "atmos_bucket", "atmos_key", ], ) def test_rejects_missing_field(self, missing_field: str) -> None: body = _valid_body() del body[missing_field] with pytest.raises(ValidationError, match=missing_field): AtmosValidationRequest.model_validate_json(json.dumps(body)) @pytest.mark.parametrize( "empty_field", [ "atmos_bucket", "atmos_key", "stereo_reference_bucket", "stereo_reference_key", ], ) def test_rejects_empty_string_field(self, empty_field: str) -> None: body = _valid_body() | {empty_field: ""} with pytest.raises(ValidationError, match="String should have at least 1 character"): AtmosValidationRequest.model_validate_json(json.dumps(body)) def test_rejects_extra_field(self) -> None: body = _valid_body() | {"unknown_field": "x"} with pytest.raises(ValidationError, match="unknown_field"): AtmosValidationRequest.model_validate_json(json.dumps(body)) def test_rejects_non_string_field(self) -> None: body = _valid_body() | {"atmos_bucket": 42} with pytest.raises(ValidationError, match="atmos_bucket"): AtmosValidationRequest.model_validate_json(json.dumps(body)) def test_parses_lookup_body(self) -> None: atmos_validation_request = AtmosValidationRequest.model_validate_json(json.dumps(_lookup_body())) assert atmos_validation_request.lookup_stereo_reference is True assert atmos_validation_request.stereo_reference_bucket is None assert atmos_validation_request.stereo_reference_key is None def test_report_validation_result_can_be_set(self) -> None: body = _valid_body() | {"report_validation_result": True} atmos_validation_request = AtmosValidationRequest.model_validate_json(json.dumps(body)) assert atmos_validation_request.report_validation_result is True @pytest.mark.parametrize( "explicit_field", ["stereo_reference_bucket", "stereo_reference_key"], ) def test_rejects_explicit_stereo_reference_when_lookup_true(self, explicit_field: str) -> None: body = _lookup_body() | {explicit_field: "some-value"} with pytest.raises(ValidationError, match="must be omitted when lookup_stereo_reference is True"): AtmosValidationRequest.model_validate_json(json.dumps(body)) @pytest.mark.parametrize( "missing_field", ["stereo_reference_bucket", "stereo_reference_key"], ) def test_rejects_missing_stereo_reference_when_lookup_false(self, missing_field: str) -> None: body = _valid_body() del body[missing_field] with pytest.raises(ValidationError, match="required when lookup_stereo_reference is False"): AtmosValidationRequest.model_validate_json(json.dumps(body)) class TestAtmosValidationBuilder: def test_to_result_with_no_findings_is_valid(self) -> None: result = AtmosValidationBuilder().to_result() assert result.is_valid is True assert result.errors == {} assert result.warnings == {} assert result.metadata is None def test_to_result_with_error_is_invalid(self) -> None: atmos_validation_builder = AtmosValidationBuilder() atmos_validation_builder.error(AtmosValidationFindingCode.NOT_PCM, "not pcm") result = atmos_validation_builder.to_result() assert result.is_valid is False # error is keyed by the code's metadata_key (codec for NOT_PCM) assert result.errors == {"codec": "not pcm"} assert result.warnings == {} def test_to_result_with_warning_is_valid(self) -> None: atmos_validation_builder = AtmosValidationBuilder() atmos_validation_builder.warning(AtmosValidationFindingCode.DURATION_MISMATCH, "off by 60ms") result = atmos_validation_builder.to_result() # Warnings do not flip is_valid; only errors do. assert result.is_valid is True assert result.errors == {} assert result.warnings == {"stereo_reference_duration_diff_ms": "off by 60ms"} def test_update_metadata_copies_fields_when_metadata_present(self) -> None: atmos_validation_builder = AtmosValidationBuilder() atmos_validation_builder.metadata = AtmosValidationMetadata( container="Wave", codec="PCM", codec_id="1", channels=12, sample_rate_hz=48000, bits_per_sample=24, is_truncated=False, adm_profile="Dolby Atmos Master", duration_ms=180500, ) atmos_validation_builder.update_metadata(lfe_above_400hz_dbfs=-42.0, integrated_loudness_lkfs=-18.5) assert atmos_validation_builder.metadata is not None assert atmos_validation_builder.metadata.lfe_above_400hz_dbfs == -42.0 assert atmos_validation_builder.metadata.integrated_loudness_lkfs == -18.5 # Fields not named in the update survive the copy. assert atmos_validation_builder.metadata.duration_ms == 180500 def test_update_metadata_is_a_noop_when_metadata_absent(self) -> None: # metadata is None when the atmos mediainfo failed; updating must not raise or synthesize it. atmos_validation_builder = AtmosValidationBuilder() atmos_validation_builder.update_metadata(lfe_above_400hz_dbfs=-42.0) assert atmos_validation_builder.metadata is None class TestAtmosValidationFindingCodeMetadataKey: @pytest.mark.parametrize( ("code", "expected_metadata_key"), [ pytest.param(AtmosValidationFindingCode.NOT_PCM, "codec", id="NOT_PCM"), pytest.param(AtmosValidationFindingCode.NOT_LPCM, "codec_id", id="NOT_LPCM"), pytest.param(AtmosValidationFindingCode.WRONG_BIT_DEPTH, "bits_per_sample", id="WRONG_BIT_DEPTH"), pytest.param(AtmosValidationFindingCode.WRONG_SAMPLE_RATE, "sample_rate_hz", id="WRONG_SAMPLE_RATE"), pytest.param(AtmosValidationFindingCode.NOT_WAVE, "container", id="NOT_WAVE"), pytest.param(AtmosValidationFindingCode.NOT_DOLBY_ATMOS, "adm_profile", id="NOT_DOLBY_ATMOS"), pytest.param(AtmosValidationFindingCode.ATMOS_FILE_TRUNCATED, "is_truncated", id="ATMOS_FILE_TRUNCATED"), pytest.param( AtmosValidationFindingCode.STEREO_FILE_TRUNCATED, "stereo_reference_is_truncated", id="STEREO_FILE_TRUNCATED", ), pytest.param( AtmosValidationFindingCode.DURATION_MISMATCH, "stereo_reference_duration_diff_ms", id="DURATION_MISMATCH", ), pytest.param( AtmosValidationFindingCode.SYNC_MISMATCH, "stereo_reference_alignment_offset_ms", id="SYNC_MISMATCH", ), pytest.param( AtmosValidationFindingCode.CONTENT_MISMATCH, "stereo_reference_content_match_r", id="CONTENT_MISMATCH", ), pytest.param( AtmosValidationFindingCode.LOUDNESS_TOO_HIGH, "integrated_loudness_lkfs", id="LOUDNESS_TOO_HIGH", ), pytest.param( AtmosValidationFindingCode.TRUE_PEAK_TOO_HIGH, "true_peak_dbtp", id="TRUE_PEAK_TOO_HIGH", ), pytest.param( AtmosValidationFindingCode.LFE_LEVEL_ABOVE_400HZ_TOO_HIGH, "lfe_above_400hz_dbfs", id="LFE_LEVEL_ABOVE_400HZ_TOO_HIGH", ), pytest.param( AtmosValidationFindingCode.LFE_LEVEL_ABOVE_2000HZ_TOO_HIGH, "lfe_above_2000hz_dbfs", id="LFE_LEVEL_ABOVE_2000HZ_TOO_HIGH", ), pytest.param( AtmosValidationFindingCode.SILENT_OBJECT, "silent_object_track_indices", id="SILENT_OBJECT", ), pytest.param( AtmosValidationFindingCode.SILENT_HEIGHT, "silent_height_track_indices", id="SILENT_HEIGHT", ), ], ) def test_metadata_key_for_each_code(self, code: AtmosValidationFindingCode, expected_metadata_key: str) -> None: assert code.metadata_key == expected_metadata_key def test_metadata_keys_are_unique_across_all_codes(self) -> None: # AtmosValidationBuilder writes to errors/warnings dicts keyed by metadata_key, so any # two codes that shared a key would silently overwrite each other. metadata_keys = [code.metadata_key for code in AtmosValidationFindingCode] assert len(metadata_keys) == len(set(metadata_keys)) class TestAtmosValidationMetadataDurationDiff: @pytest.fixture def base_metadata(self) -> AtmosValidationMetadata: return AtmosValidationMetadata( container="Wave", codec="PCM", codec_id="1", channels=12, sample_rate_hz=48000, bits_per_sample=24, is_truncated=False, adm_profile="Dolby Atmos Master", duration_ms=180500, ) def test_diff_is_none_until_stereo_measured(self, base_metadata: AtmosValidationMetadata) -> None: # Default state after check_mediainfo runs but before check_stereo_reference. assert base_metadata.stereo_reference_is_truncated is None assert base_metadata.stereo_reference_duration_ms is None assert base_metadata.stereo_reference_duration_diff_ms is None def test_diff_computed_when_both_durations_present(self, base_metadata: AtmosValidationMetadata) -> None: metadata = base_metadata.model_copy( update={"stereo_reference_is_truncated": False, "stereo_reference_duration_ms": 180400} ) assert metadata.stereo_reference_duration_diff_ms == 100 def test_diff_is_absolute_value(self, base_metadata: AtmosValidationMetadata) -> None: # stereo longer than atmos — diff should be positive. metadata = base_metadata.model_copy( update={"stereo_reference_is_truncated": False, "stereo_reference_duration_ms": 180600} ) assert metadata.stereo_reference_duration_diff_ms == 100 def test_diff_is_none_when_atmos_duration_missing(self, base_metadata: AtmosValidationMetadata) -> None: metadata = base_metadata.model_copy(update={"duration_ms": None, "stereo_reference_duration_ms": 180400}) assert metadata.stereo_reference_duration_diff_ms is None def test_diff_is_none_when_stereo_duration_missing(self, base_metadata: AtmosValidationMetadata) -> None: metadata = base_metadata.model_copy( update={"stereo_reference_is_truncated": False, "stereo_reference_duration_ms": None} ) assert metadata.stereo_reference_duration_diff_ms is None class TestAtmosValidationResult: def test_rejects_extra_field(self) -> None: with pytest.raises(ValidationError, match="extra_field"): AtmosValidationResult.model_validate( { "is_valid": True, "metadata": None, "errors": {}, "warnings": {}, "extra_field": "bad", }, )