diff --git a/lib/ramble/ramble/test/gcs_fetch.py b/lib/ramble/ramble/test/gcs_fetch.py index 72c371ab0..1f903ed0d 100644 --- a/lib/ramble/ramble/test/gcs_fetch.py +++ b/lib/ramble/ramble/test/gcs_fetch.py @@ -6,7 +6,9 @@ # option. This file may not be copied, modified, or distributed # except according to those terms. +import io import os +from unittest import mock import pytest @@ -14,6 +16,8 @@ import ramble.fetch_strategy import ramble.stage +import spack.util.gcs + @pytest.mark.parametrize("_fetch_method", ["curl", "urllib"]) def test_gcsfetchstrategy_without_url(_fetch_method): @@ -58,26 +62,80 @@ def archive_file(self): fetcher.fetch() -@pytest.mark.network +class MockBlob: + def __init__(self, data=b'{"test": "data"}', content_type="application/json", exists_val=True): + self.data = data + self.content_type = content_type + self.content_encoding = None + self.content_language = None + self.md5_hash = None + self._exists = exists_val + + def exists(self): + return self._exists + + def open(self, mode="rb"): + return io.BytesIO(self.data) + + +class MockBucket: + def __init__(self, name, blob_obj=None): + self.name = name + self.blob_obj = blob_obj or MockBlob() + + def exists(self): + return True + + def create(self): + pass + + def blob(self, blob_path): + return self.blob_obj + + def get_blob(self, blob_path): + return self.blob_obj + + +class MockGcsClient: + def __init__(self, blob_obj=None): + self.blob_obj = blob_obj or MockBlob() + + def bucket(self, name): + return MockBucket(name, blob_obj=self.blob_obj) + + @pytest.mark.parametrize("_fetch_method", ["curl", "urllib"]) def test_gcsfetchstrategy_download(tmpdir, _fetch_method, monkeypatch): - """Ensure fetch of fie.""" - # Remove this env var, otherwise for newer google-auth lib, it will try to invoke the mTLS path - # and fail if pyopenssl is not installed. - monkeypatch.delenv("CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE", raising=False) - google_api_core_exceptions = pytest.importorskip("google.api_core.exceptions") - google_auth_exceptions = pytest.importorskip("google.auth.exceptions") - try: - testpath = str(tmpdir) - path = "gs://hpc-toolkit-demos/build_cache/index.json" - - with ramble.config.override("config:url_fetch_method", _fetch_method): - fetcher = ramble.fetch_strategy.GCSFetchStrategy(url=path) - with ramble.stage.InputStage(fetcher, name="test", path=testpath): - fetcher.fetch() - except google_api_core_exceptions.Forbidden as e: - pytest.skip(f"{e}") - except google_auth_exceptions.RefreshError as e: - pytest.skip(f"{e}") - except google_auth_exceptions.DefaultCredentialsError as e: - pytest.skip(f"{e}") + """Ensure GCS fetch downloads file properly using mock client.""" + mock_data = b'{"key": "value"}' + mock_blob = MockBlob(data=mock_data, content_type="application/json") + monkeypatch.setattr(spack.util.gcs, "gcs_client", lambda: MockGcsClient(blob_obj=mock_blob)) + + testpath = str(tmpdir) + path = "gs://mock-bucket/build_cache/index.json" + + with ramble.config.override("config:url_fetch_method", _fetch_method): + fetcher = ramble.fetch_strategy.GCSFetchStrategy(url=path) + with ramble.stage.InputStage(fetcher, name="test", path=testpath): + fetcher.fetch() + downloaded = os.path.join(testpath, "index.json") + assert os.path.exists(downloaded) + with open(downloaded, "rb") as f: + assert f.read() == mock_data + + +def test_gcsfetchstrategy_content_type_mismatch(tmpdir, monkeypatch): + """Ensure GCS fetch warns when content type is text/html.""" + mock_warn = mock.MagicMock() + monkeypatch.setattr(ramble.fetch_strategy, "warn_content_type_mismatch", mock_warn) + + mock_blob = MockBlob(data=b"", content_type="text/html") + monkeypatch.setattr(spack.util.gcs, "gcs_client", lambda: MockGcsClient(blob_obj=mock_blob)) + + testpath = str(tmpdir) + path = "gs://mock-bucket/build_cache/index.json" + + fetcher = ramble.fetch_strategy.GCSFetchStrategy(url=path) + with ramble.stage.InputStage(fetcher, name="test", path=testpath): + fetcher.fetch() + mock_warn.assert_called_once()