mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-14 16:08:41 +00:00
630 lines
24 KiB
Python
630 lines
24 KiB
Python
"""Regression tests for provisioner request-path K8s IO threading."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import inspect
|
|
import logging
|
|
import threading
|
|
import time
|
|
from contextlib import contextmanager
|
|
from types import SimpleNamespace
|
|
|
|
import httpx
|
|
import pytest
|
|
from blockbuster import BlockBuster
|
|
from kubernetes.client.rest import ApiException
|
|
|
|
|
|
def test_provisioner_thread_id_pattern_matches_gateway_contract(provisioner_module) -> None:
|
|
from deerflow.utils.thread_id import THREAD_ID_PATTERN
|
|
|
|
assert provisioner_module.SAFE_THREAD_ID_PATTERN == THREAD_ID_PATTERN
|
|
|
|
|
|
@pytest.mark.parametrize("thread_id", ["", "thread.with.dot", "../escape", "x" * 65])
|
|
def test_provisioner_rejects_noncanonical_thread_ids(provisioner_module, thread_id: str) -> None:
|
|
from pydantic import ValidationError
|
|
|
|
with pytest.raises(ValidationError):
|
|
provisioner_module.CreateSandboxRequest(
|
|
sandbox_id="sandbox-validation",
|
|
thread_id=thread_id,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("thread_id", ["a", "A1_b-2", "x" * 64])
|
|
def test_provisioner_accepts_canonical_thread_ids(provisioner_module, thread_id: str) -> None:
|
|
request = provisioner_module.CreateSandboxRequest(
|
|
sandbox_id="sandbox-validation",
|
|
thread_id=thread_id,
|
|
)
|
|
|
|
assert request.thread_id == thread_id
|
|
|
|
|
|
def test_sandbox_access_url_sanitizes_transient_error_log(
|
|
caplog: pytest.LogCaptureFixture,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provisioner_module,
|
|
) -> None:
|
|
class FailingCoreV1:
|
|
def read_namespaced_service(self, _name: str, _namespace: str):
|
|
raise ApiException(status=500, reason="upstream\nforged\rentry")
|
|
|
|
monkeypatch.setattr(provisioner_module, "core_v1", FailingCoreV1())
|
|
caplog.set_level(logging.WARNING, logger=provisioner_module.logger.name)
|
|
|
|
assert (
|
|
provisioner_module._sandbox_access_url(
|
|
"sandbox\nforged",
|
|
tolerate_read_errors=True,
|
|
)
|
|
is None
|
|
)
|
|
|
|
assert len(caplog.records) == 1
|
|
message = caplog.records[0].getMessage()
|
|
assert "sandboxforged" in message
|
|
assert "upstreamforgedentry" in message
|
|
assert "\n" not in message
|
|
assert "\r" not in message
|
|
|
|
|
|
def test_create_rejects_mount_contract_precondition_before_k8s_io(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provisioner_module,
|
|
) -> None:
|
|
fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=-1)
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
required_version = provisioner_module.MOUNT_CONTRACT_VERSION + 1
|
|
|
|
with pytest.raises(provisioner_module.HTTPException) as exc_info:
|
|
provisioner_module.create_sandbox(
|
|
provisioner_module.CreateSandboxRequest(
|
|
sandbox_id="sandbox-wrong-contract",
|
|
thread_id="thread-1",
|
|
user_id="alice",
|
|
required_mount_contract_version=required_version,
|
|
)
|
|
)
|
|
|
|
assert exc_info.value.status_code == 409
|
|
assert exc_info.value.detail == {
|
|
"code": "mount_contract_changed",
|
|
"expected": required_version,
|
|
"actual": provisioner_module.MOUNT_CONTRACT_VERSION,
|
|
}
|
|
assert fake_core_v1.thread_ids == []
|
|
|
|
|
|
class _RecordingCoreV1:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
event_loop_thread_id: int,
|
|
ready_after_service_reads: dict[str, int] | None = None,
|
|
service_read_failures: dict[str, list[int]] | None = None,
|
|
) -> None:
|
|
self.event_loop_thread_id = event_loop_thread_id
|
|
self.thread_ids: list[int] = []
|
|
self.service_sandboxes: set[str] = {"sandbox-existing"}
|
|
self.ready_after_service_reads = ready_after_service_reads or {}
|
|
self.service_read_failures = service_read_failures or {}
|
|
self.service_read_counts: dict[str, int] = {}
|
|
self.created_pods: list[str] = []
|
|
self.created_pod_specs: dict[str, object] = {}
|
|
self.existing_pod_specs: dict[str, object] = {}
|
|
self.missing_pods: set[str] = set()
|
|
self.pod_read_counts: dict[str, int] = {}
|
|
self.created_services: list[str] = []
|
|
self.namespace_uid = "namespace-uid-1"
|
|
|
|
def _record_k8s_call(self) -> None:
|
|
thread_id = threading.get_ident()
|
|
self.thread_ids.append(thread_id)
|
|
time.sleep(0)
|
|
if thread_id == self.event_loop_thread_id:
|
|
raise AssertionError("Kubernetes client call ran on the ASGI event-loop thread")
|
|
try:
|
|
asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
return
|
|
raise AssertionError("Kubernetes client call ran inside an asyncio event loop")
|
|
|
|
def read_namespaced_service(self, _name: str, _namespace: str):
|
|
self._record_k8s_call()
|
|
sandbox_id = _sandbox_id_from_service_name(_name)
|
|
self.service_read_counts[sandbox_id] = self.service_read_counts.get(sandbox_id, 0) + 1
|
|
failures = self.service_read_failures.get(sandbox_id) or []
|
|
if failures:
|
|
raise ApiException(status=failures.pop(0))
|
|
ready_after_reads = self.ready_after_service_reads.get(sandbox_id, 1)
|
|
if sandbox_id not in self.service_sandboxes or self.service_read_counts[sandbox_id] < ready_after_reads:
|
|
raise ApiException(status=404)
|
|
return _node_port_service(sandbox_id)
|
|
|
|
def read_namespaced_pod(self, _name: str, _namespace: str):
|
|
self._record_k8s_call()
|
|
sandbox_id = _name.removeprefix("sandbox-")
|
|
self.pod_read_counts[sandbox_id] = self.pod_read_counts.get(sandbox_id, 0) + 1
|
|
if sandbox_id in self.missing_pods:
|
|
raise ApiException(status=404)
|
|
pod = self.existing_pod_specs.get(sandbox_id) or self.created_pod_specs.get(sandbox_id)
|
|
if pod is not None:
|
|
if not getattr(pod.metadata, "uid", None):
|
|
pod.metadata.uid = f"pod-uid-{sandbox_id}"
|
|
pod.status = SimpleNamespace(phase="Running")
|
|
return pod
|
|
return SimpleNamespace(status=SimpleNamespace(phase="Running"))
|
|
|
|
def read_namespace(self, _namespace: str):
|
|
self._record_k8s_call()
|
|
return SimpleNamespace(metadata=SimpleNamespace(uid=self.namespace_uid))
|
|
|
|
def create_namespaced_pod(self, _namespace: str, pod) -> None:
|
|
self._record_k8s_call()
|
|
sandbox_id = pod.metadata.labels["sandbox-id"]
|
|
self.created_pods.append(sandbox_id)
|
|
self.created_pod_specs[sandbox_id] = pod
|
|
|
|
def create_namespaced_service(self, _namespace: str, service) -> None:
|
|
self._record_k8s_call()
|
|
sandbox_id = service.metadata.labels["sandbox-id"]
|
|
self.created_services.append(sandbox_id)
|
|
self.service_sandboxes.add(sandbox_id)
|
|
|
|
def delete_namespaced_service(self, _name: str, _namespace: str) -> None:
|
|
self._record_k8s_call()
|
|
|
|
def delete_namespaced_pod(self, _name: str, _namespace: str) -> None:
|
|
self._record_k8s_call()
|
|
|
|
def list_namespaced_service(self, _namespace: str, *, label_selector: str):
|
|
self._record_k8s_call()
|
|
assert label_selector == "app=deer-flow-sandbox"
|
|
return SimpleNamespace(items=[_node_port_service("sandbox-listed")])
|
|
|
|
|
|
def _node_port_service(sandbox_id: str):
|
|
return SimpleNamespace(
|
|
metadata=SimpleNamespace(labels={"sandbox-id": sandbox_id}),
|
|
spec=SimpleNamespace(ports=[SimpleNamespace(name="http", port=8080, node_port=32123)]),
|
|
)
|
|
|
|
|
|
def _sandbox_id_from_service_name(name: str) -> str:
|
|
assert name.startswith("sandbox-")
|
|
assert name.endswith("-svc")
|
|
return name[len("sandbox-") : -len("-svc")]
|
|
|
|
|
|
@contextmanager
|
|
def _detect_provisioner_blocking_io(provisioner_module):
|
|
detector = BlockBuster(scanned_modules=[provisioner_module.__name__])
|
|
detector.activate()
|
|
try:
|
|
yield
|
|
finally:
|
|
detector.deactivate()
|
|
|
|
|
|
def test_sandbox_business_route_handlers_are_sync(provisioner_module) -> None:
|
|
"""FastAPI runs sync handlers in its worker pool, away from the event loop."""
|
|
for handler in (
|
|
provisioner_module.create_sandbox,
|
|
provisioner_module.destroy_sandbox,
|
|
provisioner_module.reconcile_sandbox_identity,
|
|
provisioner_module.get_sandbox,
|
|
provisioner_module.list_sandboxes,
|
|
):
|
|
assert not inspect.iscoroutinefunction(handler)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("method", "path", "json_body", "expected_created_sandbox"),
|
|
[
|
|
("POST", "/api/sandboxes", {"sandbox_id": "sandbox-existing", "thread_id": "thread-1", "user_id": "user-1"}, None),
|
|
("POST", "/api/sandboxes", {"sandbox_id": "sandbox-new", "thread_id": "thread-1", "user_id": "user-1"}, "sandbox-new"),
|
|
("DELETE", "/api/sandboxes/sandbox-existing", None, None),
|
|
("GET", "/api/sandboxes/sandbox-existing", None, None),
|
|
("GET", "/api/sandboxes", None, None),
|
|
],
|
|
ids=["create-existing", "create-new", "destroy", "get", "list"],
|
|
)
|
|
async def test_sandbox_business_routes_run_k8s_client_off_event_loop_thread(
|
|
method: str,
|
|
path: str,
|
|
json_body: dict[str, str] | None,
|
|
expected_created_sandbox: str | None,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provisioner_module,
|
|
) -> None:
|
|
fake_core_v1 = _RecordingCoreV1(
|
|
event_loop_thread_id=threading.get_ident(),
|
|
ready_after_service_reads={"sandbox-new": 3},
|
|
)
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
monkeypatch.setattr(provisioner_module, "PROVISIONER_API_KEY", "test-secret")
|
|
fake_core_v1.existing_pod_specs["sandbox-existing"] = provisioner_module._build_pod(
|
|
"sandbox-existing",
|
|
"thread-1",
|
|
user_id="user-1",
|
|
)
|
|
|
|
with _detect_provisioner_blocking_io(provisioner_module):
|
|
transport = httpx.ASGITransport(app=provisioner_module.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
headers = {"X-API-Key": "test-secret"}
|
|
if json_body is None:
|
|
response = await client.request(method, path, headers=headers)
|
|
else:
|
|
response = await client.request(method, path, json=json_body, headers=headers)
|
|
|
|
assert response.status_code == 200
|
|
assert fake_core_v1.thread_ids
|
|
if expected_created_sandbox is not None:
|
|
assert fake_core_v1.created_pods == [expected_created_sandbox]
|
|
assert fake_core_v1.created_services == [expected_created_sandbox]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconciliation_endpoint_reports_pod_absence_authoritatively(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provisioner_module,
|
|
) -> None:
|
|
fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=threading.get_ident())
|
|
fake_core_v1.missing_pods.add("sandbox-gone")
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
monkeypatch.setattr(provisioner_module, "PROVISIONER_API_KEY", "test-secret")
|
|
|
|
with _detect_provisioner_blocking_io(provisioner_module):
|
|
transport = httpx.ASGITransport(app=provisioner_module.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
response = await client.get(
|
|
"/api/reconciliation/sandboxes/sandbox-gone",
|
|
headers={"X-API-Key": "test-secret"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"sandbox_id": "sandbox-gone",
|
|
"status": "absent",
|
|
"backend_namespace": "namespace-uid-1",
|
|
"incarnation_id": None,
|
|
"sandbox_url": None,
|
|
"user_id": None,
|
|
"thread_id": None,
|
|
"mount_contract_version": None,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconciliation_endpoint_keeps_live_pod_when_service_is_missing(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provisioner_module,
|
|
) -> None:
|
|
fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=threading.get_ident())
|
|
fake_core_v1.existing_pod_specs["sandbox-partial"] = provisioner_module._build_pod(
|
|
"sandbox-partial",
|
|
"thread-1",
|
|
user_id="alice",
|
|
)
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
monkeypatch.setattr(provisioner_module, "PROVISIONER_API_KEY", "test-secret")
|
|
|
|
with _detect_provisioner_blocking_io(provisioner_module):
|
|
transport = httpx.ASGITransport(app=provisioner_module.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
response = await client.get(
|
|
"/api/reconciliation/sandboxes/sandbox-partial",
|
|
headers={"X-API-Key": "test-secret"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["status"] == "found"
|
|
assert payload["backend_namespace"] == "namespace-uid-1"
|
|
assert payload["incarnation_id"] == "pod-uid-sandbox-partial"
|
|
assert payload["sandbox_url"] is None
|
|
assert payload["user_id"] == "alice"
|
|
assert payload["thread_id"] == "thread-1"
|
|
assert fake_core_v1.pod_read_counts["sandbox-partial"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconciliation_endpoint_returns_service_url_for_live_pod(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provisioner_module,
|
|
) -> None:
|
|
fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=threading.get_ident())
|
|
fake_core_v1.existing_pod_specs["sandbox-live"] = provisioner_module._build_pod(
|
|
"sandbox-live",
|
|
"thread-1",
|
|
user_id="alice",
|
|
)
|
|
fake_core_v1.service_sandboxes.add("sandbox-live")
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
monkeypatch.setattr(provisioner_module, "PROVISIONER_API_KEY", "test-secret")
|
|
|
|
with _detect_provisioner_blocking_io(provisioner_module):
|
|
transport = httpx.ASGITransport(app=provisioner_module.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
response = await client.get(
|
|
"/api/reconciliation/sandboxes/sandbox-live",
|
|
headers={"X-API-Key": "test-secret"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["status"] == "found"
|
|
assert payload["incarnation_id"] == "pod-uid-sandbox-live"
|
|
assert payload["sandbox_url"] == provisioner_module._sandbox_url(
|
|
"sandbox-live",
|
|
node_port=32123,
|
|
)
|
|
assert fake_core_v1.pod_read_counts["sandbox-live"] == 1
|
|
|
|
|
|
def test_existing_sandbox_rejects_cross_tenant_reuse(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provisioner_module,
|
|
) -> None:
|
|
fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=-1)
|
|
fake_core_v1.existing_pod_specs["sandbox-existing"] = provisioner_module._build_pod(
|
|
"sandbox-existing",
|
|
"thread-1",
|
|
user_id="alice",
|
|
)
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
|
|
with pytest.raises(provisioner_module.HTTPException) as exc_info:
|
|
provisioner_module.create_sandbox(
|
|
provisioner_module.CreateSandboxRequest(
|
|
sandbox_id="sandbox-existing",
|
|
thread_id="thread-1",
|
|
user_id="bob",
|
|
)
|
|
)
|
|
|
|
assert exc_info.value.status_code == 409
|
|
assert "another user" in exc_info.value.detail
|
|
|
|
|
|
def test_existing_sandbox_validates_read_only_conversion_before_fast_path(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provisioner_module,
|
|
) -> None:
|
|
fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=-1)
|
|
fake_core_v1.existing_pod_specs["sandbox-existing"] = provisioner_module._build_pod(
|
|
"sandbox-existing",
|
|
"thread-1",
|
|
user_id="alice",
|
|
)
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
|
|
with pytest.raises(provisioner_module.HTTPException) as exc_info:
|
|
provisioner_module.create_sandbox(
|
|
provisioner_module.CreateSandboxRequest(
|
|
sandbox_id="sandbox-existing",
|
|
thread_id="thread-1",
|
|
user_id="alice",
|
|
extra_mounts=[
|
|
provisioner_module.ExtraMount(
|
|
host_path=("/.deer-flow/users/alice/threads/thread-1/user-data/.upload-conversions"),
|
|
container_path="/mnt/user-data/.upload-conversions",
|
|
read_only=False,
|
|
)
|
|
],
|
|
)
|
|
)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "read-only" in exc_info.value.detail
|
|
|
|
|
|
def test_existing_sandbox_rejects_tampered_mount_spec(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provisioner_module,
|
|
) -> None:
|
|
fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=-1)
|
|
existing = provisioner_module._build_pod(
|
|
"sandbox-existing",
|
|
"thread-1",
|
|
user_id="alice",
|
|
)
|
|
userdata = next(volume for volume in existing.spec.volumes if volume.name == "user-data")
|
|
userdata.host_path.path = "/state/users/mallory/threads/thread-1/user-data"
|
|
fake_core_v1.existing_pod_specs["sandbox-existing"] = existing
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
|
|
with pytest.raises(provisioner_module.HTTPException) as exc_info:
|
|
provisioner_module.create_sandbox(
|
|
provisioner_module.CreateSandboxRequest(
|
|
sandbox_id="sandbox-existing",
|
|
thread_id="thread-1",
|
|
user_id="alice",
|
|
)
|
|
)
|
|
|
|
assert exc_info.value.status_code == 409
|
|
assert "incompatible or unverifiable" in exc_info.value.detail
|
|
|
|
|
|
def test_existing_sandbox_contract_ignores_kubernetes_service_account_injection(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provisioner_module,
|
|
) -> None:
|
|
"""Admission-added service-account mounts are outside the DeerFlow contract."""
|
|
fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=-1)
|
|
existing = provisioner_module._build_pod(
|
|
"sandbox-existing",
|
|
"thread-1",
|
|
user_id="alice",
|
|
)
|
|
existing.spec.volumes.append(
|
|
provisioner_module.k8s_client.V1Volume(
|
|
name="kube-api-access-abcde",
|
|
projected=provisioner_module.k8s_client.V1ProjectedVolumeSource(sources=[]),
|
|
)
|
|
)
|
|
existing.spec.containers[0].volume_mounts.append(
|
|
provisioner_module.k8s_client.V1VolumeMount(
|
|
name="kube-api-access-abcde",
|
|
mount_path="/var/run/secrets/kubernetes.io/serviceaccount",
|
|
read_only=True,
|
|
)
|
|
)
|
|
fake_core_v1.existing_pod_specs["sandbox-existing"] = existing
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
|
|
response = provisioner_module.create_sandbox(
|
|
provisioner_module.CreateSandboxRequest(
|
|
sandbox_id="sandbox-existing",
|
|
thread_id="thread-1",
|
|
user_id="alice",
|
|
)
|
|
)
|
|
|
|
assert response.status == "Running"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("include_legacy_skills", "expected_mount_names"),
|
|
[
|
|
(
|
|
False,
|
|
["skills-public", "skills-custom", "skills-legacy", "user-data"],
|
|
),
|
|
(
|
|
True,
|
|
["skills-public", "skills-custom", "skills-legacy", "user-data"],
|
|
),
|
|
],
|
|
ids=["without-legacy", "with-legacy"],
|
|
)
|
|
def test_create_sandbox_route_builds_expected_skills_mount_layout(
|
|
include_legacy_skills: bool,
|
|
expected_mount_names: list[str],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
provisioner_module,
|
|
) -> None:
|
|
fake_core_v1 = _RecordingCoreV1(
|
|
event_loop_thread_id=-1,
|
|
ready_after_service_reads={"sandbox-layout": 1},
|
|
)
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
|
|
response = provisioner_module.create_sandbox(
|
|
provisioner_module.CreateSandboxRequest(
|
|
sandbox_id="sandbox-layout",
|
|
thread_id="thread-1",
|
|
user_id="user-1",
|
|
include_legacy_skills=include_legacy_skills,
|
|
)
|
|
)
|
|
|
|
assert response.status == "Running"
|
|
pod = fake_core_v1.created_pod_specs["sandbox-layout"]
|
|
volume_names = [volume.name for volume in pod.spec.volumes]
|
|
mount_names = [mount.name for mount in pod.spec.containers[0].volume_mounts]
|
|
assert volume_names == expected_mount_names
|
|
assert mount_names == expected_mount_names
|
|
|
|
|
|
def test_create_sandbox_retries_transient_service_read_errors(monkeypatch: pytest.MonkeyPatch, provisioner_module) -> None:
|
|
fake_core_v1 = _RecordingCoreV1(
|
|
event_loop_thread_id=-1,
|
|
ready_after_service_reads={"sandbox-transient": 3},
|
|
service_read_failures={"sandbox-transient": [503, 429]},
|
|
)
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
monkeypatch.setattr(provisioner_module.time, "sleep", lambda _seconds: None)
|
|
|
|
response = provisioner_module.create_sandbox(
|
|
provisioner_module.CreateSandboxRequest(
|
|
sandbox_id="sandbox-transient",
|
|
thread_id="thread-1",
|
|
user_id="user-1",
|
|
)
|
|
)
|
|
|
|
assert response.status == "Running"
|
|
assert response.sandbox_url == provisioner_module._sandbox_url("sandbox-transient", node_port=32123)
|
|
assert fake_core_v1.service_read_counts["sandbox-transient"] == 3
|
|
|
|
|
|
def test_sandbox_service_defaults_to_node_port_with_node_host_url(provisioner_module) -> None:
|
|
provisioner_module.K8S_NAMESPACE = "mdv-sit"
|
|
provisioner_module.SANDBOX_CONTAINER_PORT = 8080
|
|
provisioner_module.SANDBOX_SERVICE_TYPE = "NodePort"
|
|
provisioner_module.NODE_HOST = "node.example"
|
|
|
|
service = provisioner_module._build_service("abc123")
|
|
|
|
assert service.spec.type == "NodePort"
|
|
assert service.spec.ports[0].port == 8080
|
|
assert service.spec.ports[0].target_port == 8080
|
|
assert provisioner_module._sandbox_url("abc123", node_port=32123) == "http://node.example:32123"
|
|
|
|
|
|
def test_sandbox_service_supports_cluster_ip_with_dns_url(provisioner_module) -> None:
|
|
provisioner_module.K8S_NAMESPACE = "mdv-sit"
|
|
provisioner_module.SANDBOX_CONTAINER_PORT = 8080
|
|
provisioner_module.SANDBOX_SERVICE_TYPE = "ClusterIP"
|
|
|
|
service = provisioner_module._build_service("abc123")
|
|
|
|
assert service.spec.type == "ClusterIP"
|
|
assert service.spec.ports[0].port == 8080
|
|
assert service.spec.ports[0].target_port == 8080
|
|
assert provisioner_module._sandbox_url("abc123") == ("http://sandbox-abc123-svc.mdv-sit.svc.cluster.local:8080")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_middleware(monkeypatch: pytest.MonkeyPatch, provisioner_module) -> None:
|
|
"""Verify the X-API-Key middleware: /health is open; /api/* requires a correct key."""
|
|
monkeypatch.setattr(provisioner_module, "PROVISIONER_API_KEY", "test-secret")
|
|
fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=-1)
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
|
|
transport = httpx.ASGITransport(app=provisioner_module.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
# /health is always open — no key needed
|
|
r = await client.get("/health")
|
|
assert r.status_code == 200
|
|
|
|
# /api/* with no header → 401
|
|
r = await client.get("/api/sandboxes")
|
|
assert r.status_code == 401
|
|
|
|
# /api/* with wrong key → 401
|
|
r = await client.get("/api/sandboxes", headers={"X-API-Key": "wrong-key"})
|
|
assert r.status_code == 401
|
|
|
|
# /api/* with correct key → not 401 (auth passed; handler runs with the K8s mock)
|
|
r = await client.get("/api/sandboxes", headers={"X-API-Key": "test-secret"})
|
|
assert r.status_code != 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_middleware_unset_key(monkeypatch: pytest.MonkeyPatch, provisioner_module) -> None:
|
|
"""When PROVISIONER_API_KEY is unset/empty, all /api/* routes return 401."""
|
|
monkeypatch.setattr(provisioner_module, "PROVISIONER_API_KEY", "")
|
|
fake_core_v1 = _RecordingCoreV1(event_loop_thread_id=-1)
|
|
monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
|
|
|
|
transport = httpx.ASGITransport(app=provisioner_module.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
# /health is always open even when key is unset
|
|
r = await client.get("/health")
|
|
assert r.status_code == 200
|
|
|
|
# /api/* is always 401 when key is unset — even with a header
|
|
r = await client.get("/api/sandboxes")
|
|
assert r.status_code == 401
|
|
|
|
r = await client.get("/api/sandboxes", headers={"X-API-Key": "anything"})
|
|
assert r.status_code == 401
|