diff --git a/lambda/src/data_hub_lambda/dishcam/process_file.py b/lambda/src/data_hub_lambda/dishcam/process_file.py index ac2f6c9..6ef6dac 100644 --- a/lambda/src/data_hub_lambda/dishcam/process_file.py +++ b/lambda/src/data_hub_lambda/dishcam/process_file.py @@ -8,6 +8,7 @@ from data_hub_lambda.dishcam.encode_video import encode_tiff_stack from data_hub_lambda.dishcam.filenames import RUN_JSON_NAME, is_tiff, matches_filename from data_hub_lambda.dishcam.parse_metadata import encode_fps, parse_run_json, playback_fps +from data_hub_lambda.models import FileResponse from data_hub_shared import s3_utils from data_hub_shared.config import config @@ -27,13 +28,26 @@ def process_file(instrument_id: str, run_id: str, filename: str) -> None: sidecar is already in S3. Reprocess already marks the trigger `processing`, so a missing sibling - fails that file instead of leaving it stuck. + fails that file instead of leaving it stuck. A parsed sidecar is + completed even if a stack failed: it has no stack of its own, and + leaving it in `processing` stranded the run. Stack status, not the + sidecar, decides whether the run looks failed. TIFF and `run.json` are separate S3 events, so two invocations can - encode the same stack. `create_file` is idempotent; completed/failed - updates swallow 409 so the loser does not fail a successful run. - Duplicate compute is accepted — a lock would need run-level state - we do not have. + encode the same stack. The `run.json` batch skips stacks that are + already completed or processing: `completed → processing` is a legal + transition, and a later disk-full failure would otherwise reopen a + sibling's success and mark it failed. A stack left in `processing` + after a timeout is retried by reprocessing that TIFF, not another + `run.json` batch. A TIFF-triggered invoke always encodes that stack. + completed/failed updates swallow 409 so the loser does not fail a + successful run. + + Run metadata is written from the parsed sidecar even when every stack + is skipped, so a corrected `run.json` still updates the run. + + High-quality stacks are a few GB and Lambda `/tmp` is capped, so each + encode deletes its local TIFF/MP4/JPEG before the next stack. """ if not matches_filename(filename): logger.info("Ignoring DishCam file %s; not a TIFF or run.json.", filename) @@ -82,11 +96,19 @@ def process_file(instrument_id: str, run_id: str, filename: str) -> None: s3_key=tiff_key, filename=tiff_filename, ) - _update_file_status(client, record.id, "failed", error_message=str(exc)) + _fail_file(client, record, str(exc)) + _fail_file(client, _sidecar_file(client, instrument_id, run_id), str(exc)) raise + sidecar = _sidecar_file(client, instrument_id, run_id) + # Only bump uploaded/failed → processing. A completed sidecar from a + # sibling invocation must stay completed so the run does not flicker + # back to processing during a duplicate encode. + if sidecar.status in {"uploaded", "failed"}: + _update_file_status(client, sidecar.id, "processing") + last_error: Exception | None = None - encoded_any = False + owned = is_tiff(filename) for tiff_filename in tiff_filenames: try: _encode_tiff( @@ -97,14 +119,22 @@ def process_file(instrument_id: str, run_id: str, filename: str) -> None: raw_dir, tiff_filename, fps, + owned=owned, ) - encoded_any = True except Exception as exc: logger.error("Error processing DishCam file %s: %s", tiff_filename, exc) last_error = exc - if encoded_any: - client.update_run(instrument_id, run_id, metadata=metadata) + client.update_run(instrument_id, run_id, metadata=metadata) + # The sidecar parsed; complete it even if a stack failed. Do not let a + # status PATCH hide the encode error the caller should see. + if sidecar.status != "completed": + try: + _update_file_status(client, sidecar.id, "completed") + except Exception: + logger.exception("Failed to complete DishCam run.json for %s.", run_id) + if last_error is None: + raise if last_error is not None: raise last_error @@ -138,7 +168,17 @@ def _encode_tiff( raw_dir: Path, tiff_filename: str, fps: float, -) -> None: + *, + owned: bool, +) -> bool: + """Encode one stack. Return True if this invoke produced an MP4. + + `owned` is True when the S3/reprocess trigger is this TIFF, so a + duplicate event or an intentional retry still runs. The `run.json` + batch passes False and leaves in-flight and finished stacks alone. A + stack stuck in `processing` after a timeout is retried by + reprocessing that TIFF, not another `run.json` batch. + """ tiff_key = f"{instrument_id}/{run_id}/{tiff_filename}" tiff_uri = f"s3://{raw_bucket}/{tiff_key}" tiff_record = client.create_file( @@ -149,15 +189,22 @@ def _encode_tiff( filename=tiff_filename, ) tiff_id = tiff_record.id + local_tiff = raw_dir / tiff_filename + mp4_path = raw_dir / f"{Path(tiff_filename).stem}.mp4" + poster_path = raw_dir / f"{Path(tiff_filename).stem}.jpg" + + if not owned and tiff_record.status in {"completed", "processing"}: + logger.info( + "Skipping DishCam file %s; already %s.", + tiff_filename, + tiff_record.status, + ) + return False try: client.update_file(tiff_id, status="processing") - local_tiff = raw_dir / tiff_filename s3_utils.download_file(tiff_uri, local_tiff) - - mp4_path = raw_dir / f"{Path(tiff_filename).stem}.mp4" - poster_path = raw_dir / f"{Path(tiff_filename).stem}.jpg" encode_tiff_stack(local_tiff, mp4_path, poster_path, fps) processed_bucket = config.AWS_S3_PROCESSED_DATA_BUCKET or "" @@ -183,11 +230,40 @@ def _encode_tiff( "DishCam file %s already finished by a sibling invocation.", tiff_filename, ) - return + return True logger.info("DishCam file %s marked as completed.", tiff_filename) + return True except Exception as exc: _update_file_status(client, tiff_id, "failed", error_message=str(exc)) raise + finally: + # One high-quality stack can be several GB; leaving it on disk + # fills the Lambda `/tmp` cap before the next stack in the batch. + _remove_local(local_tiff, mp4_path, poster_path) + + +def _remove_local(*paths: Path) -> None: + for path in paths: + path.unlink(missing_ok=True) + + +def _sidecar_file(client: DataHubClient, instrument_id: str, run_id: str) -> FileResponse: + """Return the `run.json` row. The run already exists via `ensure_run`.""" + raw_bucket = config.AWS_S3_RAW_DATA_BUCKET or "" + return client.create_file( + instrument_id=instrument_id, + run_id=run_id, + s3_bucket=raw_bucket, + s3_key=f"{instrument_id}/{run_id}/{RUN_JSON_NAME}", + filename=RUN_JSON_NAME, + ) + + +def _fail_file(client: DataHubClient, record: FileResponse, error_message: str) -> None: + """Mark failed. Terminal and `uploaded` states must go through `processing` first.""" + if record.status != "processing": + _update_file_status(client, record.id, "processing") + _update_file_status(client, record.id, "failed", error_message=error_message) def _upload_processed( diff --git a/lambda/tests/dishcam/test_process_file.py b/lambda/tests/dishcam/test_process_file.py index 50ca128..ed6037a 100644 --- a/lambda/tests/dishcam/test_process_file.py +++ b/lambda/tests/dishcam/test_process_file.py @@ -1,6 +1,8 @@ """Unit tests for DishCam `process_file` orchestration.""" from __future__ import annotations +from collections.abc import Iterator +from contextlib import contextmanager from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch @@ -51,6 +53,9 @@ def _run_response() -> RunResponse: ) +SIDECAR_ID = 99 + + def _exists_for(*keys: str): present = set(keys) @@ -61,6 +66,85 @@ def _exists(s3_uri: str, **_: Any) -> bool: return _exists +def _completed_file_ids(client: MagicMock) -> list[int]: + return [ + call.args[0] + for call in client.update_file.call_args_list + if call.kwargs.get("status") == "completed" + ] + + +def _write_download(s3_uri: str, local_path: Path, **_: Any) -> None: + local_path.parent.mkdir(parents=True, exist_ok=True) + if s3_uri.endswith("run.json"): + local_path.write_text('{"fps": 1.0}') + else: + local_path.write_bytes(b"tiff") + + +def _write_encode(tiff_path: Path, mp4_path: Path, poster_path: Path, fps: float) -> None: + mp4_path.write_bytes(b"mp4") + poster_path.write_bytes(b"jpg") + + +def _status_updates(client: MagicMock, file_id: int) -> list[str]: + return [ + call.kwargs.get("status") + for call in client.update_file.call_args_list + if call.args[0] == file_id and call.kwargs.get("status") is not None + ] + + +@contextmanager +def _patched_process( + tmp_path: Path, + client: MagicMock, + *, + list_objects: list[str] | None = None, + encode: Any = None, + download: Any = _write_download, +) -> Iterator[Any]: + encode_mock = encode if encode is not None else MagicMock(side_effect=_write_encode) + with ( + patch( + "data_hub_lambda.dishcam.process_file.get_client", + return_value=client, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.object_exists", + return_value=True, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.list_objects", + return_value=list_objects or [], + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.download_file", + side_effect=download, + ), + patch("data_hub_lambda.dishcam.process_file.s3_utils.upload_file"), + patch( + "data_hub_lambda.dishcam.process_file.encode_tiff_stack", + encode_mock, + ), + patch( + "data_hub_shared.config.config.AWS_S3_RAW_DATA_BUCKET", + "raw", + ), + patch( + "data_hub_shared.config.config.AWS_S3_PROCESSED_DATA_BUCKET", + "processed", + ), + patch( + "data_hub_shared.config.config.LOCAL_RAW_DATA_DIRPATH", + tmp_path, + ), + ): + from data_hub_lambda.dishcam.process_file import process_file + + yield process_file, encode_mock + + class TestProcessFileSkipUntilBothPresent: def test_tiff_without_json_does_not_ensure_run_or_download(self) -> None: client = MagicMock() @@ -161,6 +245,7 @@ def test_run_json_trigger_encodes_sibling_tiff(self, tmp_path: Path) -> None: client = MagicMock() client.ensure_run.return_value = _run_response() client.create_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json"), _file_response(10, "stack.tif"), _file_response(11, "stack.mp4", category="processed"), _file_response(12, "stack.jpg", category="processed"), @@ -233,27 +318,92 @@ def _encode(tiff_path: Path, mp4_path: Path, poster_path: Path, fps: float) -> N metadata = client.update_run.call_args.kwargs["metadata"] assert metadata["measured_fps"] == 0.9 assert metadata["frames"] == 4 - completed = [ - call + assert _completed_file_ids(client) == [10, SIDECAR_ID] + + def test_processing_sidecar_is_completed_after_encode(self, tmp_path: Path) -> None: + client = MagicMock() + client.ensure_run.return_value = _run_response() + client.create_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json", status="processing"), + _file_response(10, "stack.tif"), + _file_response(11, "stack.mp4", category="processed"), + _file_response(12, "stack.jpg", category="processed"), + ] + + def _download(s3_uri: str, local_path: Path, **_: Any) -> None: + local_path.parent.mkdir(parents=True, exist_ok=True) + if s3_uri.endswith("run.json"): + local_path.write_text('{"fps": 1.0}') + else: + local_path.write_bytes(b"tiff") + + def _encode(tiff_path: Path, mp4_path: Path, poster_path: Path, fps: float) -> None: + mp4_path.write_bytes(b"mp4") + poster_path.write_bytes(b"jpg") + + with ( + patch( + "data_hub_lambda.dishcam.process_file.get_client", + return_value=client, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.object_exists", + return_value=True, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.list_objects", + return_value=["s3://raw/dishcam/run-xyz/stack.tif"], + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.download_file", + side_effect=_download, + ), + patch("data_hub_lambda.dishcam.process_file.s3_utils.upload_file"), + patch( + "data_hub_lambda.dishcam.process_file.encode_tiff_stack", + side_effect=_encode, + ), + patch( + "data_hub_shared.config.config.AWS_S3_RAW_DATA_BUCKET", + "raw", + ), + patch( + "data_hub_shared.config.config.AWS_S3_PROCESSED_DATA_BUCKET", + "processed", + ), + patch( + "data_hub_shared.config.config.LOCAL_RAW_DATA_DIRPATH", + tmp_path, + ), + ): + from data_hub_lambda.dishcam.process_file import process_file + + process_file("dishcam", "run-xyz", "run.json") + + sidecar_statuses = [ + call.kwargs.get("status") for call in client.update_file.call_args_list - if call.kwargs.get("status") == "completed" + if call.args[0] == SIDECAR_ID and call.kwargs.get("status") is not None ] - assert completed - assert completed[-1].args[0] == 10 + assert sidecar_statuses == ["completed"] + assert _completed_file_ids(client) == [10, SIDECAR_ID] def test_completed_conflict_is_not_a_failure(self, tmp_path: Path) -> None: client = MagicMock() client.ensure_run.return_value = _run_response() client.create_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json"), _file_response(10, "stack.tif"), _file_response(11, "stack.mp4", category="processed"), _file_response(12, "stack.jpg", category="processed"), ] client.update_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json", status="processing"), _file_response(10, "stack.tif", status="processing"), _file_response(11, "stack.mp4", category="processed"), _file_response(12, "stack.jpg", category="processed"), ApiError("conflict", status_code=409), + _file_response(SIDECAR_ID, "run.json", status="completed"), ] def _download(s3_uri: str, local_path: Path, **_: Any) -> None: @@ -317,6 +467,7 @@ def test_run_json_trigger_encodes_every_tiff(self, tmp_path: Path) -> None: client = MagicMock() client.ensure_run.return_value = _run_response() client.create_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json"), _file_response(10, "empty.tif"), _file_response(11, "empty.mp4", category="processed"), _file_response(12, "empty.jpg", category="processed"), @@ -383,17 +534,13 @@ def _encode(tiff_path: Path, mp4_path: Path, poster_path: Path, fps: float) -> N encoded = [call.args[0].name for call in encode.call_args_list] assert encoded == ["empty.tif", "ruler.tif"] - completed = [ - call.args[0] - for call in client.update_file.call_args_list - if call.kwargs.get("status") == "completed" - ] - assert completed == [10, 20] + assert _completed_file_ids(client) == [10, 20, SIDECAR_ID] def test_tiff_trigger_encodes_only_that_stack(self, tmp_path: Path) -> None: client = MagicMock() client.ensure_run.return_value = _run_response() client.create_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json"), _file_response(20, "ruler.tif"), _file_response(21, "ruler.mp4", category="processed"), _file_response(22, "ruler.jpg", category="processed"), @@ -459,11 +606,13 @@ def _encode(tiff_path: Path, mp4_path: Path, poster_path: Path, fps: float) -> N list_objects.assert_not_called() assert [call.args[0].name for call in encode.call_args_list] == ["ruler.tif"] + assert _completed_file_ids(client) == [20, SIDECAR_ID] def test_one_failed_stack_does_not_block_the_others(self, tmp_path: Path) -> None: client = MagicMock() client.ensure_run.return_value = _run_response() client.create_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json"), _file_response(10, "empty.tif"), _file_response(20, "ruler.tif"), _file_response(21, "ruler.mp4", category="processed"), @@ -533,4 +682,400 @@ def _encode(tiff_path: Path, mp4_path: Path, poster_path: Path, fps: float) -> N } assert statuses[10] == "failed" assert statuses[20] == "completed" + assert statuses[SIDECAR_ID] == "completed" client.update_run.assert_called_once() + + def test_run_json_skips_completed_and_in_flight_stacks(self, tmp_path: Path) -> None: + client = MagicMock() + client.ensure_run.return_value = _run_response() + records = { + "run.json": _file_response(SIDECAR_ID, "run.json"), + "empty.tif": _file_response(10, "empty.tif", status="completed"), + "ruler.tif": _file_response(20, "ruler.tif", status="processing"), + "gk.tif": _file_response(30, "gk.tif"), + "gk.mp4": _file_response(31, "gk.mp4", category="processed"), + "gk.jpg": _file_response(32, "gk.jpg", category="processed"), + } + client.create_file.side_effect = lambda **kwargs: records[kwargs["filename"]] + + encode = MagicMock(side_effect=_write_encode) + + with ( + patch( + "data_hub_lambda.dishcam.process_file.get_client", + return_value=client, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.object_exists", + return_value=True, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.list_objects", + return_value=[ + "s3://raw/dishcam/run-xyz/empty.tif", + "s3://raw/dishcam/run-xyz/gk.tif", + "s3://raw/dishcam/run-xyz/ruler.tif", + ], + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.download_file", + side_effect=_write_download, + ), + patch("data_hub_lambda.dishcam.process_file.s3_utils.upload_file"), + patch( + "data_hub_lambda.dishcam.process_file.encode_tiff_stack", + encode, + ), + patch( + "data_hub_shared.config.config.AWS_S3_RAW_DATA_BUCKET", + "raw", + ), + patch( + "data_hub_shared.config.config.AWS_S3_PROCESSED_DATA_BUCKET", + "processed", + ), + patch( + "data_hub_shared.config.config.LOCAL_RAW_DATA_DIRPATH", + tmp_path, + ), + ): + from data_hub_lambda.dishcam.process_file import process_file + + process_file("dishcam", "run-xyz", "run.json") + + assert [call.args[0].name for call in encode.call_args_list] == ["gk.tif"] + assert _status_updates(client, 10) == [] + assert _status_updates(client, 20) == [] + assert _completed_file_ids(client) == [30, SIDECAR_ID] + + def test_run_json_completes_sidecar_when_every_stack_is_already_done( + self, tmp_path: Path + ) -> None: + client = MagicMock() + client.ensure_run.return_value = _run_response() + records = { + "run.json": _file_response(SIDECAR_ID, "run.json", status="processing"), + "empty.tif": _file_response(10, "empty.tif", status="completed"), + } + client.create_file.side_effect = lambda **kwargs: records[kwargs["filename"]] + + encode = MagicMock(side_effect=_write_encode) + + with ( + patch( + "data_hub_lambda.dishcam.process_file.get_client", + return_value=client, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.object_exists", + return_value=True, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.list_objects", + return_value=["s3://raw/dishcam/run-xyz/empty.tif"], + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.download_file", + side_effect=_write_download, + ), + patch("data_hub_lambda.dishcam.process_file.s3_utils.upload_file"), + patch( + "data_hub_lambda.dishcam.process_file.encode_tiff_stack", + encode, + ), + patch( + "data_hub_shared.config.config.AWS_S3_RAW_DATA_BUCKET", + "raw", + ), + patch( + "data_hub_shared.config.config.AWS_S3_PROCESSED_DATA_BUCKET", + "processed", + ), + patch( + "data_hub_shared.config.config.LOCAL_RAW_DATA_DIRPATH", + tmp_path, + ), + ): + from data_hub_lambda.dishcam.process_file import process_file + + process_file("dishcam", "run-xyz", "run.json") + + encode.assert_not_called() + client.update_run.assert_called_once() + assert client.update_run.call_args.kwargs["metadata"] == {"fps": 1.0} + assert _completed_file_ids(client) == [SIDECAR_ID] + + def test_tiff_trigger_encodes_even_if_already_completed(self, tmp_path: Path) -> None: + client = MagicMock() + client.ensure_run.return_value = _run_response() + client.create_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json", status="completed"), + _file_response(20, "ruler.tif", status="completed"), + _file_response(21, "ruler.mp4", category="processed"), + _file_response(22, "ruler.jpg", category="processed"), + ] + + encode = MagicMock(side_effect=_write_encode) + + with ( + patch( + "data_hub_lambda.dishcam.process_file.get_client", + return_value=client, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.object_exists", + return_value=True, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.download_file", + side_effect=_write_download, + ), + patch("data_hub_lambda.dishcam.process_file.s3_utils.upload_file"), + patch( + "data_hub_lambda.dishcam.process_file.encode_tiff_stack", + encode, + ), + patch( + "data_hub_shared.config.config.AWS_S3_RAW_DATA_BUCKET", + "raw", + ), + patch( + "data_hub_shared.config.config.AWS_S3_PROCESSED_DATA_BUCKET", + "processed", + ), + patch( + "data_hub_shared.config.config.LOCAL_RAW_DATA_DIRPATH", + tmp_path, + ), + ): + from data_hub_lambda.dishcam.process_file import process_file + + process_file("dishcam", "run-xyz", "ruler.tif") + + assert [call.args[0].name for call in encode.call_args_list] == ["ruler.tif"] + assert "processing" in _status_updates(client, 20) + assert 20 in _completed_file_ids(client) + + def test_local_stack_files_are_removed_after_each_encode(self, tmp_path: Path) -> None: + client = MagicMock() + client.ensure_run.return_value = _run_response() + client.create_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json"), + _file_response(10, "empty.tif"), + _file_response(11, "empty.mp4", category="processed"), + _file_response(12, "empty.jpg", category="processed"), + _file_response(20, "ruler.tif"), + _file_response(21, "ruler.mp4", category="processed"), + _file_response(22, "ruler.jpg", category="processed"), + ] + + tiffs_on_disk: list[list[str]] = [] + + def _encode(tiff_path: Path, mp4_path: Path, poster_path: Path, fps: float) -> None: + tiffs_on_disk.append(sorted(path.name for path in tiff_path.parent.glob("*.tif"))) + _write_encode(tiff_path, mp4_path, poster_path, fps) + + with ( + patch( + "data_hub_lambda.dishcam.process_file.get_client", + return_value=client, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.object_exists", + return_value=True, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.list_objects", + return_value=[ + "s3://raw/dishcam/run-xyz/empty.tif", + "s3://raw/dishcam/run-xyz/ruler.tif", + ], + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.download_file", + side_effect=_write_download, + ), + patch("data_hub_lambda.dishcam.process_file.s3_utils.upload_file"), + patch( + "data_hub_lambda.dishcam.process_file.encode_tiff_stack", + side_effect=_encode, + ), + patch( + "data_hub_shared.config.config.AWS_S3_RAW_DATA_BUCKET", + "raw", + ), + patch( + "data_hub_shared.config.config.AWS_S3_PROCESSED_DATA_BUCKET", + "processed", + ), + patch( + "data_hub_shared.config.config.LOCAL_RAW_DATA_DIRPATH", + tmp_path, + ), + ): + from data_hub_lambda.dishcam.process_file import process_file + + process_file("dishcam", "run-xyz", "run.json") + + assert tiffs_on_disk == [["empty.tif"], ["ruler.tif"]] + leftover = list((tmp_path / "dishcam" / "run-xyz").glob("*.tif")) + leftover += list((tmp_path / "dishcam" / "run-xyz").glob("*.mp4")) + leftover += list((tmp_path / "dishcam" / "run-xyz").glob("*.jpg")) + assert leftover == [] + + def test_failed_encode_still_removes_local_files(self, tmp_path: Path) -> None: + client = MagicMock() + client.ensure_run.return_value = _run_response() + client.create_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json"), + _file_response(10, "empty.tif"), + _file_response(20, "ruler.tif"), + _file_response(21, "ruler.mp4", category="processed"), + _file_response(22, "ruler.jpg", category="processed"), + ] + + tiffs_on_disk: list[list[str]] = [] + + def _encode(tiff_path: Path, mp4_path: Path, poster_path: Path, fps: float) -> None: + tiffs_on_disk.append(sorted(path.name for path in tiff_path.parent.glob("*.tif"))) + if tiff_path.name == "empty.tif": + raise RuntimeError("empty stack is corrupt") + _write_encode(tiff_path, mp4_path, poster_path, fps) + + with ( + patch( + "data_hub_lambda.dishcam.process_file.get_client", + return_value=client, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.object_exists", + return_value=True, + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.list_objects", + return_value=[ + "s3://raw/dishcam/run-xyz/empty.tif", + "s3://raw/dishcam/run-xyz/ruler.tif", + ], + ), + patch( + "data_hub_lambda.dishcam.process_file.s3_utils.download_file", + side_effect=_write_download, + ), + patch("data_hub_lambda.dishcam.process_file.s3_utils.upload_file"), + patch( + "data_hub_lambda.dishcam.process_file.encode_tiff_stack", + side_effect=_encode, + ), + patch( + "data_hub_shared.config.config.AWS_S3_RAW_DATA_BUCKET", + "raw", + ), + patch( + "data_hub_shared.config.config.AWS_S3_PROCESSED_DATA_BUCKET", + "processed", + ), + patch( + "data_hub_shared.config.config.LOCAL_RAW_DATA_DIRPATH", + tmp_path, + ), + ): + from data_hub_lambda.dishcam.process_file import process_file + + with pytest.raises(RuntimeError, match="corrupt"): + process_file("dishcam", "run-xyz", "run.json") + + assert tiffs_on_disk == [["empty.tif"], ["ruler.tif"]] + + @pytest.mark.parametrize( + ("tiff_status", "sidecar_status", "tiff_expected", "sidecar_expected"), + [ + ("uploaded", "uploaded", ["processing", "failed"], ["processing", "failed"]), + ("processing", "processing", ["failed"], ["failed"]), + ], + ) + def test_unreadable_run_json_fails_sidecar_and_stacks( + self, + tmp_path: Path, + tiff_status: str, + sidecar_status: str, + tiff_expected: list[str], + sidecar_expected: list[str], + ) -> None: + client = MagicMock() + client.ensure_run.return_value = _run_response() + client.create_file.side_effect = [ + _file_response(10, "stack.tif", status=tiff_status), + _file_response(SIDECAR_ID, "run.json", status=sidecar_status), + ] + + def _download(s3_uri: str, local_path: Path, **_: Any) -> None: + local_path.parent.mkdir(parents=True, exist_ok=True) + local_path.write_text("not json") + + with _patched_process( + tmp_path, + client, + list_objects=["s3://raw/dishcam/run-xyz/stack.tif"], + download=_download, + ) as (process_file, encode): + with pytest.raises(ValueError, match="Invalid run.json"): + process_file("dishcam", "run-xyz", "run.json") + + encode.assert_not_called() + assert _status_updates(client, 10) == tiff_expected + assert _status_updates(client, SIDECAR_ID) == sidecar_expected + + def test_sidecar_complete_error_does_not_hide_encode_error(self, tmp_path: Path) -> None: + client = MagicMock() + client.ensure_run.return_value = _run_response() + client.create_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json"), + _file_response(10, "empty.tif"), + ] + + def _update(file_id: int, **kwargs: Any) -> FileResponse: + if file_id == SIDECAR_ID and kwargs.get("status") == "completed": + raise ApiError("sidecar complete failed", status_code=500) + return _file_response(file_id, "x", status=kwargs.get("status", "uploaded")) + + client.update_file.side_effect = _update + + def _encode(tiff_path: Path, mp4_path: Path, poster_path: Path, fps: float) -> None: + raise RuntimeError("empty stack is corrupt") + + with _patched_process( + tmp_path, + client, + list_objects=["s3://raw/dishcam/run-xyz/empty.tif"], + encode=MagicMock(side_effect=_encode), + ) as (process_file, _encode_mock): + with pytest.raises(RuntimeError, match="corrupt"): + process_file("dishcam", "run-xyz", "run.json") + + client.update_run.assert_called_once() + + def test_sidecar_complete_error_is_raised_when_encode_succeeds(self, tmp_path: Path) -> None: + client = MagicMock() + client.ensure_run.return_value = _run_response() + client.create_file.side_effect = [ + _file_response(SIDECAR_ID, "run.json"), + _file_response(10, "stack.tif"), + _file_response(11, "stack.mp4", category="processed"), + _file_response(12, "stack.jpg", category="processed"), + ] + + def _update(file_id: int, **kwargs: Any) -> FileResponse: + if file_id == SIDECAR_ID and kwargs.get("status") == "completed": + raise ApiError("sidecar complete failed", status_code=500) + return _file_response(file_id, "x", status=kwargs.get("status", "uploaded")) + + client.update_file.side_effect = _update + + with _patched_process( + tmp_path, + client, + list_objects=["s3://raw/dishcam/run-xyz/stack.tif"], + ) as (process_file, _encode): + with pytest.raises(ApiError, match="sidecar complete failed"): + process_file("dishcam", "run-xyz", "run.json") diff --git a/lambda/tests/integration/test_lambda_api.py b/lambda/tests/integration/test_lambda_api.py index b1b6b5e..8d79ab2 100644 --- a/lambda/tests/integration/test_lambda_api.py +++ b/lambda/tests/integration/test_lambda_api.py @@ -277,6 +277,8 @@ def test_run_json_encodes_sibling_tiff( raw_tiff = next(f for f in run["files"] if f["filename"] == tiff_name) assert raw_tiff["status"] == "completed" + sidecar = next(f for f in run["files"] if f["filename"] == "run.json") + assert sidecar["status"] == "completed" processed = [f for f in run["files"] if f["category"] == "processed"] names = {f["filename"] for f in processed} @@ -312,6 +314,7 @@ def test_run_json_encodes_every_tiff( raw = {f["filename"]: f for f in run["files"] if f["category"] == "raw"} assert raw["empty.tif"]["status"] == "completed" assert raw["ruler.tif"]["status"] == "completed" + assert raw["run.json"]["status"] == "completed" processed = {f["filename"] for f in run["files"] if f["category"] == "processed"} assert processed == {