fix: s3 tests

Author: Lundez

.github/workflows/ci.yml

@@ -38,3 +38,24 @@ jobs:
 
             - name: Build package
               run: uv build
+
+    s3-integration:
+        runs-on: ubuntu-latest
+
+        steps:
+            - name: Check out repository
+              uses: actions/checkout@v4
+
+            - name: Set up Python
+              uses: actions/setup-python@v5
+              with:
+                  python-version: "3.12"
+
+            - name: Set up uv
+              uses: astral-sh/setup-uv@v4
+
+            - name: Install dependencies
+              run: uv sync --group dev
+
+            - name: Run S3 integration tests against MiniStack
+              run: bash scripts/run_s3_integration.sh

README.md

@@ -200,6 +200,22 @@ uv run fluxel verify --repo /tmp/fluxel-demo --ref main --path images --path roo
 
 Fluxel includes `integration`-marked tests for real S3-compatible behavior. The preferred target is Ministack.
 
+For the standard local workflow, run a single command from the repository root:
+
+```bash
+bash scripts/run_s3_integration.sh
+```
+
+That script starts a temporary Ministack container on `127.0.0.1:4566`, waits for the health endpoint, resets emulator state, runs `tests/test_s3_integration.py`, and cleans up the container when the test run finishes.
+
+If you prefer task-runner aliases, the repo also provides:
+
+```bash
+make test-s3-integration
+```
+
+GitHub Actions runs the same script in the dedicated S3 integration job.
+
 Start Ministack locally:
 
 ```bash

scripts/run_s3_integration.sh

@@ -0,0 +1,91 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+MINISTACK_IMAGE="${FLUXEL_MINISTACK_IMAGE:-nahuelnucera/ministack}"
+MINISTACK_HOST="${FLUXEL_MINISTACK_HOST:-127.0.0.1}"
+MINISTACK_PORT="${FLUXEL_MINISTACK_PORT:-4566}"
+DEFAULT_ENDPOINT="http://${MINISTACK_HOST}:${MINISTACK_PORT}"
+MANAGE_CONTAINER="${FLUXEL_MINISTACK_MANAGE_CONTAINER:-auto}"
+
+export FLUXEL_MINISTACK_ENDPOINT="${FLUXEL_MINISTACK_ENDPOINT:-$DEFAULT_ENDPOINT}"
+export FLUXEL_MINISTACK_ACCESS_KEY="${FLUXEL_MINISTACK_ACCESS_KEY:-test}"
+export FLUXEL_MINISTACK_SECRET_KEY="${FLUXEL_MINISTACK_SECRET_KEY:-test}"
+export FLUXEL_MINISTACK_REGION="${FLUXEL_MINISTACK_REGION:-us-east-1}"
+
+HEALTH_URL="${FLUXEL_MINISTACK_ENDPOINT%/}/_ministack/health"
+RESET_URL="${FLUXEL_MINISTACK_ENDPOINT%/}/_ministack/reset"
+
+container_started=0
+container_name="fluxel-ministack-${USER:-user}-$$"
+
+cleanup() {
+    if [[ "$container_started" -eq 1 ]]; then
+        docker rm -f "$container_name" >/dev/null 2>&1 || true
+    fi
+}
+
+print_docker_daemon_help() {
+    case "$(uname -s)" in
+        Darwin)
+            echo "docker daemon is not running. Start Docker Desktop, Colima, or another local Docker runtime, then rerun this script." >&2
+            ;;
+        Linux)
+            echo "docker daemon is not running. Start Docker Engine or point docker at a running daemon, then rerun this script." >&2
+            ;;
+        *)
+            echo "docker daemon is not running. Start your local Docker runtime, then rerun this script." >&2
+            ;;
+    esac
+}
+
+wait_for_ministack() {
+    local attempt
+
+    for attempt in $(seq 1 30); do
+        if curl -fsS "$HEALTH_URL" >/dev/null; then
+            return 0
+        fi
+        sleep 1
+    done
+
+    echo "MiniStack did not become healthy at $HEALTH_URL" >&2
+    return 1
+}
+
+start_ministack_if_needed() {
+    if curl -fsS "$HEALTH_URL" >/dev/null 2>&1; then
+        return 0
+    fi
+
+    if [[ "$MANAGE_CONTAINER" == "0" ]]; then
+        return 0
+    fi
+
+    if [[ "$MANAGE_CONTAINER" == "auto" && "$FLUXEL_MINISTACK_ENDPOINT" != "$DEFAULT_ENDPOINT" ]]; then
+        return 0
+    fi
+
+    if ! command -v docker >/dev/null 2>&1; then
+        echo "docker is required to start MiniStack locally" >&2
+        return 1
+    fi
+
+    if ! docker info >/dev/null 2>&1; then
+        print_docker_daemon_help
+        return 1
+    fi
+
+    docker run -d --rm -p "${MINISTACK_PORT}:4566" --name "$container_name" "$MINISTACK_IMAGE" >/dev/null
+    container_started=1
+}
+
+trap cleanup EXIT
+
+start_ministack_if_needed
+wait_for_ministack
+curl -fsS -X POST "$RESET_URL" >/dev/null 2>&1 || true
+
+cd "$ROOT_DIR"
+uv run pytest tests/test_s3_integration.py -m integration "$@"

src/fluxel/core/client_state.py

@@ -1,21 +1,36 @@
 from __future__ import annotations
 
+import json
+from dataclasses import dataclass
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 
 
 HEAD_FILE = "HEAD"
 
 
+@dataclass(frozen=True)
+class LocalBranchSnapshot:
+    branch: str
+    commit_id: str | None
+    version_token: str | None
+
+
 class LocalClientState:
     def __init__(self, root: str | Path) -> None:
         self.root = Path(root).resolve()
         self.fluxel_dir = self.root / ".fluxel"
         self.refs_dir = self.fluxel_dir / "refs"
+        self.branch_state_dir = self.refs_dir / "heads"
         self.staging_dir = self.fluxel_dir / "staging"
         self.head_path = self.refs_dir / HEAD_FILE
 
-        for path in (self.fluxel_dir, self.refs_dir, self.staging_dir):
+        for path in (
+            self.fluxel_dir,
+            self.refs_dir,
+            self.branch_state_dir,
+            self.staging_dir,
+        ):
             path.mkdir(parents=True, exist_ok=True)
 
     def ensure_current_branch(self, default_branch: str) -> None:
@@ -44,6 +59,36 @@ def write_staging_payload(self, branch: str, payload: str | None) -> None:
             return
         self._atomic_write_text(stage_path, payload)
 
+    def read_branch_snapshot(self, branch: str) -> LocalBranchSnapshot | None:
+        snapshot_path = self.branch_snapshot_path(branch)
+        if not snapshot_path.exists():
+            return None
+        payload = json.loads(snapshot_path.read_text(encoding="utf-8"))
+        return LocalBranchSnapshot(
+            branch=branch,
+            commit_id=payload.get("commit_id"),
+            version_token=payload.get("version_token"),
+        )
+
+    def write_branch_snapshot(
+        self,
+        branch: str,
+        *,
+        commit_id: str | None,
+        version_token: str | None,
+    ) -> None:
+        payload = json.dumps(
+            {
+                "commit_id": commit_id,
+                "version_token": version_token,
+            },
+            sort_keys=True,
+        )
+        self._atomic_write_text(self.branch_snapshot_path(branch), f"{payload}\n")
+
+    def branch_snapshot_path(self, branch: str) -> Path:
+        return self.branch_state_dir / f"{branch}.json"
+
     def stage_path(self, branch: str) -> Path:
         return self.staging_dir / f"{branch}.json"
 

src/fluxel/core/repository.py

@@ -216,8 +216,16 @@ def branch(self, name: str) -> Path:
             name,
             head_commit or None,
             expected_version_token=None,
+            expected_commit_id=None,
         ):
             raise ValueError(f"Branch already exists: {name}")
+        created_state = self.store.read_branch_ref(name)
+        if created_state is not None:
+            self.client_state.write_branch_snapshot(
+                name,
+                commit_id=created_state.commit_id,
+                version_token=created_state.version_token,
+            )
         self._resolved_ref_cache.pop(name, None)
         return self._branch_path(name)
 
@@ -1123,9 +1131,22 @@ def _ensure_branch_exists(self, branch: str) -> None:
         self._require_branch_state(branch)
 
     def _require_branch_state(self, branch: str) -> BranchRefState:
+        cached_state = self.client_state.read_branch_snapshot(branch)
+        if cached_state is not None:
+            return BranchRefState(
+                branch=branch,
+                commit_id=cached_state.commit_id,
+                version_token=cached_state.version_token,
+            )
+
         branch_state = self.store.read_branch_ref(branch)
         if branch_state is None:
             raise ValueError(f"Unknown branch: {branch}")
+        self.client_state.write_branch_snapshot(
+            branch,
+            commit_id=branch_state.commit_id,
+            version_token=branch_state.version_token,
+        )
         return branch_state
 
     def _require_commit_for_metadata_mutation(self, branch: str) -> CommitObject:
@@ -1293,14 +1314,26 @@ def _update_branch_ref(
             branch,
             commit_id,
             expected_version_token=expected_version_token,
+            expected_commit_id=expected_commit_id,
         )
         if updated:
             self._resolved_ref_cache.pop(branch, None)
             current_state = self.store.read_branch_ref(branch)
             if current_state is not None:
                 self._resolved_ref_cache[branch] = current_state
+                self.client_state.write_branch_snapshot(
+                    branch,
+                    commit_id=current_state.commit_id,
+                    version_token=current_state.version_token,
+                )
             return
         current_state = self.store.read_branch_ref(branch)
+        if current_state is not None:
+            self.client_state.write_branch_snapshot(
+                branch,
+                commit_id=current_state.commit_id,
+                version_token=current_state.version_token,
+            )
         raise RefConflictError(
             branch=branch,
             operation=operation,

src/fluxel/core/repository_store.py

@@ -81,6 +81,7 @@ def compare_and_set_branch_ref(
         commit_id: str | None,
         *,
         expected_version_token: str | None,
+        expected_commit_id: str | None = None,
     ) -> bool: ...
 
     def read_blob_bytes(self, blob_hash: str) -> bytes: ...
@@ -185,10 +186,15 @@ def compare_and_set_branch_ref(
         commit_id: str | None,
         *,
         expected_version_token: str | None,
+        expected_commit_id: str | None = None,
     ) -> bool:
         current_token = self.version_token("ref", branch)
         if current_token != expected_version_token:
             return False
+        current_state = self.read_branch_ref(branch)
+        current_commit_id = current_state.commit_id if current_state else None
+        if current_commit_id != expected_commit_id:
+            return False
         self.write_branch_ref(branch, commit_id)
         return True
 
@@ -414,6 +420,7 @@ def compare_and_set_branch_ref(
         commit_id: str | None,
         *,
         expected_version_token: str | None,
+        expected_commit_id: str | None = None,
     ) -> bool:
         lock_token = str(uuid4())
         if not self._acquire_branch_lock(branch, lock_token):
@@ -423,6 +430,9 @@ def compare_and_set_branch_ref(
             current_version = current.version_token if current else None
             if current_version != expected_version_token:
                 return False
+            current_commit_id = current.commit_id if current else None
+            if current_commit_id != expected_commit_id:
+                return False
             self.write_branch_ref(branch, commit_id)
             return True
         finally:

tests/conftest.py

@@ -47,6 +47,7 @@ def paginate(self, *, Bucket: str, Prefix: str) -> list[dict[str, object]]:
 class FakeS3Client:
     def __init__(self, objects: dict[str, dict[str, object]]) -> None:
         self._objects = objects
+        self.fixed_etag: str | None = None
 
     def get_paginator(self, operation_name: str) -> FakeS3Paginator:
         assert operation_name == "list_objects_v2"
@@ -116,6 +117,8 @@ def delete_object(self, *, Bucket: str, Key: str) -> dict[str, object]:
         return {}
 
     def _etag(self, payload: bytes) -> str:
+        if self.fixed_etag is not None:
+            return self.fixed_etag
         return f'"{len(payload):x}-{sum(payload):x}"'
 
     def _client_error(self, code: str) -> ClientError: