refactory gcs tests with fake test

Author: Tingting-Chang

devtools/Tiltfile

@@ -30,6 +30,7 @@ components = {
     "ddb-local": [],
     "sfn-local": ["ddb-local"],
     "airflow": ["postgresql"],
+    "fake-gcs-server": [],
 }
 
 # ---------------------------------------------------------------------------
@@ -93,6 +94,7 @@ load('./tilt/localbatch.tiltfile', 'setup_localbatch')
 load('./tilt/ddb_local.tiltfile', 'setup_ddb_local')
 load('./tilt/sfn_local.tiltfile', 'setup_sfn_local')
 load('./tilt/airflow.tiltfile', 'setup_airflow')
+load('./tilt/fake_gcs_server.tiltfile', 'setup_fake_gcs_server')
 
 _SETUP = {
     "minio": setup_minio,
@@ -104,6 +106,7 @@ _SETUP = {
     "ddb-local": setup_ddb_local,
     "sfn-local": setup_sfn_local,
     "airflow": setup_airflow,
+    "fake-gcs-server": setup_fake_gcs_server,
 }
 
 # ---------------------------------------------------------------------------

devtools/pick_services.sh

@@ -21,6 +21,7 @@ SERVICE_OPTIONS=(
     "ddb-local"
     "sfn-local"
     "airflow"
+    "fake-gcs-server"
 )
 
 gum style "$LOGO" \

devtools/tilt/fake_gcs_server.tiltfile

@@ -0,0 +1,23 @@
+load('./_result.tiltfile', 'new_result')
+
+def setup_fake_gcs_server(ctx):
+    k8s_yaml(read_file('./tilt/k8s/fake-gcs-server.yaml'))
+    k8s_yaml(read_file('./tilt/k8s/fake-gcs-secret.yaml'))
+    k8s_yaml(read_file('./tilt/k8s/gcs-bucket-init-job.yaml'))
+
+    k8s_resource(
+        'fake-gcs-server',
+        port_forwards=['4443:4443'],
+        links=[link('http://localhost:4443/storage/v1/b', 'fake-gcs-server buckets')],
+        labels=['fake-gcs-server'],
+    )
+
+    k8s_resource('gcs-bucket-init', resource_deps=['fake-gcs-server'],
+                 labels=['fake-gcs-server'])
+
+    return new_result(
+        config={"METAFLOW_DATASTORE_SYSROOT_GS": "gs://metaflow-test/metaflow"},
+        shell_env={"STORAGE_EMULATOR_HOST": "http://localhost:4443"},
+        config_resources=['gcs-bucket-init'],
+        k8s_secrets=['fake-gcs-secret'],
+    )

devtools/tilt/k8s/fake-gcs-secret.yaml

@@ -0,0 +1,7 @@
+apiVersion: v1
+kind: Secret
+metadata:
+  name: fake-gcs-secret
+type: Opaque
+stringData:
+  STORAGE_EMULATOR_HOST: http://fake-gcs-server:4443

devtools/tilt/k8s/fake-gcs-server.yaml

@@ -0,0 +1,39 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: fake-gcs-server
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: fake-gcs-server
+  template:
+    metadata:
+      labels:
+        app: fake-gcs-server
+    spec:
+      containers:
+        - name: fake-gcs-server
+          image: fsouza/fake-gcs-server:1.49.2
+          args: ["-scheme", "http", "-host", "0.0.0.0", "-port", "4443"]
+          ports:
+            - containerPort: 4443
+          resources:
+            requests:
+              cpu: 50m
+              memory: 64Mi
+            limits:
+              cpu: 200m
+              memory: 128Mi
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: fake-gcs-server
+spec:
+  type: LoadBalancer
+  selector:
+    app: fake-gcs-server
+  ports:
+    - port: 4443
+      targetPort: 4443

devtools/tilt/k8s/gcs-bucket-init-job.yaml

@@ -0,0 +1,27 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: gcs-bucket-init
+spec:
+  ttlSecondsAfterFinished: 120
+  template:
+    spec:
+      restartPolicy: OnFailure
+      containers:
+        - name: init
+          image: curlimages/curl:8.11.1
+          command: ["/bin/sh", "-ec"]
+          args:
+            - |
+              curl -sf -X POST \
+                http://fake-gcs-server:4443/storage/v1/b \
+                -H "Content-Type: application/json" \
+                -d '{"name":"metaflow-test"}'
+              echo "Bucket 'metaflow-test' created."
+          resources:
+            requests:
+              cpu: 25m
+              memory: 32Mi
+            limits:
+              cpu: 100m
+              memory: 64Mi

metaflow/plugins/gcp/gs_storage_client_factory.py

@@ -12,12 +12,18 @@ def _get_gs_storage_client_default():
     cache_key = _get_cache_key()
     if cache_key not in _client_cache:
         from google.cloud import storage
-        import google.auth
 
-        credentials, project_id = google.auth.default(scopes=storage.Client.SCOPE)
-        _client_cache[cache_key] = storage.Client(
-            credentials=credentials, project=project_id
-        )
+        if os.environ.get("STORAGE_EMULATOR_HOST"):
+            # Emulator mode: anonymous client, no real GCP credentials needed.
+            # google-cloud-storage routes requests to STORAGE_EMULATOR_HOST automatically.
+            _client_cache[cache_key] = storage.Client()
+        else:
+            import google.auth
+
+            credentials, project_id = google.auth.default(scopes=storage.Client.SCOPE)
+            _client_cache[cache_key] = storage.Client(
+                credentials=credentials, project=project_id
+            )
     return _client_cache[cache_key]
 
 

test/core/conftest.py

@@ -1,15 +1,55 @@
+import importlib
+import json
 import os
 import sys
 from typing import Any
 
 import pytest
 
-# Ensure test/core/ is on sys.path so run_tests and metaflow_test are importable.
+from metaflow_test import MetaflowTest
+from metaflow_test.formatter import FlowFormatter
+
+# Ensure test/core/ is on sys.path so metaflow_test is importable.
 _CORE_DIR = os.path.dirname(os.path.abspath(__file__))
 if _CORE_DIR not in sys.path:
     sys.path.insert(0, _CORE_DIR)
 
 
+# ---------------------------------------------------------------------------
+# Test discovery — owned by pytest, no dependency on run_tests.py
+# ---------------------------------------------------------------------------
+
+
+def _iter_graphs():
+    root = os.path.join(_CORE_DIR, "graphs")
+    for graphfile in os.listdir(root):
+        if graphfile.endswith(".json") and not graphfile[0] == ".":
+            with open(os.path.join(root, graphfile)) as f:
+                yield json.load(f)
+
+
+def _iter_tests():
+    root = os.path.join(_CORE_DIR, "tests")
+    if root not in sys.path:
+        sys.path.insert(0, root)
+    for testfile in os.listdir(root):
+        if testfile.endswith(".py") and not testfile[0] == ".":
+            mod = importlib.import_module(testfile[:-3], "metaflow_test")
+            for name in dir(mod):
+                obj = getattr(mod, name)
+                if (
+                    name != "MetaflowTest"
+                    and isinstance(obj, type)
+                    and issubclass(obj, MetaflowTest)
+                ):
+                    yield obj()
+
+
+# ---------------------------------------------------------------------------
+# pytest hooks
+# ---------------------------------------------------------------------------
+
+
 def pytest_addoption(parser: Any) -> None:
     parser.addoption(
         "--core-tests",
@@ -28,9 +68,6 @@ def pytest_generate_tests(metafunc: Any) -> None:
         return
 
     try:
-        from run_tests import iter_graphs, iter_tests
-        from metaflow_test.formatter import FlowFormatter
-
         ok_tests_raw = metafunc.config.getoption("--core-tests", default=None)
         ok_graphs_raw = metafunc.config.getoption("--core-graphs", default=None)
         ok_tests = (
@@ -54,8 +91,8 @@ def pytest_generate_tests(metafunc: Any) -> None:
         disable_parallel = os.environ.get("METAFLOW_CORE_DISABLE_PARALLEL", "") == "1"
 
         mark = getattr(pytest.mark, marker_name)
-        all_tests = sorted(iter_tests(), key=lambda t: t.PRIORITY)
-        all_graphs = list(iter_graphs())
+        all_tests = sorted(_iter_tests(), key=lambda t: t.PRIORITY)
+        all_graphs = list(_iter_graphs())
 
         params = []
         for graph in all_graphs: