From 488742588efaffca8a09d6c2bdd566a0b9cbf193 Mon Sep 17 00:00:00 2001 From: Sujen Shah Date: Mon, 19 Oct 2026 21:00:27 -0700 Subject: [PATCH] fix: use maap-py v5 download_granule, pin maap-py, retry STAC fetches - stage_in: call download_granule (downloadGranule is a raising stub in maap-py v5, so every asset download failed with an unexpected 'online_access_url' keyword error) - pin maap-py==5.1.0 in environment.yaml and the Dockerfile - create_inputs: retry transient HTTP errors (e.g. CMR 503) with backoff and a timeout, and raise a clear error instead of an AttributeError when a STAC input cannot be loaded - tests: maap-py signature contract test, retry and download_granule tests --- create_inputs.py | 34 ++++++++++++------- docker/Dockerfile | 2 +- environment.yaml | 2 +- stage_in.py | 4 +-- tests/test_create_inputs.py | 26 ++++++++++++++- tests/test_stage_in.py | 65 +++++++++++++++++++++++-------------- 6 files changed, 92 insertions(+), 41 deletions(-) diff --git a/create_inputs.py b/create_inputs.py index 9a28831..b2c46ba 100644 --- a/create_inputs.py +++ b/create_inputs.py @@ -1,5 +1,6 @@ import json import pathlib +import time import yaml import os @@ -93,6 +94,9 @@ def convert_value_to_cwl_type(value, cwl_type_str): return value +HTTP_RETRIES = 4 + + def load_json_from_uri(uri): parsed = urllib.parse.urlparse(uri) @@ -110,17 +114,20 @@ def load_json_from_uri(uri): print(f"Failed to parse JSON: {e}") return None - # http / https - try: - response = requests.get(uri) - response.raise_for_status() - return response.json() - except requests.exceptions.RequestException as e: - print(f"Request error: {e}") - return None - except ValueError as e: - print(f"Failed to parse JSON: {e}") - return None + # http / https: retry transient failures (e.g. CMR STAC 503s) + for attempt in range(1, HTTP_RETRIES + 1): + try: + response = requests.get(uri, timeout=60) + response.raise_for_status() + return response.json() + except requests.exceptions.RequestException as e: + print(f"Request error (attempt {attempt}/{HTTP_RETRIES}): {e}") + if attempt < HTTP_RETRIES: + time.sleep(2 ** attempt) + except ValueError as e: + print(f"Failed to parse JSON: {e}") + return None + return None def parse_cwl_inputs(cwl_file_path) -> [CommandInputParameter]: @@ -211,7 +218,10 @@ def map_inputs_from_context(job_inputs: dict, cwl_inputs: [CommandInputParameter # URI, HySDS serves the stale cached version. Fetching from the URI at runtime # guarantees we always get the current content. # TODO: Update the HySDS spec builder to not map this input as a localize field. - stac_json = load_json_from_uri(job_inputs.get(input_name)) + stac_uri = job_inputs.get(input_name) + stac_json = load_json_from_uri(stac_uri) + if stac_json is None: + raise RuntimeError(f"Could not load STAC JSON for input '{input_name}' from {stac_uri}") output_dir = str(pathlib.Path(input_name).resolve()) stage_in.download_stac_assets_from_json(stac_json, output_dir) workflow_inputs[input_name] = {"class": "Directory", "path": output_dir} diff --git a/docker/Dockerfile b/docker/Dockerfile index 9c64f14..aded5a2 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -22,7 +22,7 @@ RUN pip install --no-cache-dir \ PyYAML \ awscli \ pystac==1.15.2 \ - maap-py \ + maap-py==5.1.0 \ requests \ "urllib3>=2" diff --git a/environment.yaml b/environment.yaml index fd98240..0e1c907 100644 --- a/environment.yaml +++ b/environment.yaml @@ -4,4 +4,4 @@ dependencies: - cwl_utils==0.35 - nodejs - pystac==1.15.2 - - maap-py \ No newline at end of file + - maap-py==5.1.0 \ No newline at end of file diff --git a/stage_in.py b/stage_in.py index 47d1d73..b146c0f 100644 --- a/stage_in.py +++ b/stage_in.py @@ -19,7 +19,7 @@ def _get_maap(): def _download_asset(href, destination_path): - """Route asset download: MAAP S3 buckets use boto3, everything else uses maap.downloadGranule.""" + """Route asset download: MAAP S3 buckets use boto3, everything else uses maap.download_granule.""" if href.startswith("s3://maap-"): parts = href[5:].split("/", 1) bucket, key = parts[0], parts[1] @@ -27,7 +27,7 @@ def _download_asset(href, destination_path): dest = os.path.join(destination_path, filename) boto3.client("s3").download_file(bucket, key, dest) return dest - return _get_maap().downloadGranule( + return _get_maap().download_granule( online_access_url=href, destination_path=destination_path, overwrite=True, diff --git a/tests/test_create_inputs.py b/tests/test_create_inputs.py index 94efd0f..f4c686f 100644 --- a/tests/test_create_inputs.py +++ b/tests/test_create_inputs.py @@ -1,7 +1,7 @@ import pytest import sys from pathlib import Path -from unittest.mock import Mock, patch +from unittest.mock import MagicMock, Mock, patch # Add parent directory to path to import create_inputs sys.path.insert(0, str(Path(__file__).parent.parent)) @@ -76,3 +76,27 @@ def test_empty_optional_file_does_not_affect_other_inputs(self): job_inputs = {"req_str": "x", "opt_str": ""} assert map_inputs_from_context(job_inputs, cwl_inputs) == {"req_str": "x"} + + +class TestLoadJsonFromUriRetry: + @patch('create_inputs.time.sleep') + @patch('create_inputs.requests.get') + def test_retries_transient_503_then_succeeds(self, mock_get, _sleep): + import requests + import create_inputs + bad = MagicMock() + bad.raise_for_status.side_effect = requests.exceptions.HTTPError("503") + good = MagicMock() + good.json.return_value = {"type": "Feature"} + mock_get.side_effect = [bad, bad, good] + assert create_inputs.load_json_from_uri("https://example.com/item") == {"type": "Feature"} + assert mock_get.call_count == 3 + + @patch('create_inputs.time.sleep') + @patch('create_inputs.requests.get') + def test_returns_none_after_exhausting_retries(self, mock_get, _sleep): + import requests + import create_inputs + mock_get.side_effect = requests.exceptions.ConnectionError("down") + assert create_inputs.load_json_from_uri("https://example.com/item") is None + assert mock_get.call_count == create_inputs.HTTP_RETRIES diff --git a/tests/test_stage_in.py b/tests/test_stage_in.py index 3d5ac00..b71ba28 100644 --- a/tests/test_stage_in.py +++ b/tests/test_stage_in.py @@ -228,8 +228,8 @@ def test_subdirectory_created(self, mock_get_maap, tmp_path): @patch('stage_in._get_maap') def test_download_granule_called_per_asset(self, mock_get_maap, tmp_path): - """maap.downloadGranule should be called once per asset with the correct URL.""" - mock_get_maap.return_value.downloadGranule.side_effect = [ + """maap.download_granule should be called once per asset with the correct URL.""" + mock_get_maap.return_value.download_granule.side_effect = [ str(tmp_path / "test-item-123" / "asset1.tif"), str(tmp_path / "test-item-123" / "asset2.json") ] @@ -251,15 +251,15 @@ def test_download_granule_called_per_asset(self, mock_get_maap, tmp_path): parent_container = Mock() _download_single_item(item_json, str(tmp_path), parent_container) - assert mock_get_maap.return_value.downloadGranule.call_count == 2 - calls = mock_get_maap.return_value.downloadGranule.call_args_list + assert mock_get_maap.return_value.download_granule.call_count == 2 + calls = mock_get_maap.return_value.download_granule.call_args_list assert calls[0][1]['online_access_url'] == "https://example.com/data.tif" assert calls[1][1]['online_access_url'] == "https://example.com/meta.json" @patch('stage_in._get_maap') def test_returned_item_added_to_parent(self, mock_get_maap, tmp_path): """The returned item should be added to the parent container.""" - mock_get_maap.return_value.downloadGranule.return_value = str(tmp_path / "test-item-123" / "asset.tif") + mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-123" / "asset.tif") item_json = { "type": "Feature", @@ -285,7 +285,7 @@ def test_returned_item_added_to_parent(self, mock_get_maap, tmp_path): @patch('stage_in._get_maap') def test_asset_hrefs_are_relative(self, mock_get_maap, tmp_path): """All asset hrefs on the returned item should be relative paths.""" - mock_get_maap.return_value.downloadGranule.return_value = str(tmp_path / "test-item-123" / "asset.tif") + mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-123" / "asset.tif") item_json = { "type": "Feature", @@ -316,7 +316,7 @@ def side_effect(*args, **kwargs): raise Exception("Download failed") return str(tmp_path / "test-item-123" / "success.tif") - mock_get_maap.return_value.downloadGranule.side_effect = side_effect + mock_get_maap.return_value.download_granule.side_effect = side_effect item_json = { "type": "Feature", @@ -336,7 +336,7 @@ def side_effect(*args, **kwargs): parent_container = Mock() result = _download_single_item(item_json, str(tmp_path), parent_container) - assert mock_get_maap.return_value.downloadGranule.call_count == 3 + assert mock_get_maap.return_value.download_granule.call_count == 3 assert len(result.assets) == 2 assert "good1" in result.assets assert "good2" in result.assets @@ -344,7 +344,7 @@ def side_effect(*args, **kwargs): @patch('stage_in._get_maap') def test_zero_assets_no_download_calls(self, mock_get_maap, tmp_path): - """Zero assets should result in downloadGranule never being called.""" + """Zero assets should result in download_granule never being called.""" item_json = { "type": "Feature", "stac_version": "1.0.0", @@ -359,7 +359,7 @@ def test_zero_assets_no_download_calls(self, mock_get_maap, tmp_path): parent_container = Mock() _download_single_item(item_json, str(tmp_path), parent_container) - mock_get_maap.return_value.downloadGranule.assert_not_called() + mock_get_maap.return_value.download_granule.assert_not_called() class TestStacMetadataPreservation: @@ -368,7 +368,7 @@ class TestStacMetadataPreservation: @patch('stage_in._get_maap') def test_extra_fields_preserved(self, mock_get_maap, tmp_path): """Extra fields (non-standard STAC properties) should be preserved via clone() and explicit update.""" - mock_get_maap.return_value.downloadGranule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") + mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") item_json = { "type": "Feature", @@ -395,7 +395,7 @@ def test_extra_fields_preserved(self, mock_get_maap, tmp_path): @patch('stage_in._get_maap') def test_extra_fields_survive_serialization(self, mock_get_maap, tmp_path): """Extra fields should survive to_dict() serialization for JSON output.""" - mock_get_maap.return_value.downloadGranule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") + mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") item_json = { "type": "Feature", @@ -423,7 +423,7 @@ def test_extra_fields_survive_serialization(self, mock_get_maap, tmp_path): @patch('stage_in._get_maap') def test_stac_extensions_preserved(self, mock_get_maap, tmp_path): """stac_extensions list should be preserved during clone.""" - mock_get_maap.return_value.downloadGranule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") + mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") item_json = { "type": "Feature", @@ -454,7 +454,7 @@ def test_stac_extensions_preserved(self, mock_get_maap, tmp_path): @patch('stage_in._get_maap') def test_stac_extensions_empty_list_preserved(self, mock_get_maap, tmp_path): """Empty stac_extensions list should be preserved.""" - mock_get_maap.return_value.downloadGranule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") + mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") item_json = { "type": "Feature", @@ -478,7 +478,7 @@ def test_stac_extensions_empty_list_preserved(self, mock_get_maap, tmp_path): @patch('stage_in._get_maap') def test_original_item_dict_not_mutated(self, mock_get_maap, tmp_path): """The input item_json dict should not be mutated by clone().""" - mock_get_maap.return_value.downloadGranule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") + mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") original_href = "https://example.com/data.tif" item_json = { @@ -503,7 +503,7 @@ def test_original_item_dict_not_mutated(self, mock_get_maap, tmp_path): @patch('stage_in._get_maap') def test_links_cleared_and_self_href_reset(self, mock_get_maap, tmp_path): """Remote links should be cleared and replaced with local self href.""" - mock_get_maap.return_value.downloadGranule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") + mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") item_json = { "type": "Feature", @@ -532,7 +532,7 @@ def test_links_cleared_and_self_href_reset(self, mock_get_maap, tmp_path): @patch('stage_in._get_maap') def test_self_href_set_to_local_path(self, mock_get_maap, tmp_path): """Self href should be the resolved absolute path of //.json.""" - mock_get_maap.return_value.downloadGranule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") + mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif") item_json = { "type": "Feature", @@ -605,7 +605,7 @@ def test_catalog_json_exists(self, mock_save, mock_get_maap, tmp_path): def test_catalog_contains_one_item(self, mock_save, mock_get_maap, tmp_path): """The catalog should contain exactly the one item from input.""" output_dir = tmp_path / "test_output" - mock_get_maap.return_value.downloadGranule.return_value = str(output_dir / "test-item" / "asset.tif") + mock_get_maap.return_value.download_granule.return_value = str(output_dir / "test-item" / "asset.tif") item_json = { "type": "Feature", @@ -832,7 +832,7 @@ def test_feature_collection_json_file_parsed_and_dispatched(self, mock_dispatch, class TestAssetRouting: - """Test _download_asset() routes to boto3 vs maap.downloadGranule based on URL.""" + """Test _download_asset() routes to boto3 vs maap.download_granule based on URL.""" @patch('stage_in.boto3') @patch('stage_in._get_maap') @@ -846,18 +846,18 @@ def test_maap_s3_url_uses_boto3(self, mock_maap, mock_boto3, tmp_path): mock_boto3.client.assert_called_once_with("s3") mock_s3.download_file.assert_called_once_with("maap-data-store", "path/to/file.tif", str(tmp_path / "file.tif")) - mock_maap.return_value.downloadGranule.assert_not_called() + mock_maap.return_value.download_granule.assert_not_called() assert dest == str(tmp_path / "file.tif") @patch('stage_in.boto3') @patch('stage_in._get_maap') def test_daac_url_uses_maap_download_granule(self, mock_maap, mock_boto3): href = "https://daac.gsfc.nasa.gov/data/file.tif" - mock_maap.return_value.downloadGranule.return_value = "/local/file.tif" + mock_maap.return_value.download_granule.return_value = "/local/file.tif" dest = _download_asset(href, "/local") - mock_maap.return_value.downloadGranule.assert_called_once_with( + mock_maap.return_value.download_granule.assert_called_once_with( online_access_url=href, destination_path="/local", overwrite=True ) mock_boto3.client.assert_not_called() @@ -867,11 +867,11 @@ def test_daac_url_uses_maap_download_granule(self, mock_maap, mock_boto3): @patch('stage_in._get_maap') def test_non_maap_s3_url_falls_through_to_maap(self, mock_maap, mock_boto3): href = "s3://some-other-bucket/data/file.tif" - mock_maap.return_value.downloadGranule.return_value = "/local/file.tif" + mock_maap.return_value.download_granule.return_value = "/local/file.tif" _download_asset(href, "/local") - mock_maap.return_value.downloadGranule.assert_called_once() + mock_maap.return_value.download_granule.assert_called_once() mock_boto3.client.assert_not_called() @@ -895,3 +895,20 @@ def test_client_created_once_on_demand(self, mock_cls): _download_asset("https://example.com/g.tif", "/tmp") mock_cls.assert_called_once() stage_in._maap = None + + +class TestMaapPyContract: + """Guard against maap-py API drift: stage_in must call a method the installed SDK really has.""" + + def test_download_granule_signature(self): + import inspect + from maap.maap import MAAP + params = inspect.signature(MAAP.download_granule).parameters + assert {"online_access_url", "destination_path", "overwrite"} <= set(params) + + @patch('stage_in._get_maap') + def test_download_asset_uses_download_granule(self, mock_get_maap): + import stage_in + stage_in._download_asset("https://example.com/a.tif", "/tmp/dest") + mock_get_maap.return_value.download_granule.assert_called_once_with( + online_access_url="https://example.com/a.tif", destination_path="/tmp/dest", overwrite=True)