mirror of
https://github.com/Katakate/k7.git
synced 2026-09-23 18:26:17 +00:00
G
13fafe0997
Per-key node pins, k7d-fc pause/resume/exec, and HA-soak fixes. Playbook pins k7d 0.7.0. GitHub .deb, Launchpad PPA, and PyPI k7-sdk are 0.4.0.
610 lines
26 KiB
Python
610 lines
26 KiB
Python
"""Unit tests for the k7d backend paths in K7Core.
|
|
|
|
Everything here is mocked — the daemon socket, crictl, and the k8s API
|
|
are never touched. The real end-to-end behaviour is covered by
|
|
``tests/integration/test_k7d.py`` on a k7d-capable node.
|
|
"""
|
|
|
|
from contextlib import contextmanager
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from k7.core.core import (
|
|
K7_TENANT_LABEL,
|
|
K7D_ANN_FORK_SOURCE_CLUSTER,
|
|
K7D_ANN_FORK_SOURCE_VM,
|
|
K7D_ANN_PAUSED,
|
|
K7D_FC_VIRTIOFS_REFUSED,
|
|
K7Core,
|
|
)
|
|
from k7.core.models import OperationResult, SandboxConfig
|
|
from tests.unit.conftest import mock_deployment, mock_pod
|
|
|
|
|
|
@contextmanager
|
|
def _local_node(core: K7Core, node: str = "k7-node-01", pod: str = "src-sb-abc"):
|
|
"""Sandbox appears LOCAL — direct daemon-socket path, not agent forward."""
|
|
with (
|
|
patch.object(core, "_k7d_local_node", return_value=node),
|
|
patch.object(core, "_k7d_sandbox_node", new=AsyncMock(return_value=node)),
|
|
patch.object(core, "_k7d_sandbox_placement", new=AsyncMock(return_value=(node, pod))),
|
|
):
|
|
yield
|
|
|
|
|
|
def _k7d_deployment(name: str = "src-sb", sidecar: str | None = None):
|
|
annotations = {"k7.katakate.org/backend": "k7d"}
|
|
if sidecar:
|
|
annotations["k7.katakate.org/sidecar"] = sidecar
|
|
dep = mock_deployment(name=name, runtime_class="k7", annotations=annotations)
|
|
dep.spec.template.metadata.annotations = dict(annotations)
|
|
dep.spec.template.metadata.labels = {"app": name, "katakate.org/sandbox": name}
|
|
dep.metadata.labels = {"app": name, "runtime": "kata", "katakate.org/sandbox": name}
|
|
dep.spec.selector.match_labels = {"app": name}
|
|
return dep
|
|
|
|
|
|
# --- backend canonicalization / detection ---
|
|
|
|
|
|
class TestK7dBackendDetection:
|
|
def test_k7_alias_canonicalizes_to_k7d(self):
|
|
assert K7Core._canonicalize_backend("k7") == "k7d"
|
|
|
|
def test_k7d_passes_through(self):
|
|
assert K7Core._canonicalize_backend("k7d") == "k7d"
|
|
|
|
async def test_detect_backend_accepts_k7d_annotation(self, core: K7Core):
|
|
dep = mock_deployment(annotations={"k7.katakate.org/backend": "k7d"})
|
|
mock_apps = AsyncMock()
|
|
mock_apps.read_namespaced_deployment.return_value = dep
|
|
core._apps_v1_client = mock_apps
|
|
|
|
assert await core._detect_backend("test-sb", "default") == "k7d"
|
|
|
|
def test_k7_fc_alias_canonicalizes_to_k7d_fc(self):
|
|
assert K7Core._canonicalize_backend("k7-fc") == "k7d-fc"
|
|
|
|
def test_k7d_fc_passes_through(self):
|
|
assert K7Core._canonicalize_backend("k7d-fc") == "k7d-fc"
|
|
|
|
async def test_detect_backend_accepts_k7d_fc_annotation(self, core: K7Core):
|
|
dep = mock_deployment(annotations={"k7.katakate.org/backend": "k7d-fc"})
|
|
mock_apps = AsyncMock()
|
|
mock_apps.read_namespaced_deployment.return_value = dep
|
|
core._apps_v1_client = mock_apps
|
|
|
|
assert await core._detect_backend("test-sb", "default") == "k7d-fc"
|
|
|
|
|
|
# --- create_sandbox rendering ---
|
|
|
|
|
|
class TestK7dCreateSandbox:
|
|
async def test_create_uses_runtime_class_k7_and_no_pvc(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_v1 = AsyncMock()
|
|
mock_net = AsyncMock()
|
|
core._apps_v1_client = mock_apps
|
|
core._core_v1_client = mock_v1
|
|
core._networking_v1_client = mock_net
|
|
|
|
sched_ok = MagicMock()
|
|
sched_ok.success = True
|
|
with patch.object(core, "_check_scheduling", new=AsyncMock(return_value=sched_ok)):
|
|
cfg = SandboxConfig(name="k7d-sb", image="alpine:3.20", backend="k7d")
|
|
result = await core.create_sandbox(cfg)
|
|
|
|
assert result.success, result.error
|
|
deployment = mock_apps.create_namespaced_deployment.call_args.kwargs["body"]
|
|
pod_spec = deployment.spec.template.spec
|
|
assert pod_spec.runtime_class_name == "k7"
|
|
assert pod_spec.node_selector == {"k7.katakate.org/backend-k7d": "true"}
|
|
# No Longhorn PVC, no persist-wrapper ConfigMap, no init container.
|
|
mock_v1.create_namespaced_persistent_volume_claim.assert_not_called()
|
|
mock_v1.create_namespaced_config_map.assert_not_called()
|
|
assert not pod_spec.init_containers
|
|
# Deployment carries the backend annotation for later detection.
|
|
assert deployment.metadata.annotations["k7.katakate.org/backend"] == "k7d"
|
|
|
|
async def test_node_name_uses_hostname_selector_not_spec_node_name(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_v1 = AsyncMock()
|
|
mock_net = AsyncMock()
|
|
node = MagicMock()
|
|
node.spec.taints = []
|
|
mock_v1.read_node.return_value = node
|
|
core._apps_v1_client = mock_apps
|
|
core._core_v1_client = mock_v1
|
|
core._networking_v1_client = mock_net
|
|
|
|
sched_ok = MagicMock()
|
|
sched_ok.success = True
|
|
with patch.object(core, "_check_scheduling", new=AsyncMock(return_value=sched_ok)):
|
|
cfg = SandboxConfig(name="k7d-pin", image="alpine:3.20", backend="k7d", node_name="k7-node-01")
|
|
result = await core.create_sandbox(cfg)
|
|
|
|
assert result.success, result.error
|
|
pod_spec = mock_apps.create_namespaced_deployment.call_args.kwargs["body"].spec.template.spec
|
|
assert pod_spec.node_name in (None, "")
|
|
assert pod_spec.node_selector == {
|
|
"k7.katakate.org/backend-k7d": "true",
|
|
"kubernetes.io/hostname": "k7-node-01",
|
|
}
|
|
assert not pod_spec.tolerations
|
|
mock_v1.read_node.assert_awaited_once_with("k7-node-01")
|
|
|
|
async def test_dedicated_node_adds_tenant_toleration(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_v1 = AsyncMock()
|
|
mock_net = AsyncMock()
|
|
taint = MagicMock()
|
|
taint.key = "k7.katakate.org/tenant"
|
|
taint.value = "acme"
|
|
taint.effect = "NoSchedule"
|
|
node = MagicMock()
|
|
node.spec.taints = [taint]
|
|
mock_v1.read_node.return_value = node
|
|
core._apps_v1_client = mock_apps
|
|
core._core_v1_client = mock_v1
|
|
core._networking_v1_client = mock_net
|
|
|
|
sched_ok = MagicMock()
|
|
sched_ok.success = True
|
|
with patch.object(core, "_check_scheduling", new=AsyncMock(return_value=sched_ok)):
|
|
cfg = SandboxConfig(name="k7d-ten", image="alpine:3.20", backend="k7d", node_name="k7-node-01")
|
|
result = await core.create_sandbox(cfg)
|
|
|
|
assert result.success, result.error
|
|
pod_spec = mock_apps.create_namespaced_deployment.call_args.kwargs["body"].spec.template.spec
|
|
assert pod_spec.tolerations
|
|
tol = pod_spec.tolerations[0]
|
|
assert tol.key == "k7.katakate.org/tenant"
|
|
assert tol.value == "acme"
|
|
assert tol.effect == "NoSchedule"
|
|
|
|
async def test_missing_pin_node_fails_loud(self, core: K7Core):
|
|
from kubernetes_asyncio.client.exceptions import ApiException
|
|
|
|
mock_v1 = AsyncMock()
|
|
mock_v1.read_node.side_effect = ApiException(status=404)
|
|
core._apps_v1_client = AsyncMock()
|
|
core._core_v1_client = mock_v1
|
|
core._networking_v1_client = AsyncMock()
|
|
cfg = SandboxConfig(name="k7d-gone", image="alpine:3.20", backend="k7d", node_name="no-such-node")
|
|
result = await core.create_sandbox(cfg)
|
|
assert not result.success
|
|
assert "no-such-node" in (result.error or "")
|
|
core._apps_v1_client.create_namespaced_deployment.assert_not_called()
|
|
|
|
async def test_create_k7d_fc_uses_runtime_class_k7_fc(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_v1 = AsyncMock()
|
|
mock_net = AsyncMock()
|
|
core._apps_v1_client = mock_apps
|
|
core._core_v1_client = mock_v1
|
|
core._networking_v1_client = mock_net
|
|
|
|
sched_ok = MagicMock()
|
|
sched_ok.success = True
|
|
with patch.object(core, "_check_scheduling", new=AsyncMock(return_value=sched_ok)):
|
|
cfg = SandboxConfig(name="k7d-fc-sb", image="alpine:3.20", backend="k7d-fc")
|
|
result = await core.create_sandbox(cfg)
|
|
|
|
assert result.success, result.error
|
|
deployment = mock_apps.create_namespaced_deployment.call_args.kwargs["body"]
|
|
pod_spec = deployment.spec.template.spec
|
|
assert pod_spec.runtime_class_name == "k7-fc"
|
|
assert pod_spec.node_selector == {"k7.katakate.org/backend-k7d-fc": "true"}
|
|
assert deployment.metadata.annotations["k7.katakate.org/backend"] == "k7d-fc"
|
|
|
|
async def test_create_k7d_fc_refuses_hostpath_before_pod(self, core: K7Core):
|
|
core._apps_v1_client = AsyncMock()
|
|
core._core_v1_client = AsyncMock()
|
|
core._networking_v1_client = AsyncMock()
|
|
cfg = SandboxConfig(
|
|
name="k7d-fc-hp",
|
|
image="alpine:3.20",
|
|
backend="k7d-fc",
|
|
volumes=[{"name": "hp", "hostPath": {"path": "/var/log"}}],
|
|
)
|
|
result = await core.create_sandbox(cfg)
|
|
assert not result.success
|
|
assert result.error == K7D_FC_VIRTIOFS_REFUSED
|
|
core._apps_v1_client.create_namespaced_deployment.assert_not_called()
|
|
|
|
async def test_docker_stamps_annotations_on_k7d_fc(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
core._apps_v1_client = mock_apps
|
|
core._core_v1_client = AsyncMock()
|
|
core._networking_v1_client = AsyncMock()
|
|
|
|
sched_ok = MagicMock()
|
|
sched_ok.success = True
|
|
with (
|
|
patch.object(core, "_check_scheduling", new=AsyncMock(return_value=sched_ok)),
|
|
patch("k7.core.core.k7d_supports_docker", return_value=True),
|
|
):
|
|
cfg = SandboxConfig(name="k7d-fc-dock", image="ubuntu:24.04", backend="k7-fc", docker=True)
|
|
result = await core.create_sandbox(cfg)
|
|
|
|
assert result.success, result.error
|
|
deployment = mock_apps.create_namespaced_deployment.call_args.kwargs["body"]
|
|
assert deployment.spec.template.spec.runtime_class_name == "k7-fc"
|
|
assert deployment.spec.template.metadata.annotations["k7d.katakate.org/docker"] == "true"
|
|
|
|
async def test_create_skips_kata_memory_annotation(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
core._apps_v1_client = mock_apps
|
|
core._core_v1_client = AsyncMock()
|
|
core._networking_v1_client = AsyncMock()
|
|
|
|
sched_ok = MagicMock()
|
|
sched_ok.success = True
|
|
with patch.object(core, "_check_scheduling", new=AsyncMock(return_value=sched_ok)):
|
|
cfg = SandboxConfig(
|
|
name="k7d-mem", image="alpine:3.20", backend="k7d", limits={"memory": "512Mi", "cpu": "1"}
|
|
)
|
|
result = await core.create_sandbox(cfg)
|
|
|
|
assert result.success, result.error
|
|
deployment = mock_apps.create_namespaced_deployment.call_args.kwargs["body"]
|
|
annotations = deployment.spec.template.metadata.annotations
|
|
assert "io.katacontainers.config.hypervisor.default_memory" not in annotations
|
|
# Pod resources still flow through so the k7d shim sizes the VM.
|
|
container = deployment.spec.template.spec.containers[0]
|
|
assert container.resources.limits == {"memory": "512Mi", "cpu": "1"}
|
|
|
|
async def test_docker_stamps_annotations_no_sidecar_no_privileged(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
core._apps_v1_client = mock_apps
|
|
core._core_v1_client = AsyncMock()
|
|
core._networking_v1_client = AsyncMock()
|
|
|
|
sched_ok = MagicMock()
|
|
sched_ok.success = True
|
|
with (
|
|
patch.object(core, "_check_scheduling", new=AsyncMock(return_value=sched_ok)),
|
|
patch("k7.core.core.k7d_supports_docker", return_value=True),
|
|
):
|
|
cfg = SandboxConfig(
|
|
name="k7d-dock",
|
|
image="ubuntu:24.04",
|
|
backend="k7d",
|
|
docker=True,
|
|
docker_disk="40Gi",
|
|
)
|
|
result = await core.create_sandbox(cfg)
|
|
|
|
assert result.success, result.error
|
|
deployment = mock_apps.create_namespaced_deployment.call_args.kwargs["body"]
|
|
ann = deployment.metadata.annotations
|
|
pod_ann = deployment.spec.template.metadata.annotations
|
|
assert ann["k7.katakate.org/docker"] == "true"
|
|
assert pod_ann["k7d.katakate.org/docker"] == "true"
|
|
assert pod_ann["k7d.katakate.org/docker-disk"] == "40Gi"
|
|
assert "k7.katakate.org/sidecar" not in ann
|
|
assert "k7.katakate.org/sidecar" not in pod_ann
|
|
containers = deployment.spec.template.spec.containers
|
|
assert [c.name for c in containers] == ["sandbox"]
|
|
sandbox = containers[0]
|
|
assert sandbox.security_context.privileged is not True
|
|
assert sandbox.security_context.allow_privilege_escalation is False
|
|
assert sandbox.security_context.capabilities.drop == ["ALL"]
|
|
assert sandbox.security_context.seccomp_profile.type == "RuntimeDefault"
|
|
|
|
async def test_docker_against_old_k7d_fails_loud(self, core: K7Core):
|
|
core._apps_v1_client = AsyncMock()
|
|
core._core_v1_client = AsyncMock()
|
|
core._networking_v1_client = AsyncMock()
|
|
with patch("k7.core.core.k7d_supports_docker", return_value=False):
|
|
cfg = SandboxConfig(name="k7d-old", image="ubuntu:24.04", backend="k7d", docker=True)
|
|
result = await core.create_sandbox(cfg)
|
|
assert not result.success
|
|
assert "no docker service" in result.error
|
|
assert "upgrade" in result.error
|
|
|
|
async def test_docker_and_sidecar_mutually_exclusive(self, core: K7Core):
|
|
core._apps_v1_client = AsyncMock()
|
|
core._core_v1_client = AsyncMock()
|
|
core._networking_v1_client = AsyncMock()
|
|
cfg = SandboxConfig(
|
|
name="both",
|
|
image="ubuntu:24.04",
|
|
backend="k7d",
|
|
docker=True,
|
|
sidecar="docker",
|
|
)
|
|
result = await core.create_sandbox(cfg)
|
|
assert not result.success
|
|
assert "mutually exclusive" in result.error
|
|
|
|
|
|
# --- pause / resume via the daemon socket ---
|
|
|
|
|
|
class TestK7dPauseResume:
|
|
async def test_pause_freezes_vm_in_place(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_apps.read_namespaced_deployment.return_value = _k7d_deployment("k7d-sb")
|
|
core._apps_v1_client = mock_apps
|
|
|
|
vm = {"vm_id": "vm-1-0", "guest_cid": 77, "sandbox_id": "cri123", "cluster_id": "cri123"}
|
|
with (
|
|
_local_node(core),
|
|
patch.object(core, "_k7d_vm_for_sandbox", new=AsyncMock(return_value=vm)) as lookup,
|
|
patch.object(core, "_k7d_request", new=AsyncMock(return_value={"status": "ok"})) as request,
|
|
):
|
|
result = await core.pause_sandbox("k7d-sb")
|
|
|
|
assert result.success, result.error
|
|
lookup.assert_awaited_once_with("k7d-sb", "default")
|
|
request.assert_awaited_once_with({"op": "pause_vm", "vm_id": "vm-1-0"})
|
|
# The deployment is NOT scaled down — the VM is frozen in place.
|
|
mock_apps.patch_namespaced_deployment_scale.assert_not_called()
|
|
# Paused marker annotation stamped for list/resume.
|
|
patch_body = mock_apps.patch_namespaced_deployment.call_args.kwargs["body"]
|
|
assert patch_body["metadata"]["annotations"][K7D_ANN_PAUSED] == "true"
|
|
|
|
async def test_pause_k7d_fc_uses_daemon_socket(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
dep = mock_deployment(name="fc-sb", runtime_class="k7-fc", annotations={"k7.katakate.org/backend": "k7d-fc"})
|
|
mock_apps.read_namespaced_deployment.return_value = dep
|
|
core._apps_v1_client = mock_apps
|
|
vm = {"vm_id": "vm-fc-0", "guest_cid": 88, "sandbox_id": "cri-fc"}
|
|
with (
|
|
_local_node(core),
|
|
patch.object(core, "_k7d_vm_for_sandbox", new=AsyncMock(return_value=vm)),
|
|
patch.object(core, "_k7d_request", new=AsyncMock(return_value={"status": "ok"})) as request,
|
|
):
|
|
result = await core.pause_sandbox("fc-sb")
|
|
assert result.success, result.error
|
|
request.assert_awaited_once_with({"op": "pause_vm", "vm_id": "vm-fc-0"})
|
|
mock_apps.patch_namespaced_deployment_scale.assert_not_called()
|
|
|
|
async def test_pause_with_snapshot_is_loud_error(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_apps.read_namespaced_deployment.return_value = _k7d_deployment("k7d-sb")
|
|
core._apps_v1_client = mock_apps
|
|
|
|
result = await core.pause_sandbox("k7d-sb", snapshot_name="snap1")
|
|
assert not result.success
|
|
assert "k7d" in result.error
|
|
|
|
async def test_resume_unfreezes_vm(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_apps.read_namespaced_deployment.return_value = _k7d_deployment("k7d-sb")
|
|
core._apps_v1_client = mock_apps
|
|
|
|
vm = {"vm_id": "vm-1-0", "guest_cid": 77, "sandbox_id": "cri123"}
|
|
with (
|
|
_local_node(core),
|
|
patch.object(core, "_k7d_vm_for_sandbox", new=AsyncMock(return_value=vm)),
|
|
patch.object(core, "_k7d_request", new=AsyncMock(return_value={"status": "ok"})) as request,
|
|
):
|
|
result = await core.resume_sandbox("k7d-sb")
|
|
|
|
assert result.success, result.error
|
|
request.assert_awaited_once_with({"op": "resume_vm", "vm_id": "vm-1-0"})
|
|
mock_apps.patch_namespaced_deployment_scale.assert_not_called()
|
|
|
|
async def test_pause_daemon_error_is_loud(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_apps.read_namespaced_deployment.return_value = _k7d_deployment("k7d-sb")
|
|
core._apps_v1_client = mock_apps
|
|
|
|
with (
|
|
_local_node(core),
|
|
patch.object(
|
|
core,
|
|
"_k7d_vm_for_sandbox",
|
|
new=AsyncMock(side_effect=RuntimeError("k7d control socket /run/k7d/k7d.sock not found")),
|
|
),
|
|
):
|
|
result = await core.pause_sandbox("k7d-sb")
|
|
assert not result.success
|
|
assert "k7d.sock" in result.error
|
|
|
|
|
|
# --- fork via the shim's fork-source annotations ---
|
|
|
|
|
|
class TestK7dFork:
|
|
async def test_fork_creates_annotated_deployment(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_apps.read_namespaced_deployment.return_value = _k7d_deployment("src-sb")
|
|
core._apps_v1_client = mock_apps
|
|
|
|
vm = {"vm_id": "vm-1-0", "guest_cid": 77, "sandbox_id": "cri-src", "cluster_id": "cri-src"}
|
|
src_pod = mock_pod(name="src-sb-abc")
|
|
src_pod.spec.node_name = "k7-node-01"
|
|
with (
|
|
_local_node(core),
|
|
patch.object(core, "_k7d_vm_for_sandbox", new=AsyncMock(return_value=vm)),
|
|
patch.object(core, "_k7d_running_pod", new=AsyncMock(return_value=src_pod)),
|
|
patch.object(core, "_wait_for_pod_container_started", new=AsyncMock(return_value="dst-sb-xyz")),
|
|
patch.object(
|
|
core, "_read_sandbox_egress_whitelist", new=AsyncMock(return_value=["example.com"])
|
|
) as read_egress,
|
|
patch.object(core, "_read_sandbox_ingress_rules", new=AsyncMock(return_value=([8000], ["sandbox:a"]))),
|
|
patch.object(
|
|
core, "_apply_sandbox_network_policies", new=AsyncMock(return_value=OperationResult(True))
|
|
) as apply_policies,
|
|
):
|
|
result = await core.fork_sandbox("src-sb", "dst-sb")
|
|
|
|
assert result.success, result.error
|
|
# The warm fork inherits the source's policy too.
|
|
read_egress.assert_awaited_once_with("src-sb", "default")
|
|
assert apply_policies.call_args.kwargs["egress_whitelist"] == ["example.com"]
|
|
assert apply_policies.call_args.kwargs["ingress_ports"] == [8000]
|
|
assert apply_policies.call_args.kwargs["ingress_from"] == ["sandbox:a"]
|
|
assert "warm fork" in result.message
|
|
new_dep = mock_apps.create_namespaced_deployment.call_args.kwargs["body"]
|
|
assert new_dep.metadata.name == "dst-sb"
|
|
annotations = new_dep.spec.template.metadata.annotations
|
|
assert annotations[K7D_ANN_FORK_SOURCE_CLUSTER] == "cri-src"
|
|
assert annotations[K7D_ANN_FORK_SOURCE_VM] == "cri-src"
|
|
# Pinned to the source VM's node — the daemon socket is node-local.
|
|
assert new_dep.spec.template.spec.node_name == "k7-node-01"
|
|
# No Longhorn snapshot machinery involved.
|
|
assert new_dep.spec.template.metadata.labels["app"] == "dst-sb"
|
|
|
|
async def test_fork_with_sidecar_is_loud_error(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_apps.read_namespaced_deployment.return_value = _k7d_deployment("src-sb", sidecar="docker")
|
|
core._apps_v1_client = mock_apps
|
|
|
|
result = await core.fork_sandbox("src-sb", "dst-sb")
|
|
assert not result.success
|
|
assert "sidecar" in result.error
|
|
assert "real CRI sidecar" in result.error
|
|
mock_apps.create_namespaced_deployment.assert_not_called()
|
|
|
|
async def test_fork_docker_annotation_alone_is_not_a_sidecar(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
annotations = {
|
|
"k7.katakate.org/backend": "k7d",
|
|
"k7.katakate.org/docker": "true",
|
|
"k7d.katakate.org/docker": "true",
|
|
}
|
|
dep = mock_deployment(name="src-sb", runtime_class="k7", annotations=annotations)
|
|
dep.spec.template.metadata.annotations = dict(annotations)
|
|
dep.spec.template.metadata.labels = {"app": "src-sb", "katakate.org/sandbox": "src-sb"}
|
|
dep.metadata.labels = {"app": "src-sb", "runtime": "kata", "katakate.org/sandbox": "src-sb"}
|
|
dep.spec.selector.match_labels = {"app": "src-sb"}
|
|
mock_apps.read_namespaced_deployment.return_value = dep
|
|
core._apps_v1_client = mock_apps
|
|
|
|
vm = {"vm_id": "vm-1-0", "guest_cid": 77, "sandbox_id": "cri-src", "cluster_id": "cri-src"}
|
|
src_pod = mock_pod(name="src-sb-abc")
|
|
src_pod.spec.node_name = "k7-node-01"
|
|
with (
|
|
_local_node(core),
|
|
patch.object(core, "_k7d_vm_for_sandbox", new=AsyncMock(return_value=vm)),
|
|
patch.object(core, "_k7d_running_pod", new=AsyncMock(return_value=src_pod)),
|
|
patch.object(core, "_wait_for_pod_container_started", new=AsyncMock(return_value="dst-sb-xyz")),
|
|
patch.object(core, "_read_sandbox_egress_whitelist", new=AsyncMock(return_value=[])),
|
|
patch.object(core, "_read_sandbox_ingress_rules", new=AsyncMock(return_value=([], []))),
|
|
patch.object(core, "_apply_sandbox_network_policies", new=AsyncMock(return_value=OperationResult(True))),
|
|
):
|
|
result = await core.fork_sandbox("src-sb", "dst-sb")
|
|
|
|
assert result.success, result.error
|
|
mock_apps.create_namespaced_deployment.assert_called_once()
|
|
|
|
async def test_fork_with_snapshot_name_is_loud_error(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_apps.read_namespaced_deployment.return_value = _k7d_deployment("src-sb")
|
|
core._apps_v1_client = mock_apps
|
|
|
|
result = await core.fork_sandbox("src-sb", "dst-sb", snapshot_name="keep-me")
|
|
assert not result.success
|
|
assert "k7d" in result.error
|
|
|
|
async def test_fork_dead_source_is_loud_error(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_apps.read_namespaced_deployment.return_value = _k7d_deployment("src-sb")
|
|
core._apps_v1_client = mock_apps
|
|
|
|
with (
|
|
_local_node(core),
|
|
patch.object(
|
|
core,
|
|
"_k7d_vm_for_sandbox",
|
|
new=AsyncMock(side_effect=RuntimeError("no Running pod found for sandbox src-sb")),
|
|
),
|
|
):
|
|
result = await core.fork_sandbox("src-sb", "dst-sb")
|
|
assert not result.success
|
|
assert "no Running pod" in result.error
|
|
mock_apps.create_namespaced_deployment.assert_not_called()
|
|
|
|
|
|
# --- named snapshots are rejected loudly ---
|
|
|
|
|
|
class TestK7dSnapshotRejected:
|
|
async def test_create_snapshot_rejected(self, core: K7Core):
|
|
mock_apps = AsyncMock()
|
|
mock_apps.read_namespaced_deployment.return_value = _k7d_deployment("k7d-sb")
|
|
core._apps_v1_client = mock_apps
|
|
|
|
result = await core.create_snapshot("k7d-sb", "snap1")
|
|
assert not result.success
|
|
assert "k7d" in result.error
|
|
assert "katakate/k7d" in result.error
|
|
|
|
|
|
# --- daemon socket helper ---
|
|
|
|
|
|
class TestK7dRequest:
|
|
async def test_missing_socket_is_loud(self, core: K7Core):
|
|
with (
|
|
patch.dict("os.environ", {"K7D_SOCKET": "/nonexistent/k7d.sock"}),
|
|
pytest.raises(RuntimeError, match="control socket"),
|
|
):
|
|
await core._k7d_request({"op": "ping"})
|
|
|
|
async def test_error_response_raises(self, core: K7Core):
|
|
async def fake_thread(fn):
|
|
return {"status": "error", "message": "no such VM: vm-9"}
|
|
|
|
with (
|
|
patch("k7.core.core.asyncio.to_thread", new=fake_thread),
|
|
patch("k7.core.core.os.path.exists", return_value=True),
|
|
pytest.raises(RuntimeError, match="no such VM"),
|
|
):
|
|
await core._k7d_request({"op": "pause_vm", "vm_id": "vm-9"})
|
|
|
|
|
|
class TestDedicateNode:
|
|
async def test_dedicate_writes_label_and_taint(self, core: K7Core):
|
|
mock_v1 = AsyncMock()
|
|
node = MagicMock()
|
|
node.spec.taints = []
|
|
mock_v1.read_node.return_value = node
|
|
core._core_v1_client = mock_v1
|
|
|
|
result = await core.dedicate_node("k7-node-01", "acme")
|
|
assert result.success, result.error
|
|
body = mock_v1.patch_node.await_args.args[1]
|
|
assert body["metadata"]["labels"][K7_TENANT_LABEL] == "acme"
|
|
assert body["spec"]["taints"] == [{"key": K7_TENANT_LABEL, "value": "acme", "effect": "NoSchedule"}]
|
|
|
|
async def test_dedicate_missing_node_fails_loud(self, core: K7Core):
|
|
from kubernetes_asyncio.client.exceptions import ApiException
|
|
|
|
mock_v1 = AsyncMock()
|
|
mock_v1.read_node.side_effect = ApiException(status=404)
|
|
core._core_v1_client = mock_v1
|
|
result = await core.dedicate_node("missing", "acme")
|
|
assert not result.success
|
|
assert "not found" in (result.error or "")
|
|
mock_v1.patch_node.assert_not_called()
|
|
|
|
async def test_undedicate_drops_label_and_taint(self, core: K7Core):
|
|
mock_v1 = AsyncMock()
|
|
other = MagicMock()
|
|
other.key = "node.kubernetes.io/unreachable"
|
|
other.value = None
|
|
other.effect = "NoExecute"
|
|
ours = MagicMock()
|
|
ours.key = K7_TENANT_LABEL
|
|
ours.value = "acme"
|
|
ours.effect = "NoSchedule"
|
|
node = MagicMock()
|
|
node.spec.taints = [other, ours]
|
|
mock_v1.read_node.return_value = node
|
|
core._core_v1_client = mock_v1
|
|
|
|
result = await core.undedicate_node("k7-node-01")
|
|
assert result.success, result.error
|
|
body = mock_v1.patch_node.await_args.args[1]
|
|
assert body["metadata"]["labels"][K7_TENANT_LABEL] is None
|
|
assert body["spec"]["taints"] == [
|
|
{"key": "node.kubernetes.io/unreachable", "value": None, "effect": "NoExecute"}
|
|
]
|