Migrate CLI infra state-bucket tests to the GCS emulator

docs/contributing.md

@@ -41,52 +41,144 @@ This project follows
    pre-commit install
    ```
 
-### Code quality and testing
+### Code quality
 
-Before submitting a pull request, please ensure your changes pass linting and unit tests.
+We use [Ruff](https://docs.astral.sh/ruff/) for linting and formatting. Run it before submitting a pull request:
 
-- **Linting:** We use [Ruff](https://docs.astral.sh/ruff/) for linting and formatting. Run it with:
-  ```bash
-  ruff check .
-  ```
-- **Unit tests:** We use [Pytest](https://docs.pytest.org/) for unit tests. Run them with:
+```bash
+ruff check . && ruff format --check .
+```
+
+### Testing
 
-  ```bash
-  pytest
-  ```
+Kinetic's tests are organised as a ladder of tiers. Each tier is more
+realistic than the one below it and needs one more thing installed. Every
+tier **skips itself cleanly** when its prerequisite is missing, so a fresh
+checkout always gets a green run — just a less thorough one. Install
+what you can; CI runs tiers 0 and 1 on every pull request.
 
-- **E2E tests:** End-to-end tests run real workloads against a GKE cluster. They live in `tests/e2e/` and are skipped by default unless explicitly enabled.
+| Tier | What it exercises | Needs | Runtime |
+| ---- | ----------------- | ----- | ------- |
+| Unit | Pure logic, orchestration above mocked seams | nothing | seconds |
+| 0 — emulator | Real GCS wire protocol via [fake-gcs-server](https://github.com/fsouza/fake-gcs-server) | the `fake-gcs-server` binary | seconds |
+| 1 — docker | The real runner image, run with the exact command the Job spec generates | Tier 0 + a Docker daemon | ~2 min cold, ~15 s warm |
+| e2e | Real workloads on a real GKE cluster | a GCP project (see below) | minutes |
 
-  **Prerequisites:**
-  - A GCP project with a provisioned GKE cluster.
-  - Google Cloud SDK authenticated (`gcloud auth login` and `gcloud auth application-default login`)
-  - GKE credentials configured: `gcloud container clusters get-credentials <KINETIC_CLUSTER> --zone <KINETIC_ZONE> --project <KINETIC_PROJECT>`
-  - Test dependencies installed: `uv pip install -e ".[test]"`
+Install test dependencies first:
 
-  **Required environment variables:**
+```bash
+uv pip install -e ".[test]"
+```
 
-  | Variable          | Required | Default         | Description                    |
-  | ----------------- | -------- | --------------- | ------------------------------ |
-  | `E2E_TESTS`       | Yes      | —               | Set to `1` to enable e2e tests |
-  | `KINETIC_PROJECT` | Yes      | —               | Google Cloud project ID        |
-  | `KINETIC_ZONE`    | No       | `us-central1-a` | GKE cluster zone               |
-  | `KINETIC_CLUSTER` | No       | `kinetic-cluster` | GKE cluster name             |
+#### Tier 0: the GCS emulator
 
-  **Run all e2e tests:**
+`fake-gcs-server` is the **canonical transport for every test that touches
+Cloud Storage** — there are no hand-written GCS mocks. It is a single
+static Go binary, not a Python package, so `pip` will not fetch it:
+
+```bash
+brew install fake-gcs-server
+```
 
-  ```bash
-  E2E_TESTS=1 KINETIC_PROJECT=my-project python -m pytest tests/e2e/ -v -n auto
-  ```
+Alternatives: `go install github.com/fsouza/fake-gcs-server@latest`, or
+download a [release tarball](https://github.com/fsouza/fake-gcs-server/releases)
+and point `FAKE_GCS_SERVER_BIN` at the extracted binary.
 
-  **Run a specific test file:**
+The test fixture (`kinetic/utils/fake_gcs_fixture.py`) finds the binary on
+your `PATH`, starts it on a free port, and stops it at exit — nothing to
+configure. Then run the unit and integration suites:
+
+```bash
+python -m unittest discover -s kinetic -p "*_test.py"
+```
+
+```bash
+python -m unittest discover -s tests/integration -t . -p "*_test.py"
+```
+
+Without the binary, emulator-backed tests skip with a message pointing
+here; everything else still runs.
+
+:::{note}
+The fixture exports `STORAGE_EMULATOR_HOST` for the whole process, so it
+refuses to start when `E2E_TESTS` is set. Run the e2e suite in a
+separate process, as CI does.
+:::
+
+#### Tier 1: the docker roundtrip
+
+This tier builds the actual runner image through kinetic's own build
+machinery and executes it with `docker run`, using the command line
+derived from the real Job spec, against the emulator. It needs Docker
+Desktop (or any Docker daemon) running:
+
+```bash
+python -m unittest discover -s tests/docker -t . -p "*_test.py"
+```
 
-  ```bash
-  E2E_TESTS=1 KINETIC_PROJECT=my-project python -m pytest tests/e2e/cpu_execution_test.py -v
-  ```
+The first run builds the image (a base pull plus the JAX/Keras/kinetic
+install); later runs reuse it by content hash until `remote_runner.py`
+or the Dockerfile template changes.
+
+:::{note}
+The image installs the *released* `keras-kinetic` version from PyPI —
+exactly what Cloud Build does — so a `version.py` bump past the latest
+release fails this tier's build until that version is published.
+:::
+
+#### E2E tests
+
+End-to-end tests run real workloads against a GKE cluster. They live in
+`tests/e2e/` and are skipped unless explicitly enabled.
+
+**Prerequisites:**
+- A GCP project with a provisioned Kinetic cluster and an active
+  profile — i.e. you have run `kinetic init` (which provisions or joins a
+  cluster and saves the profile).
+- Google Cloud SDK authenticated (`gcloud auth login` and `gcloud auth application-default login`).
+  Kinetic fetches the cluster's kubeconfig itself on first use.
+
+The tests submit jobs through `@kinetic.run`, so they resolve project,
+zone, and cluster exactly the way user code does: explicit argument,
+then `KINETIC_*` environment variable, then the active profile, then
+the built-in default. With a profile set, the only variable you need is
+`E2E_TESTS`:
+
+```bash
+E2E_TESTS=1 python -m pytest tests/e2e/ -v -n auto
+```
+
+**Run a specific test file:**
+
+```bash
+E2E_TESTS=1 python -m pytest tests/e2e/cpu_execution_test.py -v
+```
+
+**Optional overrides** — useful for pointing the suite at a cluster other
+than your active profile's (this is how CI runs it, with no profile on
+the runner):
+
+| Variable          | Overrides           | Default without a profile |
+| ----------------- | ------------------- | ------------------------- |
+| `KINETIC_PROJECT` | profile project     | `GOOGLE_CLOUD_PROJECT`, else required |
+| `KINETIC_ZONE`    | profile zone        | `us-central1-a`           |
+| `KINETIC_CLUSTER` | profile cluster     | `kinetic-cluster`         |
+
+:::{tip}
+Drop `-n auto` to run tests serially to make it easier to debug.
+:::
 
-  :::{tip}
-  Drop `-n auto` to run tests serially to make it easier to debug.
-  :::
+#### Writing new tests
+
+- Anything that touches Cloud Storage should use the emulator fixture
+  (`FakeGcsTestCase` or `shared_server()`), seed real blobs, and assert on
+  emulator state — never on log text.
+- Patching is fine for **fault injection** (simulating an outage or a
+  `Forbidden`) and for **spies** that record calls while the real code
+  runs. Do not patch to replace transport.
+- Tests above the storage seam (job polling, cleanup routing, batch
+  fan-out) may mock kinetic's own `storage` functions as collaborators;
+  the seam itself is covered for real by `tests/integration/`.
 
 ### Submitting changes
 

kinetic/cli/infra/state_backend_test.py

@@ -1,11 +1,22 @@
-"""Tests for kinetic.cli.infra.state_backend."""
+"""Tests for kinetic.cli.infra.state_backend.
 
+Bucket lifecycle runs against the real fake-gcs-server emulator: the
+derived bucket name is asserted by its actual existence, versioning by
+reading it back, and Conflict by a genuine duplicate create. The one
+thing the emulator does not persist is uniform bucket-level access, so
+that flag is asserted on the request itself via a spy over the real
+``create_bucket`` call.
+"""
+
+import uuid
 from unittest import mock
 
 from absl.testing import absltest
 from google.api_core import exceptions as gax
+from google.cloud import storage
 
 from kinetic.cli.infra import state_backend
+from kinetic.utils.fake_gcs_fixture import FakeGcsTestCase
 
 
 class StateBackendUrlTest(absltest.TestCase):
@@ -16,67 +27,86 @@ def test_derives_from_project(self):
     )
 
 
-class EnsureGcsBackendTest(absltest.TestCase):
+class EnsureGcsBackendTest(FakeGcsTestCase):
   """ensure_gcs_backend is best-effort. It tries to create the bucket once;
   Conflict / Forbidden / PermissionDenied are silently swallowed so that
   collaborators with only object-level perms reach Pulumi, which surfaces
   a clean object-level error if access is actually wrong."""
 
-  def _patch_client(self, bucket_mock):
-    client_mock = mock.MagicMock()
-    client_mock.bucket.return_value = bucket_mock
-    return (
-      mock.patch.object(
-        state_backend.storage, "Client", return_value=client_mock
-      ),
-      client_mock,
-    )
+  def _project(self):
+    """A unique project name so each test gets its own state bucket."""
+    return f"proj-{uuid.uuid4().hex[:12]}"
+
+  def test_creates_the_derived_bucket_with_versioning(self):
+    project = self._project()
+    expected_bucket = f"{project}-kinetic-state"
+
+    state_backend.ensure_gcs_backend(project)
 
-  def test_creates_with_versioning_and_ubla(self):
-    bucket = mock.MagicMock()
-    patcher, client_mock = self._patch_client(bucket)
-    with patcher:
-      state_backend.ensure_gcs_backend("my-proj")
+    bucket = storage.Client(project=self.PROJECT).get_bucket(expected_bucket)
+    self.assertEqual(bucket.name, expected_bucket)
     self.assertTrue(bucket.versioning_enabled)
-    self.assertTrue(
-      bucket.iam_configuration.uniform_bucket_level_access_enabled
-    )
-    client_mock.create_bucket.assert_called_once()
+
+  def test_requests_uniform_bucket_level_access(self):
+    # The emulator does not persist UBLA, so assert it on the request.
+    # A plain `wraps=` spy cannot do this: the real create_bucket
+    # reloads the bucket from the server's reply, so by the time
+    # call_args is inspected the flag reads False again. Snapshot the
+    # requested config at call time, then let the real call proceed.
+    project = self._project()
+    requested = {}
+    real_create = state_backend.storage.Client.create_bucket
+
+    def spy(client, bucket, *args, **kwargs):
+      requested["ubla"] = (
+        bucket.iam_configuration.uniform_bucket_level_access_enabled
+      )
+      requested["versioning"] = bucket.versioning_enabled
+      return real_create(client, bucket, *args, **kwargs)
+
+    with mock.patch.object(state_backend.storage.Client, "create_bucket", spy):
+      state_backend.ensure_gcs_backend(project)
+
+    self.assertEqual(requested, {"ubla": True, "versioning": True})
 
   def test_storage_client_pinned_to_project(self):
-    bucket = mock.MagicMock()
-    with mock.patch.object(state_backend.storage, "Client") as client_cls:
-      client_cls.return_value.bucket.return_value = bucket
-      state_backend.ensure_gcs_backend("kinetic-proj")
-    client_cls.assert_called_once_with(project="kinetic-proj")
-
-  def test_bucket_name_derived_from_project(self):
-    bucket = mock.MagicMock()
-    patcher, client_mock = self._patch_client(bucket)
-    with patcher:
-      state_backend.ensure_gcs_backend("my-proj")
-    client_mock.bucket.assert_called_once_with("my-proj-kinetic-state")
-
-  def test_conflict_swallowed_for_collaborators(self):
-    bucket = mock.MagicMock()
-    patcher, client_mock = self._patch_client(bucket)
-    client_mock.create_bucket.side_effect = gax.Conflict("exists")
-    with patcher:
-      state_backend.ensure_gcs_backend("my-proj")  # no exception
+    project = self._project()
+    with mock.patch.object(
+      state_backend.storage, "Client", wraps=state_backend.storage.Client
+    ) as client_cls:
+      state_backend.ensure_gcs_backend(project)
+    client_cls.assert_called_once_with(project=project)
+
+  def test_existing_bucket_is_left_alone(self):
+    """A real second create raises Conflict, which is swallowed."""
+    project = self._project()
+    state_backend.ensure_gcs_backend(project)
+
+    state_backend.ensure_gcs_backend(project)  # no exception
+
+    self.assertTrue(
+      storage.Client(project=self.PROJECT)
+      .bucket(f"{project}-kinetic-state")
+      .exists()
+    )
 
   def test_forbidden_swallowed_for_collaborators(self):
-    bucket = mock.MagicMock()
-    patcher, client_mock = self._patch_client(bucket)
-    client_mock.create_bucket.side_effect = gax.Forbidden("nope")
-    with patcher:
-      state_backend.ensure_gcs_backend("my-proj")  # no exception
+    # Fault injection: the emulator has no IAM, so a Forbidden create
+    # can only be simulated.
+    with mock.patch.object(
+      state_backend.storage.Client,
+      "create_bucket",
+      side_effect=gax.Forbidden("nope"),
+    ):
+      state_backend.ensure_gcs_backend(self._project())  # no exception
 
   def test_permission_denied_swallowed(self):
-    bucket = mock.MagicMock()
-    patcher, client_mock = self._patch_client(bucket)
-    client_mock.create_bucket.side_effect = gax.PermissionDenied("nope")
-    with patcher:
-      state_backend.ensure_gcs_backend("my-proj")  # no exception
+    with mock.patch.object(
+      state_backend.storage.Client,
+      "create_bucket",
+      side_effect=gax.PermissionDenied("nope"),
+    ):
+      state_backend.ensure_gcs_backend(self._project())  # no exception
 
 
 if __name__ == "__main__":