Skip to content

Replace live gcs fetch test with mocked #1753

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rfbgo merged 1 commit into from
Sep 4, 2026
Merged

Replace live gcs fetch test with mocked #1753

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 79 additions & 21 deletions lib/ramble/ramble/test/gcs_fetch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,14 +6,18 @@
# 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

import ramble.config
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):
Expand Down Expand Up @@ -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"<html></html>", 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()
Loading