import json from collections.abc import Iterator from concurrent.futures import Future from typing import Any, Self from unittest.mock import patch import pytest from pydantic import AnyUrl from src.atmos.mediainfo_check import AssetFetchError, MediainfoError from src.atmos.models import ( AtmosValidationBuilder, AtmosValidationFindingCode, AtmosValidationMetadata, MediainfoOutput, ) from src.atmos.validator import main, validate _ATMOS_URL = AnyUrl("file:///tmp/atmos") _STEREO_URL = AnyUrl("file:///tmp/stereo") _CHECK_MEDIAINFO = "src.atmos.validator.check_mediainfo" _RUN_MEDIAINFO = "src.atmos.validator.run_mediainfo" def _atmos_metadata(*, duration_ms: int | None = 180500) -> 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=duration_ms, ) def _stereo_output(*, duration_ms: int | None = 180500, is_truncated: bool = False) -> MediainfoOutput: return MediainfoOutput( container="FLAC", codec="FLAC", codec_id=None, channels=2, sample_rate_hz=44100, bits_per_sample=16, is_truncated=is_truncated, adm_profile=None, duration_ms=duration_ms, ) def _set_atmos_metadata(metadata: AtmosValidationMetadata | None) -> Any: def _side_effect(_url: AnyUrl, atmos_validation_builder: AtmosValidationBuilder) -> None: atmos_validation_builder.metadata = metadata return _side_effect class TestValidate: def test_populates_atmos_and_stereo_metadata(self) -> None: stereo = _stereo_output(duration_ms=180600) with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(_atmos_metadata())), patch(_RUN_MEDIAINFO, return_value=stereo), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) metadata = atmos_validation_builder.metadata assert metadata is not None assert metadata.duration_ms == 180500 assert metadata.stereo_reference_is_truncated is False assert metadata.stereo_reference_duration_ms == 180600 assert metadata.stereo_reference_duration_diff_ms == 100 def test_atmos_mediainfo_failure_leaves_metadata_unset(self) -> None: with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(None)), patch(_RUN_MEDIAINFO, return_value=_stereo_output()), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) assert atmos_validation_builder.metadata is None def test_stereo_mediainfo_failure_propagates(self) -> None: with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(_atmos_metadata())), patch(_RUN_MEDIAINFO, side_effect=MediainfoError("stereo broken")), pytest.raises(MediainfoError, match="stereo broken"), ): validate(_ATMOS_URL, _STEREO_URL) class TestDurationMatch: def test_matching_durations_no_finding(self) -> None: with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(_atmos_metadata(duration_ms=180500))), patch(_RUN_MEDIAINFO, return_value=_stereo_output(duration_ms=180500)), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) assert AtmosValidationFindingCode.DURATION_MISMATCH.metadata_key not in atmos_validation_builder.errors def test_exactly_at_threshold_passes(self) -> None: # Boundary: a difference of exactly the threshold is allowed (gate fires on strict >). with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(_atmos_metadata(duration_ms=180500))), patch(_RUN_MEDIAINFO, return_value=_stereo_output(duration_ms=178500)), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) assert AtmosValidationFindingCode.DURATION_MISMATCH.metadata_key not in atmos_validation_builder.errors def test_just_over_threshold_emits_finding(self) -> None: with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(_atmos_metadata(duration_ms=180500))), patch(_RUN_MEDIAINFO, return_value=_stereo_output(duration_ms=178499)), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) assert atmos_validation_builder.errors[AtmosValidationFindingCode.DURATION_MISMATCH.metadata_key] == ( "Atmos duration (180500 ms) and stereo reference (178499 ms) differ by 2001 ms, must not exceed 2000 ms" ) def test_atmos_without_metadata_skips_duration_check(self) -> None: with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(None)), patch(_RUN_MEDIAINFO, return_value=_stereo_output()), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) assert AtmosValidationFindingCode.DURATION_MISMATCH.metadata_key not in atmos_validation_builder.errors def test_atmos_without_duration_no_finding(self) -> None: with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(_atmos_metadata(duration_ms=None))), patch(_RUN_MEDIAINFO, return_value=_stereo_output()), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) assert AtmosValidationFindingCode.DURATION_MISMATCH.metadata_key not in atmos_validation_builder.errors def test_stereo_without_duration_no_finding(self) -> None: with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(_atmos_metadata())), patch(_RUN_MEDIAINFO, return_value=_stereo_output(duration_ms=None)), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) assert AtmosValidationFindingCode.DURATION_MISMATCH.metadata_key not in atmos_validation_builder.errors class TestStereoTruncation: def test_truncated_stereo_emits_finding(self) -> None: with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(_atmos_metadata())), patch(_RUN_MEDIAINFO, return_value=_stereo_output(is_truncated=True)), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) message = atmos_validation_builder.errors[AtmosValidationFindingCode.STEREO_FILE_TRUNCATED.metadata_key] assert "Stereo reference file is truncated" in message def test_truncated_stereo_populates_metadata_flag(self) -> None: with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(_atmos_metadata())), patch(_RUN_MEDIAINFO, return_value=_stereo_output(is_truncated=True, duration_ms=180400)), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) metadata = atmos_validation_builder.metadata assert metadata is not None assert metadata.stereo_reference_is_truncated is True assert metadata.stereo_reference_duration_ms == 180400 def test_truncated_stereo_skips_duration_check(self) -> None: # Stereo duration 100 ms off would fire DURATION_MISMATCH if it ran, but truncation # short-circuits the duration check. with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(_atmos_metadata(duration_ms=180500))), patch(_RUN_MEDIAINFO, return_value=_stereo_output(is_truncated=True, duration_ms=180400)), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) assert AtmosValidationFindingCode.DURATION_MISMATCH.metadata_key not in atmos_validation_builder.errors def test_truncated_atmos_skips_duration_check(self) -> None: truncated_atmos = _atmos_metadata(duration_ms=180500).model_copy(update={"is_truncated": True}) with ( patch(_CHECK_MEDIAINFO, side_effect=_set_atmos_metadata(truncated_atmos)), patch(_RUN_MEDIAINFO, return_value=_stereo_output(duration_ms=180400)), ): atmos_validation_builder = validate(_ATMOS_URL, _STEREO_URL) assert AtmosValidationFindingCode.DURATION_MISMATCH.metadata_key not in atmos_validation_builder.errors def _set_request(monkeypatch: pytest.MonkeyPatch, **overrides: Any) -> None: from src.worker.subprocess import WORKER_REQUEST body: dict[str, Any] = { "task_token": "tok", "asset_upload_type": "atmos", "atmos_bucket": "some-atmos-bucket", "atmos_key": "some/atmos.wav", "stereo_reference_bucket": "some-stereo-bucket", "stereo_reference_key": "some/stereo.flac", } | overrides monkeypatch.setenv(WORKER_REQUEST, json.dumps(body)) def _set_lookup_request(monkeypatch: pytest.MonkeyPatch, **overrides: Any) -> None: from src.worker.subprocess import WORKER_REQUEST body: dict[str, Any] = { "task_token": "tok", "asset_upload_type": "atmos", "atmos_bucket": "some-atmos-bucket", "atmos_key": "some/atmos.wav", "lookup_stereo_reference": True, } | overrides monkeypatch.setenv(WORKER_REQUEST, json.dumps(body)) _PROCESS_POOL = "src.atmos.validator.ProcessPoolExecutor" _RUN_RENDER_CHECKS = "src.atmos.validator.run_render_checks" class _InlineExecutor: # Stand-in for ProcessPoolExecutor that runs submitted callables in-process, so the # orchestrator's submit/result wiring can be exercised without spawning real workers. def __init__(self, *_args: Any, **_kwargs: Any) -> None: pass def __enter__(self) -> Self: return self def __exit__(self, *_exc: object) -> None: return None def submit(self, fn: Any, *args: Any, **kwargs: Any) -> Future[Any]: # Mirror ProcessPoolExecutor: a raised exception is captured onto the future and surfaces # at .result(), not at submit(). future: Future[Any] = Future() try: future.set_result(fn(*args, **kwargs)) except Exception as exc: future.set_exception(exc) return future class TestRunRenderChecks: def test_skips_render_when_blocking_errors_present(self) -> None: from src.atmos.validator import _run_render_checks atmos_validation_builder = AtmosValidationBuilder() atmos_validation_builder.error(AtmosValidationFindingCode.NOT_DOLBY_ATMOS, "not a master") with ( patch(_PROCESS_POOL) as process_pool, patch(_RUN_RENDER_CHECKS) as run_render_checks, ): _run_render_checks("bucket", "key", "stereo-bucket", "stereo/key.flac", atmos_validation_builder) process_pool.assert_not_called() run_render_checks.assert_not_called() def test_runs_render_in_spawn_pool_and_applies_all_measurements(self) -> None: from src.atmos.alignment_check import AlignmentMeasurement from src.atmos.lfe_check import LfeMeasurement from src.atmos.loudness_check import LoudnessMeasurement from src.atmos.render_checks import RenderCheckMeasurements from src.atmos.silent_object_check import SilentObjectMeasurement from src.atmos.validator import _MAX_RENDER_WORKERS, _run_render_checks atmos_validation_builder = AtmosValidationBuilder() atmos_validation_builder.metadata = _atmos_metadata() render_check_measurements = RenderCheckMeasurements( loudness=LoudnessMeasurement(integrated_loudness_lkfs=-9.0, true_peak_dbtp=0.5), alignment=AlignmentMeasurement(alignment_offset_ms=120.0, content_match_r=0.95), lfe=LfeMeasurement(above_400hz_dbfs=-42.0, above_2000hz_dbfs=-130.0), silent_object=SilentObjectMeasurement( object_track_count=3, silent_object_track_indices=(11,), height_track_count=2, silent_height_track_indices=(), ), ) with ( patch(_PROCESS_POOL) as process_pool, patch(_RUN_RENDER_CHECKS, return_value=render_check_measurements) as run_render_checks, ): inline_executor = _InlineExecutor() process_pool.return_value.__enter__.return_value = inline_executor with patch.object(inline_executor, "submit", wraps=inline_executor.submit) as submit: _run_render_checks( "some-bucket", "some/atmos.wav", "some-stereo-bucket", "some/stereo.flac", atmos_validation_builder, ) # A single render worker in a spawn-based pool (boto3 clients aren't fork-safe) produces # every measurement from one shared ADM parse — assert the actual call, not the module # constants. _, pool_kwargs = process_pool.call_args assert pool_kwargs["max_workers"] == _MAX_RENDER_WORKERS assert pool_kwargs["mp_context"].get_start_method() == "spawn" submit.assert_called_once() run_render_checks.assert_called_once_with( "some-bucket", "some/atmos.wav", "some-stereo-bucket", "some/stereo.flac" ) # apply_* are left unmocked, so every measurement flows through the real apply paths: stored # on the metadata and surfaced as warnings. assert atmos_validation_builder.metadata is not None assert atmos_validation_builder.metadata.integrated_loudness_lkfs == -9.0 assert atmos_validation_builder.metadata.stereo_reference_alignment_offset_ms == 120.0 assert atmos_validation_builder.metadata.lfe_above_400hz_dbfs == -42.0 assert atmos_validation_builder.metadata.silent_object_track_indices == (11,) assert AtmosValidationFindingCode.LOUDNESS_TOO_HIGH.metadata_key in atmos_validation_builder.warnings assert AtmosValidationFindingCode.TRUE_PEAK_TOO_HIGH.metadata_key in atmos_validation_builder.warnings assert AtmosValidationFindingCode.SYNC_MISMATCH.metadata_key in atmos_validation_builder.warnings assert ( AtmosValidationFindingCode.LFE_LEVEL_ABOVE_400HZ_TOO_HIGH.metadata_key in atmos_validation_builder.warnings ) assert AtmosValidationFindingCode.SILENT_OBJECT.metadata_key in atmos_validation_builder.warnings def test_render_failure_propagates(self) -> None: # A failed render fails the delivery: it surfaces from .result() and propagates out, rather # than degrading to a silent advisory-skipped pass. from src.atmos.render_errors import FfmpegError from src.atmos.validator import _run_render_checks atmos_validation_builder = AtmosValidationBuilder() atmos_validation_builder.metadata = _atmos_metadata() with ( patch(_PROCESS_POOL, _InlineExecutor), patch(_RUN_RENDER_CHECKS, side_effect=FfmpegError("boom")), pytest.raises(FfmpegError, match="boom"), ): _run_render_checks("bucket", "key", "stereo-bucket", "stereo/key.flac", atmos_validation_builder) assert atmos_validation_builder.warnings == {} assert atmos_validation_builder.metadata is not None assert atmos_validation_builder.metadata.integrated_loudness_lkfs is None def test_terminated_render_worker_becomes_render_worker_error(self) -> None: # A worker killed mid-render (e.g. OOM) raises BrokenProcessPool from .result(); translate # it to a named domain error so it re-hydrates legibly across the process boundary instead # of as an opaque UnknownSubprocessError. from concurrent.futures.process import BrokenProcessPool from src.atmos.render_errors import RenderWorkerError from src.atmos.validator import _run_render_checks atmos_validation_builder = AtmosValidationBuilder() atmos_validation_builder.metadata = _atmos_metadata() with ( patch(_PROCESS_POOL, _InlineExecutor), patch(_RUN_RENDER_CHECKS, side_effect=BrokenProcessPool("worker died")), pytest.raises(RenderWorkerError, match="terminated abruptly"), ): _run_render_checks("bucket", "key", "stereo-bucket", "stereo/key.flac", atmos_validation_builder) assert atmos_validation_builder.warnings == {} class TestMain: @pytest.fixture(autouse=True) def _stub_render_checks(self) -> Iterator[None]: # TestMain covers the worker plumbing; the render-based checks have their own tests # (test_loudness_check.py, TestRunRenderChecks). Stub the orchestrator so main() # doesn't spawn a render pool. with patch("src.atmos.validator._run_render_checks"): yield def test_success_emits_full_result_payload( self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: _set_request(monkeypatch) with ( patch("src.atmos.validator.presigned_url_ttl_seconds"), patch("src.atmos.validator.s3.create_presigned_url", return_value=AnyUrl("file:///tmp/x")), patch("src.atmos.validator.validate", return_value=AtmosValidationBuilder()), ): main() assert json.loads(capsys.readouterr().out) == { "is_valid": True, "metadata": None, "errors": {}, "warnings": {}, } def test_success_with_errors_emits_is_valid_false( self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: _set_request(monkeypatch) atmos_validation_builder = AtmosValidationBuilder() atmos_validation_builder.error(AtmosValidationFindingCode.NOT_PCM, "not pcm") with ( patch("src.atmos.validator.presigned_url_ttl_seconds"), patch("src.atmos.validator.s3.create_presigned_url", return_value=AnyUrl("file:///tmp/x")), patch("src.atmos.validator.validate", return_value=atmos_validation_builder), ): main() out = json.loads(capsys.readouterr().out) assert out["is_valid"] is False assert out["errors"] == {"codec": "not pcm"} assert out["warnings"] == {} def test_failure_writes_error_output_and_exits_one( self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: _set_request(monkeypatch) with ( patch("src.atmos.validator.presigned_url_ttl_seconds"), patch("src.atmos.validator.s3.create_presigned_url", return_value=AnyUrl("file:///tmp/x")), patch("src.atmos.validator.validate", side_effect=AssetFetchError("S3 503")), pytest.raises(SystemExit) as exc_info, ): main() assert exc_info.value.code == 1 assert json.loads(capsys.readouterr().out) == {"error": "AssetFetchError", "message": "S3 503"} def test_explicit_stereo_reference_skips_ows_lookup( self, monkeypatch: pytest.MonkeyPatch, ) -> None: _set_request(monkeypatch) with ( patch("src.atmos.validator.presigned_url_ttl_seconds"), patch("src.atmos.validator.s3.create_presigned_url", return_value=AnyUrl("file:///tmp/x")), patch("src.atmos.validator.validate", return_value=AtmosValidationBuilder()), patch("src.atmos.validator.ows_assets.get_stereo_for_asset") as get_stereo_for_asset, patch("src.atmos.validator.ows_assets.post_validation_result") as post_validation_result, ): main() get_stereo_for_asset.assert_not_called() post_validation_result.assert_not_called() def test_lookup_stereo_reference_calls_ows_with_atmos_key( self, monkeypatch: pytest.MonkeyPatch, ) -> None: _set_lookup_request(monkeypatch, atmos_key="bucket-prefix/atmos.wav") s3_call_args: list[tuple[str, str]] = [] def _record_presigned(bucket: str, key: str, *, expires_in_seconds: int) -> AnyUrl: s3_call_args.append((bucket, key)) return AnyUrl(f"file:///tmp/{bucket}/{key}") with ( patch("src.atmos.validator.presigned_url_ttl_seconds"), patch("src.atmos.validator.s3.create_presigned_url", side_effect=_record_presigned), patch("src.atmos.validator.validate", return_value=AtmosValidationBuilder()), patch( "src.atmos.validator.ows_assets.get_stereo_for_asset", return_value=("looked-up-bucket", "looked-up-key"), ) as get_stereo_for_asset, ): main() get_stereo_for_asset.assert_called_once_with("bucket-prefix/atmos.wav") assert ("looked-up-bucket", "looked-up-key") in s3_call_args def test_report_validation_result_calls_ows_with_full_result( self, monkeypatch: pytest.MonkeyPatch, ) -> None: _set_request(monkeypatch, report_validation_result=True, atmos_key="bucket-prefix/atmos.wav") atmos_validation_builder = AtmosValidationBuilder() atmos_validation_builder.error(AtmosValidationFindingCode.NOT_PCM, "not pcm") with ( patch("src.atmos.validator.presigned_url_ttl_seconds"), patch("src.atmos.validator.s3.create_presigned_url", return_value=AnyUrl("file:///tmp/x")), patch("src.atmos.validator.validate", return_value=atmos_validation_builder), patch("src.atmos.validator.ows_assets.post_validation_result") as post_validation_result, ): main() post_validation_result.assert_called_once() asset_key_arg, result_arg = post_validation_result.call_args.args assert asset_key_arg == "bucket-prefix/atmos.wav" assert result_arg.is_valid is False assert result_arg.errors == {"codec": "not pcm"} def test_report_validation_result_off_by_default( self, monkeypatch: pytest.MonkeyPatch, ) -> None: _set_request(monkeypatch) with ( patch("src.atmos.validator.presigned_url_ttl_seconds"), patch("src.atmos.validator.s3.create_presigned_url", return_value=AnyUrl("file:///tmp/x")), patch("src.atmos.validator.validate", return_value=AtmosValidationBuilder()), patch("src.atmos.validator.ows_assets.post_validation_result") as post_validation_result, ): main() post_validation_result.assert_not_called() def test_both_flags_lookup_and_report( self, monkeypatch: pytest.MonkeyPatch, ) -> None: _set_lookup_request(monkeypatch, atmos_key="bucket-prefix/atmos.wav", report_validation_result=True) with ( patch("src.atmos.validator.presigned_url_ttl_seconds"), patch("src.atmos.validator.s3.create_presigned_url", return_value=AnyUrl("file:///tmp/x")), patch("src.atmos.validator.validate", return_value=AtmosValidationBuilder()), patch( "src.atmos.validator.ows_assets.get_stereo_for_asset", return_value=("looked-up-bucket", "looked-up-key"), ) as get_stereo_for_asset, patch("src.atmos.validator.ows_assets.post_validation_result") as post_validation_result, ): main() get_stereo_for_asset.assert_called_once_with("bucket-prefix/atmos.wav") post_validation_result.assert_called_once() asset_key_arg, result_arg = post_validation_result.call_args.args assert asset_key_arg == "bucket-prefix/atmos.wav" assert result_arg.is_valid is True def test_lookup_failure_writes_ows_assets_error_envelope( self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: from src.clients.ows_assets import OwsAssetsError _set_lookup_request(monkeypatch) with ( patch("src.atmos.validator.presigned_url_ttl_seconds"), patch( "src.atmos.validator.ows_assets.get_stereo_for_asset", side_effect=OwsAssetsError("ows-assets returned 503 for stereo lookup"), ), pytest.raises(SystemExit) as exc_info, ): main() assert exc_info.value.code == 1 out = json.loads(capsys.readouterr().out) assert out["error"] == "OwsAssetsError" assert "503" in out["message"] def test_report_failure_writes_ows_assets_error_envelope( self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: from src.clients.ows_assets import OwsAssetsError _set_request(monkeypatch, report_validation_result=True) with ( patch("src.atmos.validator.presigned_url_ttl_seconds"), patch("src.atmos.validator.s3.create_presigned_url", return_value=AnyUrl("file:///tmp/x")), patch("src.atmos.validator.validate", return_value=AtmosValidationBuilder()), patch( "src.atmos.validator.ows_assets.post_validation_result", side_effect=OwsAssetsError("ows-assets returned 500 for /v2/asset/status"), ), pytest.raises(SystemExit) as exc_info, ): main() assert exc_info.value.code == 1 out = json.loads(capsys.readouterr().out) assert out["error"] == "OwsAssetsError" assert "500" in out["message"]