test(integration): remove flaky dependencies from the core suite (ENG-3999)

tests/conftest.py

@@ -1,6 +1,7 @@
 """Pytest configuration and shared fixtures."""
 
 import os
+import uuid
 from pathlib import Path
 from unittest.mock import patch
 
@@ -11,6 +12,11 @@
 env_path = Path(__file__).parent.parent / ".env"
 load_dotenv(env_path)
 
+# Stamp the run id here, in a conftest the master process loads before it spawns
+# any pytest-xdist worker, so every worker inherits the same value. See
+# ``tests/helpers/utils.run_id``.
+os.environ.setdefault("BL_TEST_RUN_ID", uuid.uuid4().hex[:12])
+
 
 @pytest.fixture(scope="session", autouse=True)
 def setup_test_environment():

tests/core/test_cleanup_helpers.py

@@ -0,0 +1,59 @@
+"""Unit tests for the integration-suite cleanup predicates.
+
+The session teardown decides what to delete in a workspace shared with other
+CI runs, so a parsing slip here either leaves orphans forever or, worse, deletes
+someone else's sandbox.
+"""
+
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+from tests.helpers import ORPHAN_MAX_AGE, is_stale_orphan, resource_labels
+
+NOW = datetime(2026, 8, 18, 12, 0, 0, tzinfo=timezone.utc)
+
+
+def _resource(created_at=None, labels=None):
+    metadata = SimpleNamespace(created_at=created_at)
+    if labels is not None:
+        metadata.labels = SimpleNamespace(additional_properties=labels)
+    return SimpleNamespace(metadata=metadata)
+
+
+class TestIsStaleOrphan:
+    def test_accepts_every_timestamp_shape_the_api_returns(self):
+        """RFC3339 with a Z, with or without a fraction, up to nanoseconds."""
+        old = NOW - ORPHAN_MAX_AGE - timedelta(minutes=1)
+        stamp = old.strftime("%Y-%m-%dT%H:%M:%S")
+        for created_at in (
+            f"{stamp}Z",
+            f"{stamp}.123Z",
+            f"{stamp}.123456Z",
+            f"{stamp}.03583072Z",
+        ):
+            assert is_stale_orphan(_resource(created_at), now=NOW) is True, created_at
+
+    def test_recent_resource_is_not_stale(self):
+        """A live concurrent run must never have its sandboxes swept."""
+        recent = (NOW - timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%S.123456Z")
+        assert is_stale_orphan(_resource(recent), now=NOW) is False
+
+    def test_boundary_is_exclusive(self):
+        exactly = (NOW - ORPHAN_MAX_AGE).strftime("%Y-%m-%dT%H:%M:%SZ")
+        assert is_stale_orphan(_resource(exactly), now=NOW) is False
+
+    def test_unparsable_timestamp_fails_safe(self):
+        """Anything we cannot read is treated as live, never deleted."""
+        assert is_stale_orphan(_resource("not-a-date"), now=NOW) is False
+        assert is_stale_orphan(_resource(None), now=NOW) is False
+        assert is_stale_orphan(SimpleNamespace(metadata=None), now=NOW) is False
+
+
+class TestResourceLabels:
+    def test_reads_labels_from_additional_properties(self):
+        assert resource_labels(_resource(labels={"run-id": "abc"})) == {"run-id": "abc"}
+
+    def test_missing_labels_yield_empty_dict(self):
+        """Volume listings return metadata without a labels attribute at all."""
+        assert resource_labels(_resource()) == {}
+        assert resource_labels(SimpleNamespace(metadata=None)) == {}

tests/core/test_integration_cleanup.py

@@ -7,18 +7,18 @@
 from blaxel.core.client.models.get_workspace_features_response_200 import (
     GetWorkspaceFeaturesResponse200,
 )
+from tests.helpers import resource_labels
 from tests.integration.core.conftest import (
     _TEST_RESOURCE_LABELS,
     _is_test_resource,
-    _resource_labels,
 )
 from tests.integration.core.sandbox import test_volumes
 
 
 def test_lite_volume_without_labels_is_not_treated_as_a_test_resource():
     listed_volume = SimpleNamespace(metadata=SimpleNamespace(name="volume-from-list"))
 
-    assert _resource_labels(listed_volume) == {}
+    assert resource_labels(listed_volume) == {}
     assert _is_test_resource(listed_volume) is False
 
 

tests/helpers/__init__.py

@@ -1,27 +1,37 @@
 """Test helpers and utilities."""
 
 from .utils import (
+    ORPHAN_MAX_AGE,
     async_sleep,
     default_image,
     default_labels,
     default_region,
     env,
+    is_stale_orphan,
+    resource_labels,
+    run_id,
     sleep,
     unique_name,
     wait_for_sandbox_deletion,
     wait_for_sandbox_deployed,
     wait_for_volume_deletion,
+    wait_until,
 )
 
 __all__ = [
+    "ORPHAN_MAX_AGE",
     "async_sleep",
     "default_image",
     "default_labels",
     "default_region",
     "env",
+    "is_stale_orphan",
+    "resource_labels",
+    "run_id",
     "sleep",
     "unique_name",
     "wait_for_sandbox_deletion",
     "wait_for_sandbox_deployed",
     "wait_for_volume_deletion",
+    "wait_until",
 ]

tests/helpers/echo.py

@@ -0,0 +1,118 @@
+"""Self-hosted httpbin replacement for the proxy integration tests.
+
+The proxy suite needs an external HTTPS endpoint that echoes back the headers
+and body it received. It used to call ``httpbin.org``, which rate-limits the
+shared Blaxel egress IPs and answers ``503`` in bursts -- taking ~30 tests down
+with it on roughly half of the CI runs.
+
+Instead we host the echo ourselves: one sandbox, one node server, one public
+preview. Setup costs ~3s per pytest session and the endpoint is a real external
+HTTPS host from the sandbox's point of view, so the proxy/firewall paths under
+test are exercised exactly the same way.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from urllib.parse import urlparse
+
+from blaxel.core.sandbox import SandboxInstance
+
+from .utils import default_image, default_labels, default_region, unique_name
+
+# Serves the subset of httpbin routes the proxy suite actually used:
+# /headers, /get, /post, /put, /delete, /redirect/N, /bytes/N.
+ECHO_SERVER_SCRIPT = r"""
+const http = require("http");
+
+http.createServer((req, res) => {
+  let body = "";
+  req.on("data", (c) => (body += c));
+  req.on("end", () => {
+    if (req.url.startsWith("/redirect/")) {
+      const n = parseInt(req.url.split("/")[2] || "1", 10);
+      res.writeHead(302, { Location: n > 1 ? "/redirect/" + (n - 1) : "/get" });
+      res.end();
+      return;
+    }
+    if (req.url.startsWith("/bytes/")) {
+      const n = parseInt(req.url.split("/")[2] || "0", 10);
+      res.writeHead(200, { "Content-Type": "application/octet-stream" });
+      res.end(Buffer.alloc(n, "a"));
+      return;
+    }
+    let parsed = null;
+    try {
+      parsed = body ? JSON.parse(body) : null;
+    } catch (e) {}
+    const host = req.headers["x-forwarded-host"] || req.headers.host;
+    const proto = req.headers["x-forwarded-proto"] || "https";
+    res.writeHead(200, { "Content-Type": "application/json" });
+    res.end(
+      JSON.stringify({
+        headers: req.headers,
+        method: req.method,
+        url: proto + "://" + host + req.url,
+        data: body,
+        json: parsed,
+        args: {},
+      })
+    );
+  });
+}).listen(3000, "0.0.0.0", () => console.log("echo listening on 3000"));
+""".strip()
+
+_ECHO_PORT = 3000
+
+# Cached for the lifetime of the pytest process: every test class reuses the
+# same endpoint instead of paying the setup again.
+_echo_url: str | None = None
+_echo_lock = asyncio.Lock()
+
+
+async def echo_url() -> str:
+    """Return the base HTTPS URL of the shared echo server, creating it if needed.
+
+    The sandbox carries ``default_labels`` so the session-level cleanup in
+    ``tests/integration/core/conftest.py`` deletes it with everything else.
+    """
+    global _echo_url
+    if _echo_url is not None:
+        return _echo_url
+
+    async with _echo_lock:
+        if _echo_url is not None:
+            return _echo_url
+        _echo_url = await _create_echo_sandbox()
+    return _echo_url
+
+
+async def _create_echo_sandbox() -> str:
+    sandbox = await SandboxInstance.create_if_not_exists(
+        {
+            "name": unique_name("echo"),
+            "image": default_image,
+            "region": default_region,
+            "labels": default_labels,
+        }
+    )
+    await sandbox.fs.write("/tmp/echo-server.js", ECHO_SERVER_SCRIPT)
+    await sandbox.process.exec(
+        {
+            "name": "echo-server",
+            "command": "node /tmp/echo-server.js",
+            "wait_for_ports": [_ECHO_PORT],
+        }
+    )
+    preview = await sandbox.previews.create_if_not_exists(
+        {
+            "metadata": {"name": "echo"},
+            "spec": {"port": _ECHO_PORT, "public": True},
+        }
+    )
+    return preview.spec.url
+
+
+async def echo_host() -> str:
+    """Hostname of the echo server, for firewall allow/deny lists."""
+    return urlparse(await echo_url()).hostname

tests/helpers/utils.py

@@ -4,6 +4,7 @@
 import os
 import time
 import uuid
+from datetime import datetime, timedelta, timezone
 
 from blaxel.core.sandbox import SandboxInstance
 from blaxel.core.volume import VolumeInstance
@@ -13,25 +14,22 @@
 default_region = "eu-dub-1" if env == "dev" else "us-pdx-1"
 default_image = "blaxel/base-image:latest"
 
-# Default labels identify both the test owner and this specific test job. The
-# run label prevents one CI job's cleanup from deleting another job's resources
-# when they share a workspace.
-_test_run_id = (
-    "-".join(
-        part
-        for part in (
-            os.environ.get("GITHUB_RUN_ID"),
-            os.environ.get("GITHUB_JOB"),
-            os.environ.get("GITHUB_RUN_ATTEMPT"),
-        )
-        if part
-    )
-    or f"local-{os.getpid()}"
-)
+# Unique per pytest run. CI runs of several PRs share one workspace, so the
+# end-of-session cleanup must only delete what *this* run created -- deleting by
+# ``env=integration-test`` alone tears down sandboxes a concurrent run is still
+# using, which is a large part of the suite's cross-run flakiness.
+#
+# Kept in the environment rather than in module state so that pytest-xdist
+# workers, which import this module in their own process, share the master's id
+# instead of minting one each. Otherwise the master's teardown -- the only one
+# that runs -- would match nothing the workers created.
+run_id = os.environ.setdefault("BL_TEST_RUN_ID", uuid.uuid4().hex[:12])
+
+# Default labels to identify test sandboxes in the UI
 default_labels = {
     "env": "integration-test",
     "created-by": "pytest",
-    "test-run": _test_run_id,
+    "run-id": run_id,
 }
 
 
@@ -122,6 +120,50 @@ async def wait_for_volume_deletion(volume_name: str, max_attempts: int = 30) ->
     return False
 
 
+# Orphans older than this were left behind by a crashed run, never by a live one.
+ORPHAN_MAX_AGE = timedelta(hours=2)
+
+
+def resource_labels(resource) -> dict:
+    """Labels of a sandbox/volume, or an empty dict when the API omits them."""
+    metadata = getattr(resource, "metadata", None)
+    labels = getattr(metadata, "labels", None) if metadata else None
+    if isinstance(labels, dict):
+        return labels
+    return getattr(labels, "additional_properties", {}) or {}
+
+
+def is_stale_orphan(resource, now: datetime | None = None) -> bool:
+    """True for a pytest resource old enough that no live run still needs it."""
+    created_at = getattr(getattr(resource, "metadata", None), "created_at", None)
+    if not isinstance(created_at, str):
+        return False
+    try:
+        # Timestamps arrive as RFC3339 with a trailing Z and up to nanosecond
+        # precision, both of which fromisoformat rejects on Python 3.10: drop
+        # the Z and truncate the fraction to microseconds.
+        head, _, tail = created_at.rstrip("Zz").partition(".")
+        created = datetime.fromisoformat(f"{head}.{tail[:6]}+00:00" if tail else f"{head}+00:00")
+    except ValueError:
+        return False
+    return (now or datetime.now(timezone.utc)) - created > ORPHAN_MAX_AGE
+
+
+async def wait_until(predicate, timeout: float = 10.0, interval: float = 0.1) -> bool:
+    """Poll ``predicate`` until it is true or ``timeout`` elapses.
+
+    Callbacks (watch events, log streams) usually fire in well under a second,
+    but a fixed sleep turns a slow round-trip into a test failure. Polling keeps
+    the fast path fast and the slow path green.
+    """
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return True
+        await asyncio.sleep(interval)
+    return predicate()
+
+
 def sleep(seconds: float) -> None:
     """Synchronous sleep helper."""
     time.sleep(seconds)