from typing import Any from unittest.mock import MagicMock, patch import pytest from pydantic import AnyUrl from src.atmos.mediainfo_check import ( AssetFetchError, MediainfoCrashError, MediainfoError, MediainfoOutputError, MediainfoUnavailable, _MediainfoMedia, _parse_mediainfo, check_mediainfo, run_mediainfo, ) from src.atmos.models import ( AtmosValidationBuilder, AtmosValidationFindingCode, AtmosValidationMetadata, MediainfoOutput, ) _URL = AnyUrl("file:///test.wav") def _parse_response(data: dict[str, Any]) -> MediainfoOutput: return _parse_mediainfo(_MediainfoMedia.model_validate(data["media"])) def _valid_atmos_output() -> MediainfoOutput: return MediainfoOutput( 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=7805, ) class TestRunMediainfoErrors: def test_binary_missing(self) -> None: with ( patch("src.atmos.mediainfo_check.subprocess.run", side_effect=FileNotFoundError("mediainfo")), pytest.raises(MediainfoUnavailable, match="could not execute mediainfo"), ): run_mediainfo(_URL) def test_killed_by_signal(self) -> None: proc = MagicMock(returncode=-9, stderr="") with ( patch("src.atmos.mediainfo_check.subprocess.run", return_value=proc), pytest.raises(MediainfoCrashError, match="killed by signal 9"), ): run_mediainfo(_URL) def test_nonzero_exit(self) -> None: proc = MagicMock(returncode=1, stderr="something broke") with ( patch("src.atmos.mediainfo_check.subprocess.run", return_value=proc), pytest.raises(MediainfoError, match="exited with code 1"), ): run_mediainfo(_URL) def test_invalid_json(self) -> None: proc = MagicMock(returncode=0, stdout="not json") with ( patch("src.atmos.mediainfo_check.subprocess.run", return_value=proc), pytest.raises(MediainfoOutputError, match="output unexpected"), ): run_mediainfo(_URL) @pytest.mark.parametrize( "stderr", [ pytest.param("", id="silent_fetch_failure"), pytest.param("E: , Could not resolve host: ...", id="dns_failure"), pytest.param("E: , Failed to connect to ...", id="connection_refused"), ], ) def test_null_media_raises_asset_fetch_error(self, stderr: str) -> None: proc = MagicMock(returncode=0, stdout='{"media":null}', stderr=stderr) with ( patch("src.atmos.mediainfo_check.subprocess.run", return_value=proc), pytest.raises(AssetFetchError, match="could not fetch asset"), ): run_mediainfo(_URL) def test_empty_tracks_raises_mediainfo_output_error(self) -> None: proc = MagicMock(returncode=0, stdout='{"media":{"track":[]}}', stderr="") with ( patch("src.atmos.mediainfo_check.subprocess.run", return_value=proc), pytest.raises(MediainfoOutputError, match="output unexpected"), ): run_mediainfo(_URL) _VALID_ATMOS_DATA: dict[str, Any] = { "media": { "track": [ {"@type": "General", "Format": "Wave", "Duration": "7.805", "extra": {}}, { "@type": "Audio", "Format": "PCM", "CodecID": "1", # WAVE_FORMAT_PCM (0x0001) — required by Dolby ADM Profile spec "Channels": "12", "SamplingRate": "48000", "BitDepth": "24", "extra": { "AdmProfile_Format": "Dolby Atmos Master", }, }, ], }, } class TestParseMediainfoJson: def test_parses_valid_atmos_data(self) -> None: output = _parse_response(_VALID_ATMOS_DATA) assert output.container == "Wave" assert output.codec == "PCM" assert output.codec_id == "1" assert output.channels == 12 assert output.sample_rate_hz == 48000 assert output.bits_per_sample == 24 assert output.duration_ms == 7805 assert output.adm_profile == "Dolby Atmos Master" assert output.is_truncated is False @pytest.mark.parametrize( "codec_id", [ pytest.param("1", id="legacy_wave_format_pcm"), pytest.param("00000001-0000-0010-8000-00AA00389B71", id="wave_format_extensible_guid"), ], ) def test_codec_id_stored_as_string(self, codec_id: str) -> None: data: dict[str, Any] = { "media": { "track": [ {"@type": "General", "Format": "Wave"}, {"@type": "Audio", "Format": "PCM", "CodecID": codec_id, "SamplingRate": "48000"}, ], }, } output = _parse_response(data) assert output.codec_id == codec_id @pytest.mark.parametrize( ("duration_seconds_str", "expected_ms"), [ pytest.param("7.805", 7805, id="three_decimals"), pytest.param("499.840", 499840, id="long_duration"), pytest.param("0.001", 1, id="one_ms"), pytest.param("123", 123_000, id="integer_seconds"), ], ) def test_duration_parsed_to_ms(self, duration_seconds_str: str, expected_ms: int) -> None: data: dict[str, Any] = { "media": { "track": [ {"@type": "General", "Format": "Wave", "Duration": duration_seconds_str}, {"@type": "Audio", "Format": "PCM", "CodecID": "1"}, ], }, } output = _parse_response(data) assert output.duration_ms == expected_ms def test_duration_none_when_missing(self) -> None: data: dict[str, Any] = { "media": { "track": [ {"@type": "General", "Format": "Wave"}, {"@type": "Audio", "Format": "PCM", "CodecID": "1"}, ], }, } output = _parse_response(data) assert output.duration_ms is None def test_duration_none_when_unparseable(self) -> None: data: dict[str, Any] = { "media": { "track": [ {"@type": "General", "Format": "Wave", "Duration": "not-a-number"}, {"@type": "Audio", "Format": "PCM", "CodecID": "1"}, ], }, } output = _parse_response(data) assert output.duration_ms is None @pytest.mark.parametrize( ("is_truncated_value", "expected"), [ pytest.param("Yes", True, id="truncated"), pytest.param(None, False, id="not_truncated"), ], ) def test_truncation_flag(self, is_truncated_value: str | None, expected: bool) -> None: general_extra: dict[str, Any] = {} if is_truncated_value is not None: general_extra["IsTruncated"] = is_truncated_value data: dict[str, Any] = { "media": { "track": [ {"@type": "General", "Format": "Wave", "extra": general_extra}, {"@type": "Audio", "Format": "PCM", "CodecID": "1"}, ], }, } output = _parse_response(data) assert output.is_truncated is expected def test_finds_general_track_by_type_not_index(self) -> None: data: dict[str, Any] = { "media": { "track": [ { "@type": "Audio", "Format": "PCM", "CodecID": "1", "extra": {"IsTruncated": "Yes"}, }, {"@type": "General", "Format": "Wave", "extra": {}}, ], }, } output = _parse_response(data) assert output.container == "Wave" assert output.is_truncated is False def test_missing_fields_become_none(self) -> None: data: dict[str, Any] = { "media": { "track": [ {"@type": "General"}, {"@type": "Audio"}, ], }, } output = _parse_response(data) assert output.container is None assert output.codec is None assert output.codec_id is None assert output.channels is None assert output.sample_rate_hz is None assert output.bits_per_sample is None assert output.duration_ms is None assert output.adm_profile is None def test_empty_string_fields_become_none(self) -> None: data: dict[str, Any] = { "media": { "track": [ {"@type": "General", "Format": "", "Duration": ""}, { "@type": "Audio", "Format": "", "CodecID": "", "SamplingRate": "", "BitDepth": "", "Channels": "", }, ], }, } output = _parse_response(data) assert output.container is None assert output.codec is None assert output.codec_id is None assert output.channels is None assert output.sample_rate_hz is None assert output.bits_per_sample is None assert output.duration_ms is None @pytest.mark.parametrize( ("data", "match"), [ pytest.param( {"media": {"track": [{"@type": "Audio", "Format": "PCM"}]}}, "missing General track", id="no_general_track", ), pytest.param( {"media": {"track": [{"@type": "General", "Format": "Wave"}]}}, "missing Audio track", id="no_audio_track", ), ], ) def test_missing_track_raises(self, data: dict[str, Any], match: str) -> None: with pytest.raises(MediainfoOutputError, match=match): _parse_response(data) class TestCheckMediainfoFindings: @pytest.mark.parametrize( ("update", "expected_code"), [ pytest.param({"is_truncated": True}, AtmosValidationFindingCode.ATMOS_FILE_TRUNCATED, id="truncated"), pytest.param({"codec": "AAC"}, AtmosValidationFindingCode.NOT_PCM, id="wrong_codec"), pytest.param({"codec": None}, AtmosValidationFindingCode.NOT_PCM, id="codec_not_reported"), pytest.param({"codec_id": None}, AtmosValidationFindingCode.NOT_LPCM, id="codec_id_not_reported"), # WaveFormatExtensible with PCM subformat GUID — still PCM data, but not the # literal WAVE_FORMAT_PCM (0x0001) the Dolby ADM Profile spec requires. pytest.param( {"codec_id": "00000001-0000-0010-8000-00AA00389B71"}, AtmosValidationFindingCode.NOT_LPCM, id="codec_id_wave_format_extensible_guid", ), # 65534 = 0xFFFE = WAVE_FORMAT_EXTENSIBLE as an integer codec_id. pytest.param({"codec_id": "65534"}, AtmosValidationFindingCode.NOT_LPCM, id="codec_id_extensible_int"), pytest.param({"bits_per_sample": 16}, AtmosValidationFindingCode.WRONG_BIT_DEPTH, id="wrong_bit_depth"), pytest.param( {"bits_per_sample": None}, AtmosValidationFindingCode.WRONG_BIT_DEPTH, id="bit_depth_not_reported" ), pytest.param( {"sample_rate_hz": 44100}, AtmosValidationFindingCode.WRONG_SAMPLE_RATE, id="wrong_sample_rate" ), pytest.param( {"sample_rate_hz": None}, AtmosValidationFindingCode.WRONG_SAMPLE_RATE, id="sample_rate_not_reported" ), pytest.param({"container": "MPEG-4"}, AtmosValidationFindingCode.NOT_WAVE, id="not_wave"), pytest.param({"container": None}, AtmosValidationFindingCode.NOT_WAVE, id="container_not_reported"), pytest.param( {"adm_profile": "Something Else"}, AtmosValidationFindingCode.NOT_DOLBY_ATMOS, id="wrong_profile" ), pytest.param({"adm_profile": None}, AtmosValidationFindingCode.NOT_DOLBY_ATMOS, id="profile_not_reported"), ], ) def test_field_failure_emits_finding( self, update: dict[str, Any], expected_code: AtmosValidationFindingCode ) -> None: output = _valid_atmos_output().model_copy(update=update) with patch("src.atmos.mediainfo_check.run_mediainfo", return_value=output): atmos_validation_builder = AtmosValidationBuilder() check_mediainfo(_URL, atmos_validation_builder) assert expected_code.metadata_key in atmos_validation_builder.errors def test_valid_atmos_no_errors(self) -> None: with patch("src.atmos.mediainfo_check.run_mediainfo", return_value=_valid_atmos_output()): atmos_validation_builder = AtmosValidationBuilder() check_mediainfo(_URL, atmos_validation_builder) assert atmos_validation_builder.errors == {} assert 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=7805, )