"""Mock-based unit tests for K7Core lifecycle methods (no real k8s needed).""" from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch from kubernetes_asyncio.client.exceptions import ApiException from k7.core.core import K7_TENANT_LABEL, K7Core from k7.core.models import ExecResult, OperationResult, SandboxConfig from tests.unit.conftest import mock_deployment, mock_pod # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _setup_clients(core, apps=None, v1=None, net=None, metrics=None, custom=None, batch=None, apiext=None): """Wire MagicMock clients into a K7Core instance.""" core._config_loaded = True if apps is not None: core._apps_v1_client = apps if v1 is not None: core._core_v1_client = v1 if net is not None: core._networking_v1_client = net if metrics is not None: core._metrics_client = metrics if custom is not None: core._custom_objects_client = custom if batch is not None: core._batch_v1_client = batch if apiext is not None: core._apiextensions_v1_client = apiext # --- _detect_backend --- class TestDetectBackend: async def test_annotation_takes_priority(self, core: K7Core): dep = mock_deployment(annotations={"k7.katakate.org/backend": "kata-qemu-longhorn"}) mock_apps = AsyncMock() mock_apps.read_namespaced_deployment.return_value = dep core._apps_v1_client = mock_apps core._config_loaded = True assert await core._detect_backend("test-sb", "default") == "kata-qemu-longhorn" async def test_host_file_when_no_sandbox(self, core: K7Core, tmp_path): backend_file = tmp_path / "backend" backend_file.write_text("kata-qemu-longhorn\n") with patch("os.path.exists", return_value=True), patch("builtins.open", return_value=open(backend_file)): assert await core._detect_backend(None) == "kata-qemu-longhorn" async def test_default_when_nothing_found(self, core: K7Core): with patch("os.path.exists", return_value=False): assert await core._detect_backend(None) == "kata-firecracker-devmapper" async def test_default_when_annotation_missing(self, core: K7Core): dep = mock_deployment(annotations={}) mock_apps = AsyncMock() mock_apps.read_namespaced_deployment.return_value = dep core._apps_v1_client = mock_apps core._config_loaded = True with patch("os.path.exists", return_value=False): assert await core._detect_backend("test-sb", "default") == "kata-firecracker-devmapper" async def test_annotation_api_error_falls_through(self, core: K7Core): mock_apps = AsyncMock() mock_apps.read_namespaced_deployment.side_effect = ApiException(status=404) core._apps_v1_client = mock_apps core._config_loaded = True with patch("os.path.exists", return_value=False): assert await core._detect_backend("missing", "default") == "kata-firecracker-devmapper" async def test_none_file_is_not_kfd(self, core: K7Core, tmp_path): backend_file = tmp_path / "backend" backend_file.write_text("none\n") with patch("os.path.exists", return_value=True), patch("builtins.open", return_value=open(backend_file)): assert await core._detect_backend(None) == "" # --- delete_sandbox --- class TestDeleteSandbox: async def test_successful_delete(self, core: K7Core): mock_apps = AsyncMock() mock_v1 = AsyncMock() mock_net = AsyncMock() mock_custom = AsyncMock() mock_custom.list_namespaced_custom_object.return_value = {"items": []} core._apps_v1_client = mock_apps core._core_v1_client = mock_v1 core._networking_v1_client = mock_net core._custom_objects_client = mock_custom core._config_loaded = True result = await core.delete_sandbox("my-sb", "default") assert result.success is True mock_apps.delete_namespaced_deployment.assert_called_once_with(name="my-sb", namespace="default") mock_v1.delete_namespaced_secret.assert_called_once_with(name="my-sb-env", namespace="default") mock_v1.delete_namespaced_config_map.assert_any_call(name="my-sb-persist-wrapper", namespace="default") mock_v1.delete_namespaced_config_map.assert_any_call(name="my-sb-docker-vehicle", namespace="default") mock_v1.delete_namespaced_persistent_volume_claim.assert_any_call(name="my-sb-root-lh", namespace="default") mock_v1.delete_namespaced_persistent_volume_claim.assert_any_call(name="my-sb-docker-lh", namespace="default") async def test_delete_ignores_404s(self, core: K7Core): mock_apps = AsyncMock() mock_apps.delete_namespaced_deployment.side_effect = ApiException(status=404) mock_v1 = AsyncMock() mock_v1.delete_namespaced_secret.side_effect = ApiException(status=404) mock_v1.delete_namespaced_config_map.side_effect = ApiException(status=404) mock_v1.delete_namespaced_persistent_volume_claim.side_effect = ApiException(status=404) mock_v1.patch_namespaced_persistent_volume_claim.side_effect = ApiException(status=404) mock_net = AsyncMock() mock_net.delete_namespaced_network_policy.side_effect = ApiException(status=404) mock_custom = AsyncMock() mock_custom.list_namespaced_custom_object.return_value = {"items": []} core._apps_v1_client = mock_apps core._core_v1_client = mock_v1 core._networking_v1_client = mock_net core._custom_objects_client = mock_custom core._config_loaded = True result = await core.delete_sandbox("gone-sb", "default") assert result.success is True async def test_delete_reports_non_404_errors(self, core: K7Core): mock_apps = AsyncMock() mock_apps.delete_namespaced_deployment.side_effect = ApiException(status=500, reason="Server Error") mock_v1 = AsyncMock() mock_net = AsyncMock() mock_custom = AsyncMock() mock_custom.list_namespaced_custom_object.return_value = {"items": []} core._apps_v1_client = mock_apps core._core_v1_client = mock_v1 core._networking_v1_client = mock_net core._custom_objects_client = mock_custom core._config_loaded = True result = await core.delete_sandbox("err-sb", "default") assert result.success is False assert "deployment" in result.error # --- list_sandboxes --- class TestListSandboxes: async def test_returns_sandbox_info_list(self, core: K7Core): dep = mock_deployment(name="sb1", annotations={"k7.katakate.org/backend": "kata-firecracker-devmapper"}) pod = mock_pod(name="sb1-abc", image="ubuntu:22.04") mock_apps = AsyncMock() core._apps_v1_client = mock_apps core._config_loaded = True mock_v1 = AsyncMock() mock_v1.list_namespaced_pod.return_value = MagicMock(items=[pod]) core._core_v1_client = mock_v1 with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, return_value=[dep]): result = await core.list_sandboxes(namespace="default") assert len(result) == 1 assert result[0].name == "sb1" assert result[0].image == "ubuntu:22.04" assert result[0].backend == "kata-firecracker-devmapper" assert result[0].ready == "True" async def test_returns_empty_on_no_sandboxes(self, core: K7Core): mock_apps = AsyncMock() core._apps_v1_client = mock_apps core._config_loaded = True with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, return_value=[]): result = await core.list_sandboxes(namespace="default") assert result == [] async def test_handles_pod_with_no_items(self, core: K7Core): dep = mock_deployment(name="sb-no-pod", annotations={"k7.katakate.org/backend": "kata-qemu-longhorn"}) mock_apps = AsyncMock() core._apps_v1_client = mock_apps core._config_loaded = True mock_v1 = AsyncMock() mock_v1.list_namespaced_pod.return_value = MagicMock(items=[]) core._core_v1_client = mock_v1 with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, return_value=[dep]): result = await core.list_sandboxes(namespace="default") assert len(result) == 1 assert result[0].status == "No Pods" assert result[0].ready == "False" async def test_exception_returns_empty_list(self, core: K7Core): core._config_loaded = True with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, side_effect=Exception("k8s down")): result = await core.list_sandboxes(namespace="default") assert result == [] # --- _get_kata_sandboxes --- class TestGetKataSandboxes: async def test_filters_by_runtime_class(self, core: K7Core): kata_dep = mock_deployment(name="kata-sb", runtime_class="kata") non_kata = mock_deployment(name="regular", runtime_class="runc") non_kata.metadata.labels = {} mock_apps = AsyncMock() mock_apps.list_namespaced_deployment.return_value = MagicMock(items=[kata_dep, non_kata]) core._apps_v1_client = mock_apps core._config_loaded = True result = await core._get_kata_sandboxes(namespace="default") assert len(result) == 1 assert result[0].metadata.name == "kata-sb" async def test_filters_by_label_fallback(self, core: K7Core): dep = mock_deployment(name="label-sb", runtime_class="runc", labels={"runtime": "kata"}) mock_apps = AsyncMock() mock_apps.list_namespaced_deployment.return_value = MagicMock(items=[dep]) core._apps_v1_client = mock_apps core._config_loaded = True result = await core._get_kata_sandboxes(namespace="default") assert len(result) == 1 async def test_all_namespaces_when_none(self, core: K7Core): mock_apps = AsyncMock() mock_apps.list_deployment_for_all_namespaces.return_value = MagicMock(items=[]) core._apps_v1_client = mock_apps core._config_loaded = True await core._get_kata_sandboxes(namespace=None) mock_apps.list_deployment_for_all_namespaces.assert_called_once() # --- _ensure_root_pvc --- class TestEnsureRootPvc: async def test_exists_with_filesystem_mode(self, core: K7Core): mock_v1 = AsyncMock() pvc = MagicMock() pvc.spec.volume_mode = "Filesystem" mock_v1.read_namespaced_persistent_volume_claim.return_value = pvc _setup_clients(core, v1=mock_v1) result = await core._ensure_root_pvc("sb1", "default", "10Gi") assert result.success is True assert result.data["created"] is False async def test_exists_with_wrong_volume_mode(self, core: K7Core): mock_v1 = AsyncMock() pvc = MagicMock() pvc.spec.volume_mode = "Block" mock_v1.read_namespaced_persistent_volume_claim.return_value = pvc _setup_clients(core, v1=mock_v1) result = await core._ensure_root_pvc("sb1", "default", "10Gi") assert result.success is False assert "Block" in result.error async def test_creates_when_not_found(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.read_namespaced_persistent_volume_claim.side_effect = ApiException(status=404) _setup_clients(core, v1=mock_v1) result = await core._ensure_root_pvc("sb1", "default", "10Gi") assert result.success is True assert result.data["created"] is True mock_v1.create_namespaced_persistent_volume_claim.assert_called_once() async def test_create_fails(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.read_namespaced_persistent_volume_claim.side_effect = ApiException(status=404) mock_v1.create_namespaced_persistent_volume_claim.side_effect = ApiException(status=500) _setup_clients(core, v1=mock_v1) result = await core._ensure_root_pvc("sb1", "default", "10Gi") assert result.success is False async def test_non_404_read_error(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.read_namespaced_persistent_volume_claim.side_effect = ApiException(status=500) _setup_clients(core, v1=mock_v1) result = await core._ensure_root_pvc("sb1", "default", "10Gi") assert result.success is False assert "lookup error" in result.error # --- _ensure_persist_wrapper_configmap --- class TestEnsurePersistWrapperConfigmap: async def test_exists(self, core: K7Core): mock_v1 = AsyncMock() _setup_clients(core, v1=mock_v1) result = await core._ensure_persist_wrapper_configmap("default", "sb1", "#!/bin/bash") assert result.success is True assert result.data["created"] is False async def test_creates_when_not_found(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.read_namespaced_config_map.side_effect = ApiException(status=404) _setup_clients(core, v1=mock_v1) result = await core._ensure_persist_wrapper_configmap("default", "sb1", "#!/bin/bash") assert result.success is True assert result.data["created"] is True mock_v1.create_namespaced_config_map.assert_called_once() async def test_non_404_read_error(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.read_namespaced_config_map.side_effect = ApiException(status=500) _setup_clients(core, v1=mock_v1) result = await core._ensure_persist_wrapper_configmap("default", "sb1", "#!/bin/bash") assert result.success is False assert "lookup error" in result.error async def test_create_fails(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.read_namespaced_config_map.side_effect = ApiException(status=404) mock_v1.create_namespaced_config_map.side_effect = ApiException(status=500) _setup_clients(core, v1=mock_v1) result = await core._ensure_persist_wrapper_configmap("default", "sb1", "#!/bin/bash") assert result.success is False # --- _update_deployment_annotation --- class TestUpdateDeploymentAnnotation: async def test_patches_deployment(self, core: K7Core): mock_apps = AsyncMock() _setup_clients(core, apps=mock_apps) await core._update_deployment_annotation("sb1", "default", "key", "val") mock_apps.patch_namespaced_deployment.assert_called_once_with( name="sb1", namespace="default", body={"metadata": {"annotations": {"key": "val"}}}, ) # --- pause_sandbox --- class TestPauseSandbox: async def test_scales_to_zero(self, core: K7Core): mock_apps = AsyncMock() _setup_clients(core, apps=mock_apps) result = await core.pause_sandbox("sb1") assert result.success is True assert "paused" in result.message mock_apps.patch_namespaced_deployment_scale.assert_called_once() async def test_with_snapshot_explicit_pvc(self, core: K7Core): mock_apps = AsyncMock() _setup_clients(core, apps=mock_apps) with patch.object( core, "_create_volume_snapshot", new_callable=AsyncMock, return_value=OperationResult(success=True, message="snap ok"), ) as snap_mock: result = await core.pause_sandbox("sb1", pvc_name="pvc1", snapshot_name="snap1") assert result.success is True assert "Snapshot snap1 created" in result.message snap_mock.assert_called_once() assert snap_mock.call_args.kwargs["pvc_name"] == "pvc1" assert snap_mock.call_args.kwargs["snapshot_name"] == "snap1" async def test_snapshot_only_defaults_pvc_to_root(self, core: K7Core): """Regression: snapshot_name alone triggers snapshot. The snapshot used to fire only when ``pvc_name AND snapshot_name`` were both set, so ``k7 pause foo --snapshot bar`` silently did nothing. Now ``snapshot_name`` alone is sufficient and ``pvc_name`` defaults to ``_root_pvc_name(name)``. """ mock_apps = AsyncMock() _setup_clients(core, apps=mock_apps) with patch.object( core, "_create_volume_snapshot", new_callable=AsyncMock, return_value=OperationResult(success=True, message="snap ok"), ) as snap_mock: result = await core.pause_sandbox("sb1", snapshot_name="snap1") assert result.success is True assert "Snapshot snap1 created" in result.message snap_mock.assert_called_once() assert snap_mock.call_args.kwargs["pvc_name"] == "sb1-root-lh" assert snap_mock.call_args.kwargs["snapshot_name"] == "snap1" async def test_no_snapshot_when_name_unset(self, core: K7Core): """Without ``snapshot_name`` we must NOT call _create_volume_snapshot.""" mock_apps = AsyncMock() _setup_clients(core, apps=mock_apps) with patch.object( core, "_create_volume_snapshot", new_callable=AsyncMock, ) as snap_mock: result = await core.pause_sandbox("sb1", pvc_name="ignored-without-name") assert result.success is True assert "Snapshot" not in result.message snap_mock.assert_not_called() async def test_snapshot_fails(self, core: K7Core): mock_apps = AsyncMock() _setup_clients(core, apps=mock_apps) with patch.object( core, "_create_volume_snapshot", new_callable=AsyncMock, return_value=OperationResult(success=False, error="snap err"), ): result = await core.pause_sandbox("sb1", snapshot_name="snap1") assert result.success is False assert "snap err" in result.error async def test_api_exception(self, core: K7Core): mock_apps = AsyncMock() mock_apps.patch_namespaced_deployment_scale.side_effect = ApiException(status=500) _setup_clients(core, apps=mock_apps) result = await core.pause_sandbox("sb1") assert result.success is False # --- resume_sandbox --- class TestResumeSandbox: async def test_scales_to_one(self, core: K7Core): mock_apps = AsyncMock() _setup_clients(core, apps=mock_apps) result = await core.resume_sandbox("sb1") assert result.success is True assert "resumed" in result.message async def test_api_exception(self, core: K7Core): mock_apps = AsyncMock() mock_apps.patch_namespaced_deployment_scale.side_effect = ApiException(status=500) _setup_clients(core, apps=mock_apps) result = await core.resume_sandbox("sb1") assert result.success is False # --- delete_all_sandboxes --- class TestDeleteAllSandboxes: async def test_deletes_multiple(self, core: K7Core): mock_apps = AsyncMock() mock_v1 = AsyncMock() mock_net = AsyncMock() mock_custom = AsyncMock() mock_custom.list_namespaced_custom_object.return_value = {"items": []} _setup_clients(core, apps=mock_apps, v1=mock_v1, net=mock_net, custom=mock_custom) deps = [mock_deployment(name="a"), mock_deployment(name="b")] with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, return_value=deps): result = await core.delete_all_sandboxes() assert result.success is True assert "2" in result.message async def test_one_fails(self, core: K7Core): mock_apps = AsyncMock() mock_apps.delete_namespaced_deployment.side_effect = [ None, ApiException(status=500, reason="fail"), ] mock_v1 = AsyncMock() mock_net = AsyncMock() mock_custom = AsyncMock() mock_custom.list_namespaced_custom_object.return_value = {"items": []} _setup_clients(core, apps=mock_apps, v1=mock_v1, net=mock_net, custom=mock_custom) deps = [mock_deployment(name="ok"), mock_deployment(name="fail")] with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, return_value=deps): result = await core.delete_all_sandboxes() assert result.success is False async def test_no_sandboxes(self, core: K7Core): mock_apps = AsyncMock() mock_v1 = AsyncMock() mock_net = AsyncMock() mock_custom = AsyncMock() mock_custom.list_namespaced_custom_object.return_value = {"items": []} _setup_clients(core, apps=mock_apps, v1=mock_v1, net=mock_net, custom=mock_custom) with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, return_value=[]): result = await core.delete_all_sandboxes() assert result.success is True assert result.data == [] async def test_exception_from_listing(self, core: K7Core): _setup_clients(core) with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, side_effect=Exception("boom")): result = await core.delete_all_sandboxes() assert result.success is False # --- fork_sandbox --- class TestForkSandbox: def _setup_fork(self, core): """Set up mock clients for fork tests and return them.""" mock_apps = AsyncMock() mock_v1 = AsyncMock() # Source has no egress policy at all → the fork inherits open egress. mock_net = AsyncMock() mock_net.read_namespaced_network_policy.side_effect = ApiException(status=404) mock_custom = AsyncMock() mock_custom.get_namespaced_custom_object.side_effect = ApiException(status=404) _setup_clients(core, apps=mock_apps, v1=mock_v1, net=mock_net, custom=mock_custom) core._detect_backend = AsyncMock(return_value="kata-qemu-longhorn") src_dep = MagicMock() src_dep.metadata.name = "src" src_dep.metadata.labels = {"app": "src", "katakate.org/sandbox": "src"} src_dep.spec.selector.match_labels = {"app": "src"} src_dep.spec.template.metadata.labels = {"app": "src", "katakate.org/sandbox": "src"} src_dep.spec.template.spec.volumes = [] src_dep.spec.replicas = 1 # 0 ready replicas ⇒ fork skips the pre-snapshot guest `sync` # (a paused/quiesced source needs no flush) — keeps these mocked # tests off the exec_command path. src_dep.status.ready_replicas = 0 mock_apps.read_namespaced_deployment.return_value = src_dep src_pvc = MagicMock() src_pvc.spec.access_modes = ["ReadWriteOnce"] src_pvc.spec.resources = MagicMock() src_pvc.spec.storage_class_name = "longhorn" src_pvc.spec.volume_mode = "Filesystem" mock_v1.read_namespaced_persistent_volume_claim.return_value = src_pvc return mock_apps, mock_v1 async def test_source_deployment_not_found(self, core: K7Core): mock_apps = AsyncMock() mock_apps.read_namespaced_deployment.side_effect = ApiException(status=404) _setup_clients(core, apps=mock_apps, v1=AsyncMock()) core._detect_backend = AsyncMock(return_value="kata-qemu-longhorn") result = await core.fork_sandbox("src", "dst") assert result.success is False async def test_source_pvc_not_found(self, core: K7Core): mock_apps = AsyncMock() mock_v1 = AsyncMock() mock_v1.read_namespaced_persistent_volume_claim.side_effect = ApiException(status=404) _setup_clients(core, apps=mock_apps, v1=mock_v1) core._detect_backend = AsyncMock(return_value="kata-qemu-longhorn") result = await core.fork_sandbox("src", "dst") assert result.success is False assert "not found" in result.error async def test_snapshot_creation_fails(self, core: K7Core): self._setup_fork(core) with patch.object( core, "_create_volume_snapshot", new_callable=AsyncMock, return_value=OperationResult(success=False, error="snap fail"), ): result = await core.fork_sandbox("src", "dst") assert result.success is False assert "snap fail" in result.error async def test_snapshot_not_ready(self, core: K7Core): self._setup_fork(core) with ( patch.object( core, "_create_volume_snapshot", new_callable=AsyncMock, return_value=OperationResult(success=True), ), patch.object( core, "_wait_for_snapshot_ready", new_callable=AsyncMock, return_value=OperationResult(success=False, error="timeout"), ), ): result = await core.fork_sandbox("src", "dst") assert result.success is False async def test_target_already_exists(self, core: K7Core): mock_apps, mock_v1 = self._setup_fork(core) mock_apps.create_namespaced_deployment.side_effect = ApiException(status=409) with ( patch.object( core, "_create_volume_snapshot", new_callable=AsyncMock, return_value=OperationResult(success=True), ), patch.object( core, "_wait_for_snapshot_ready", new_callable=AsyncMock, return_value=OperationResult(success=True), ), patch.object( core, "_wait_for_pvc_bound", new_callable=AsyncMock, return_value=OperationResult(success=True), ), ): result = await core.fork_sandbox("src", "dst") assert result.success is False assert "already exists" in result.error async def test_happy_path(self, core: K7Core): mock_apps, _ = self._setup_fork(core) with ( patch.object( core, "_create_volume_snapshot", new_callable=AsyncMock, return_value=OperationResult(success=True), ), patch.object( core, "_wait_for_snapshot_ready", new_callable=AsyncMock, return_value=OperationResult(success=True), ), patch.object( core, "_wait_for_pvc_bound", new_callable=AsyncMock, return_value=OperationResult(success=True), ), ): result = await core.fork_sandbox("src", "dst") assert result.success is True assert "Forked" in result.message mock_apps.create_namespaced_deployment.assert_called_once() def _patch_fork_storage(self, core: K7Core): return ( patch.object(core, "_create_volume_snapshot", new_callable=AsyncMock, return_value=OperationResult(True)), patch.object(core, "_wait_for_snapshot_ready", new_callable=AsyncMock, return_value=OperationResult(True)), patch.object(core, "_wait_for_pvc_bound", new_callable=AsyncMock, return_value=OperationResult(True)), ) async def test_fork_applies_network_policies(self, core: K7Core): """a fork used to come up with no policy at all.""" self._setup_fork(core) core._custom_objects_client.get_namespaced_custom_object.side_effect = None core._custom_objects_client.get_namespaced_custom_object.return_value = { "spec": {"egress": [{"toFQDNs": [{"matchName": "example.com"}]}, {"toCIDR": ["10.0.0.0/8"]}]} } snap, ready, bound = self._patch_fork_storage(core) with ( snap, ready, bound, patch.object( core, "_apply_sandbox_network_policies", new_callable=AsyncMock, return_value=OperationResult(True), ) as apply_policies, ): result = await core.fork_sandbox("src", "dst") assert result.success is True assert apply_policies.call_args.kwargs["name"] == "dst" assert apply_policies.call_args.kwargs["egress_whitelist"] == ["example.com", "10.0.0.0/8"] async def test_fork_of_open_egress_source_inherits_open_egress(self, core: K7Core): self._setup_fork(core) snap, ready, bound = self._patch_fork_storage(core) with ( snap, ready, bound, patch.object( core, "_apply_sandbox_network_policies", new_callable=AsyncMock, return_value=OperationResult(True), ) as apply_policies, ): result = await core.fork_sandbox("src", "dst") assert result.success is True assert apply_policies.call_args.kwargs["egress_whitelist"] is None async def test_fork_policy_failure_rolls_back_deployment(self, core: K7Core): self._setup_fork(core) snap, ready, bound = self._patch_fork_storage(core) with ( snap, ready, bound, patch.object( core, "_apply_sandbox_network_policies", new_callable=AsyncMock, return_value=OperationResult(False, error="boom"), ), patch.object( core, "_delete_sandbox_resources", new_callable=AsyncMock, return_value=OperationResult(True), ) as rollback, ): result = await core.fork_sandbox("src", "dst") assert result.success is False assert "rolled back" in result.error and "boom" in result.error rollback.assert_awaited_once_with("dst", "default") class TestApplySandboxNetworkPolicies: """opt-in ingress, deny-all by default.""" def _net(self, core: K7Core): mock_net = AsyncMock() _setup_clients(core, net=mock_net) return mock_net def _ingress_body(self, mock_net): for call in mock_net.create_namespaced_network_policy.call_args_list: body = call.kwargs["body"] if body.metadata.name.endswith("-deny-ingress"): return body raise AssertionError("no deny-ingress policy created") async def test_default_is_deny_all(self, core: K7Core): mock_net = self._net(core) result = await core._apply_sandbox_network_policies(name="sb", namespace="default", egress_whitelist=None) assert result.success is True body = self._ingress_body(mock_net) assert body.spec.policy_types == ["Ingress"] assert body.spec.ingress == [] assert body.spec.pod_selector.match_labels == {"katakate.org/sandbox": "sb"} async def test_ports_without_sources_allow_same_namespace_sandboxes(self, core: K7Core): mock_net = self._net(core) await core._apply_sandbox_network_policies( name="sb", namespace="default", egress_whitelist=None, ingress_ports=[8000, 8001] ) rule = self._ingress_body(mock_net).spec.ingress[0] assert [(p.protocol, p.port) for p in rule.ports] == [("TCP", 8000), ("TCP", 8001)] expr = rule._from[0].pod_selector.match_expressions[0] assert (expr.key, expr.operator) == ("katakate.org/sandbox", "Exists") async def test_explicit_sources_are_scoped(self, core: K7Core): mock_net = self._net(core) await core._apply_sandbox_network_policies( name="sb", namespace="default", egress_whitelist=None, ingress_ports=[8000], ingress_from=["sandbox:alice", "cidr:203.0.113.0/24"], ) peers = self._ingress_body(mock_net).spec.ingress[0]._from assert peers[0].pod_selector.match_labels == {"katakate.org/sandbox": "alice"} assert peers[1].ip_block.cidr == "203.0.113.0/24" async def test_bad_source_fails_without_creating_the_policy(self, core: K7Core): mock_net = self._net(core) result = await core._apply_sandbox_network_policies( name="sb", namespace="default", egress_whitelist=None, ingress_ports=[8000], ingress_from=["nonsense"] ) assert result.success is False assert "nonsense" in result.error mock_net.create_namespaced_network_policy.assert_not_called() async def test_policy_name_is_unchanged_when_it_carries_allow_rules(self, core: K7Core): """Renaming it would strand policies created by older k7 versions, which `_delete_sandbox_resources` deletes by name.""" mock_net = self._net(core) await core._apply_sandbox_network_policies( name="sb", namespace="default", egress_whitelist=None, ingress_ports=[8000] ) assert self._ingress_body(mock_net).metadata.name == "sb-deny-ingress" class TestExposeSandboxPorts: """NodePort Service in front of matching ingress rules.""" async def test_service_uses_local_external_traffic_policy(self, core: K7Core): mock_v1 = AsyncMock() created = MagicMock() port = MagicMock() port.port = 8000 port.node_port = 31234 created.spec.ports = [port] mock_v1.create_namespaced_service.return_value = created _setup_clients(core, v1=mock_v1) result = await core._apply_sandbox_expose_service("sb", "default", [8000]) assert result.success is True assert result.data == {"node_ports": {8000: 31234}} body = mock_v1.create_namespaced_service.call_args.kwargs["body"] assert body.metadata.name == "sb-expose" assert body.spec.type == "NodePort" # Without Local the client's source IP is SNAT'd and cidr: rules are theatre. assert body.spec.external_traffic_policy == "Local" assert body.spec.selector == {"app": "sb"} async def test_unallocated_node_port_is_a_loud_failure(self, core: K7Core): mock_v1 = AsyncMock() created = MagicMock() port = MagicMock() port.port = 8000 port.node_port = None created.spec.ports = [port] mock_v1.create_namespaced_service.return_value = created _setup_clients(core, v1=mock_v1) result = await core._apply_sandbox_expose_service("sb", "default", [8000]) assert result.success is False assert "no node port" in result.error async def test_expose_without_matching_ingress_port_is_rejected(self, core: K7Core): mock_apps = AsyncMock() _setup_clients(core, apps=mock_apps, v1=AsyncMock(), net=AsyncMock()) cfg = SandboxConfig(name="sb1", image="alpine:3.20", ingress_ports=[8000], expose_ports=[9000]) result = await core.create_sandbox(cfg) assert result.success is False assert "[9000]" in result.error mock_apps.create_namespaced_deployment.assert_not_called() class TestReadSandboxIngressRules: async def test_deny_all_reads_back_as_none(self, core: K7Core): np = MagicMock() np.spec.ingress = [] mock_net = AsyncMock() mock_net.read_namespaced_network_policy.return_value = np _setup_clients(core, net=mock_net) assert await core._read_sandbox_ingress_rules("sb", "default") == (None, None) async def test_round_trips_ports_and_sources(self, core: K7Core): """A fork must re-derive the source's policy from what it reads back.""" core_net = AsyncMock() _setup_clients(core, net=core_net) await core._apply_sandbox_network_policies( name="sb", namespace="default", egress_whitelist=None, ingress_ports=[8000], ingress_from=["sandbox:alice", "namespace:team-a", "cidr:203.0.113.0/24"], ) applied = [ c.kwargs["body"] for c in core_net.create_namespaced_network_policy.call_args_list if c.kwargs["body"].metadata.name == "sb-deny-ingress" ][0] core_net.read_namespaced_network_policy.return_value = applied ports, sources = await core._read_sandbox_ingress_rules("sb", "default") assert ports == [8000] assert sources == ["sandbox:alice", "namespace:team-a", "cidr:203.0.113.0/24"] async def test_default_source_reads_back_as_empty_list(self, core: K7Core): core_net = AsyncMock() _setup_clients(core, net=core_net) await core._apply_sandbox_network_policies( name="sb", namespace="default", egress_whitelist=None, ingress_ports=[8000] ) applied = [ c.kwargs["body"] for c in core_net.create_namespaced_network_policy.call_args_list if c.kwargs["body"].metadata.name == "sb-deny-ingress" ][0] core_net.read_namespaced_network_policy.return_value = applied assert await core._read_sandbox_ingress_rules("sb", "default") == ([8000], []) class TestReadSandboxEgressWhitelist: async def test_cilium_wildcard_round_trips(self, core: K7Core): mock_custom = AsyncMock() mock_custom.get_namespaced_custom_object.return_value = { "spec": {"egress": [{"toFQDNs": [{"matchPattern": "**.docker.com"}, {"matchName": "docker.io"}]}]} } _setup_clients(core, custom=mock_custom) assert await core._read_sandbox_egress_whitelist("sb", "default") == ["*.docker.com", "docker.io"] async def test_v1_netpol_cidrs(self, core: K7Core): np = MagicMock() peer = MagicMock() peer.ip_block.cidr = "10.0.0.0/8" rule = MagicMock() rule.to = [peer] np.spec.egress = [rule] mock_net = AsyncMock() mock_net.read_namespaced_network_policy.return_value = np mock_custom = AsyncMock() mock_custom.get_namespaced_custom_object.side_effect = ApiException(status=404) _setup_clients(core, net=mock_net, custom=mock_custom) assert await core._read_sandbox_egress_whitelist("sb", "default") == ["10.0.0.0/8"] async def test_block_all_egress_round_trips_as_empty_list(self, core: K7Core): """`egress_whitelist=[]` (block everything) must not degrade to None (open) when a fork inherits it.""" np = MagicMock() np.spec.egress = [] mock_net = AsyncMock() mock_net.read_namespaced_network_policy.return_value = np mock_custom = AsyncMock() mock_custom.get_namespaced_custom_object.side_effect = ApiException(status=404) _setup_clients(core, net=mock_net, custom=mock_custom) assert await core._read_sandbox_egress_whitelist("sb", "default") == [] async def test_no_policy_means_open_egress(self, core: K7Core): mock_net = AsyncMock() mock_net.read_namespaced_network_policy.side_effect = ApiException(status=404) mock_custom = AsyncMock() mock_custom.get_namespaced_custom_object.side_effect = ApiException(status=404) _setup_clients(core, net=mock_net, custom=mock_custom) assert await core._read_sandbox_egress_whitelist("sb", "default") is None # --- per-node agent forwarding --- class TestK7dAgentForwarding: def _remote_pod(self, node: str = "node-remote", name: str = "sb1-pod"): pod = mock_pod(name=name, phase="Running") pod.spec.node_name = node return pod async def test_pause_forwards_to_remote_agent(self, core: K7Core): """k7d pause of a sandbox on another node is forwarded, not failed.""" dep = mock_deployment(annotations={"k7.katakate.org/backend": "k7d"}) mock_apps = AsyncMock() mock_apps.read_namespaced_deployment.return_value = dep mock_v1 = AsyncMock() mock_v1.list_namespaced_pod.return_value = MagicMock(items=[self._remote_pod()]) _setup_clients(core, apps=mock_apps, v1=mock_v1) with ( patch.dict("os.environ", {"K7_NODE_NAME": "node-local"}), patch.object( core, "_k7d_forward_vm_op", new_callable=AsyncMock, return_value=OperationResult(success=True, message="forwarded"), ) as fwd, ): result = await core.pause_sandbox("sb1") assert result.success is True fwd.assert_called_once() assert fwd.call_args.args[0] == "node-remote" assert fwd.call_args.args[1] == "pause" assert fwd.call_args.args[2]["pod_name"] == "sb1-pod" mock_apps.patch_namespaced_deployment.assert_awaited() async def test_fork_looks_up_on_remote_creates_locally(self, core: K7Core): """k7d fork Kubernetes writes stay on k7-api; agent only looks up the VM.""" dep = mock_deployment(annotations={"k7.katakate.org/backend": "k7d"}) dep.spec.template.metadata.labels = {"app": "sb1"} dep.metadata.labels = {"app": "sb1"} dep.spec.selector.match_labels = {"app": "sb1"} mock_apps = AsyncMock() mock_apps.read_namespaced_deployment.return_value = dep mock_v1 = AsyncMock() mock_v1.list_namespaced_pod.return_value = MagicMock(items=[self._remote_pod()]) _setup_clients(core, apps=mock_apps, v1=mock_v1) vm = {"vm_id": "vm-1-0", "sandbox_id": "cri-src", "cluster_id": "cri-src"} with ( patch.dict("os.environ", {"K7_NODE_NAME": "node-local"}), patch.object(core, "lookup_k7d_vm", new=AsyncMock(return_value=vm)), patch.object(core, "_wait_for_pod_container_started", new=AsyncMock(return_value="fork-pod")), 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))), patch.object(core, "_k7d_forward_vm_op", new_callable=AsyncMock) as fwd, ): result = await core.fork_sandbox("sb1", "sb1-fork") assert result.success, result.error fwd.assert_not_called() mock_apps.create_namespaced_deployment.assert_awaited() new_dep = mock_apps.create_namespaced_deployment.call_args.kwargs["body"] assert new_dep.spec.template.spec.node_name == "node-remote" async def test_agent_refuses_to_reforward(self, core: K7Core): """An agent that resolves a sandbox to yet another node must fail loudly instead of forwarding again (no proxy loops).""" with patch.dict("os.environ", {"K7_AGENT": "1", "K7_NODE_NAME": "node-a"}): try: await core._k7d_forward_vm_op("node-b", "pause", {}, timeout=5) raise AssertionError("expected RuntimeError") except RuntimeError as e: assert "refusing to re-forward" in str(e) async def test_missing_agent_pod_fails_loudly(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.list_namespaced_pod.return_value = MagicMock(items=[]) _setup_clients(core, v1=mock_v1) try: await core._k7d_agent_base_url("node-remote") raise AssertionError("expected RuntimeError") except RuntimeError as e: assert "k7-agent" in str(e) async def test_missing_token_fails_loudly(self, core: K7Core, tmp_path): with patch.dict("os.environ", {"K7_AGENT_TOKEN_FILE": str(tmp_path / "absent")}): try: core._k7d_agent_token() raise AssertionError("expected RuntimeError") except RuntimeError as e: assert "agent token" in str(e) async def test_nodes_storage_reports_unreachable_agent(self, core: K7Core, tmp_path): """A node with no Ready agent gets a loud error entry, not omission.""" token_file = tmp_path / "agent_token" token_file.write_text("tok\n") mock_v1 = AsyncMock() mock_v1.list_node.return_value = SimpleNamespace(items=[_node_named("n1")]) mock_v1.list_namespaced_pod.return_value = MagicMock(items=[]) _setup_clients(core, v1=mock_v1) with patch.dict("os.environ", {"K7_AGENT_TOKEN_FILE": str(token_file)}): result = await core.nodes_storage() assert "n1" in result assert "k7-agent" in result["n1"]["error"] def _node_named(name: str): n = MagicMock() n.metadata.name = name return n # --- exec_command --- class TestExecCommand: async def test_sandbox_not_found(self, core: K7Core): mock_apps = AsyncMock() mock_apps.read_namespaced_deployment.side_effect = ApiException(status=404) _setup_clients(core, apps=mock_apps, v1=AsyncMock()) result = await core.exec_command("missing", "echo hi") assert result.exit_code == 1 assert "not found" in result.stderr async def test_no_pods(self, core: K7Core): mock_apps = AsyncMock() mock_v1 = AsyncMock() mock_v1.list_namespaced_pod.return_value = MagicMock(items=[]) _setup_clients(core, apps=mock_apps, v1=mock_v1) result = await core.exec_command("sb1", "echo hi") assert result.exit_code == 1 assert "No pods" in result.stderr async def test_pod_not_running(self, core: K7Core): mock_apps = AsyncMock() mock_v1 = AsyncMock() pod = mock_pod(phase="Pending") mock_v1.list_namespaced_pod.return_value = MagicMock(items=[pod]) _setup_clients(core, apps=mock_apps, v1=mock_v1) result = await core.exec_command("sb1", "echo hi") assert result.exit_code == 1 assert "not running" in result.stderr async def test_successful_exec(self, core: K7Core): mock_apps = AsyncMock() mock_v1 = AsyncMock() pod = mock_pod(phase="Running") mock_v1.list_namespaced_pod.return_value = MagicMock(items=[pod]) _setup_clients(core, apps=mock_apps, v1=mock_v1) with patch("k7.core.core.WsApiClient") as mock_ws_cls: mock_ws_ctx = AsyncMock() mock_ws_cls.return_value.__aenter__ = AsyncMock(return_value=mock_ws_ctx) mock_ws_cls.return_value.__aexit__ = AsyncMock(return_value=False) mock_v1_ws = AsyncMock() mock_v1_ws.connect_get_namespaced_pod_exec.return_value = "hello\n" with patch("k7.core.core.client.CoreV1Api", return_value=mock_v1_ws): result = await core.exec_command("sb1", "echo hello") assert result.exit_code == 0 assert result.stdout == "hello\n" assert result.duration_ms >= 0 # --- get_sandbox_metrics --- class TestGetSandboxMetrics: async def test_running_pod_with_metrics(self, core: K7Core): mock_apps = AsyncMock() mock_v1 = AsyncMock() mock_metrics = AsyncMock() _setup_clients(core, apps=mock_apps, v1=mock_v1, metrics=mock_metrics) dep = mock_deployment(name="sb1") pod = mock_pod(name="sb1-abc", phase="Running") mock_v1.list_namespaced_pod.return_value = MagicMock(items=[pod]) mock_metrics.get_namespaced_custom_object.return_value = { "containers": [{"usage": {"cpu": "100m", "memory": "128Mi"}}] } with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, return_value=[dep]): result = await core.get_sandbox_metrics(namespace="default") assert len(result) == 1 assert result[0]["cpu_usage"] == "100m" assert result[0]["memory_usage"] == "128Mi" async def test_no_pods_skipped(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.list_namespaced_pod.return_value = MagicMock(items=[]) mock_metrics = AsyncMock() _setup_clients(core, v1=mock_v1, metrics=mock_metrics) dep = mock_deployment(name="sb1") with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, return_value=[dep]): result = await core.get_sandbox_metrics(namespace="default") assert result == [] async def test_pod_not_running_skipped(self, core: K7Core): mock_v1 = AsyncMock() pod = mock_pod(phase="Pending") mock_v1.list_namespaced_pod.return_value = MagicMock(items=[pod]) mock_metrics = AsyncMock() _setup_clients(core, v1=mock_v1, metrics=mock_metrics) dep = mock_deployment(name="sb1") with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, return_value=[dep]): result = await core.get_sandbox_metrics(namespace="default") assert result == [] async def test_metrics_api_error(self, core: K7Core): _setup_clients(core) with patch.object(core, "_get_kata_sandboxes", new_callable=AsyncMock, side_effect=Exception("boom")): result = await core.get_sandbox_metrics(namespace="default") assert result == [] # --- _wait_for_pvc_bound --- class TestWaitForPvcBound: async def test_bound_immediately(self, core: K7Core): mock_v1 = AsyncMock() pvc = MagicMock() pvc.status.phase = "Bound" mock_v1.read_namespaced_persistent_volume_claim.return_value = pvc _setup_clients(core, v1=mock_v1) result = await core._wait_for_pvc_bound("pvc1", timeout=5) assert result.success is True async def test_bound_after_retry(self, core: K7Core): mock_v1 = AsyncMock() pending = MagicMock() pending.status.phase = "Pending" bound = MagicMock() bound.status.phase = "Bound" mock_v1.read_namespaced_persistent_volume_claim.side_effect = [pending, bound] _setup_clients(core, v1=mock_v1) with patch("k7.core.core.asyncio.sleep", new_callable=AsyncMock): result = await core._wait_for_pvc_bound("pvc1", timeout=10) assert result.success is True async def test_404_then_bound(self, core: K7Core): mock_v1 = AsyncMock() bound = MagicMock() bound.status.phase = "Bound" mock_v1.read_namespaced_persistent_volume_claim.side_effect = [ ApiException(status=404), bound, ] _setup_clients(core, v1=mock_v1) with patch("k7.core.core.asyncio.sleep", new_callable=AsyncMock): result = await core._wait_for_pvc_bound("pvc1", timeout=10) assert result.success is True async def test_non_404_error(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.read_namespaced_persistent_volume_claim.side_effect = ApiException(status=500) _setup_clients(core, v1=mock_v1) result = await core._wait_for_pvc_bound("pvc1", timeout=5) assert result.success is False assert "check error" in result.error async def test_timeout(self, core: K7Core): mock_v1 = AsyncMock() pending = MagicMock() pending.status.phase = "Pending" mock_v1.read_namespaced_persistent_volume_claim.return_value = pending _setup_clients(core, v1=mock_v1) elapsed = 0.0 def advancing_time(): nonlocal elapsed elapsed += 1.0 return elapsed with ( patch("k7.core.core.time.time", side_effect=advancing_time), patch("k7.core.core.asyncio.sleep", new_callable=AsyncMock), ): result = await core._wait_for_pvc_bound("pvc1", timeout=2) assert result.success is False assert "not Bound" in result.error # --- _wait_for_job --- class TestWaitForJob: async def test_job_succeeds(self, core: K7Core): mock_batch = AsyncMock() job = MagicMock() job.status.succeeded = 1 job.status.failed = 0 mock_batch.read_namespaced_job.return_value = job _setup_clients(core, batch=mock_batch) result = await core._wait_for_job("j1", timeout=5) assert result.success is True async def test_job_fails(self, core: K7Core): mock_batch = AsyncMock() job = MagicMock() job.status.succeeded = 0 job.status.failed = 1 mock_batch.read_namespaced_job.return_value = job _setup_clients(core, batch=mock_batch) result = await core._wait_for_job("j1", timeout=5) assert result.success is False assert "failed" in result.error async def test_404_then_succeeds(self, core: K7Core): mock_batch = AsyncMock() job_ok = MagicMock() job_ok.status.succeeded = 1 job_ok.status.failed = 0 mock_batch.read_namespaced_job.side_effect = [ApiException(status=404), job_ok] _setup_clients(core, batch=mock_batch) with patch("k7.core.core.asyncio.sleep", new_callable=AsyncMock): result = await core._wait_for_job("j1", timeout=10) assert result.success is True async def test_non_404_error(self, core: K7Core): mock_batch = AsyncMock() mock_batch.read_namespaced_job.side_effect = ApiException(status=500) _setup_clients(core, batch=mock_batch) result = await core._wait_for_job("j1", timeout=5) assert result.success is False assert "wait error" in result.error async def test_timeout(self, core: K7Core): mock_batch = AsyncMock() job = MagicMock() job.status.succeeded = 0 job.status.failed = 0 mock_batch.read_namespaced_job.return_value = job _setup_clients(core, batch=mock_batch) elapsed = 0.0 def advancing_time(): nonlocal elapsed elapsed += 1.0 return elapsed with ( patch("k7.core.core.time.time", side_effect=advancing_time), patch("k7.core.core.asyncio.sleep", new_callable=AsyncMock), ): result = await core._wait_for_job("j1", timeout=2) assert result.success is False assert "did not complete" in result.error # --- create_sandbox validation --- class TestCreateSandboxValidation: async def test_invalid_limits_rejected(self, core: K7Core): mock_apps = AsyncMock() mock_v1 = AsyncMock() mock_net = AsyncMock() _setup_clients(core, apps=mock_apps, v1=mock_v1, net=mock_net) cfg = SandboxConfig(name="sb1", image="alpine:3.18", limits={"cpu": "0m"}) result = await core.create_sandbox(cfg) assert result.success is False assert "Invalid resource limits" in result.error async def test_empty_env_file_rejected(self, core: K7Core, tmp_path): mock_apps = AsyncMock() mock_v1 = AsyncMock() mock_net = AsyncMock() _setup_clients(core, apps=mock_apps, v1=mock_v1, net=mock_net) env_file = tmp_path / ".env" env_file.write_text("# only comments\n\n") cfg = SandboxConfig(name="sb1", image="alpine:3.18", env_file=str(env_file)) result = await core.create_sandbox(cfg) assert result.success is False assert "empty or invalid" in result.error async def test_ingress_from_without_ports_rejected(self, core: K7Core): _setup_clients(core, apps=AsyncMock(), v1=AsyncMock(), net=AsyncMock()) cfg = SandboxConfig(name="sb1", image="alpine:3.18", ingress_from=["sandbox:alice"]) result = await core.create_sandbox(cfg) assert result.success is False assert "ingress_from requires ingress_ports" in result.error async def test_unparseable_ingress_source_rejected_before_provisioning(self, core: K7Core): mock_apps = AsyncMock() _setup_clients(core, apps=mock_apps, v1=AsyncMock(), net=AsyncMock()) cfg = SandboxConfig(name="sb1", image="alpine:3.18", ingress_ports=[8000], ingress_from=["nonsense"]) result = await core.create_sandbox(cfg) assert result.success is False assert "nonsense" in result.error mock_apps.create_namespaced_deployment.assert_not_called() # --- _wait_for_pod_container_started --- class TestWaitForPodContainerStarted: async def test_returns_pod_name_when_running(self, core: K7Core): mock_v1 = AsyncMock() pod = mock_pod(name="sb1-abc", phase="Running") cs = MagicMock() cs.name = "sandbox" cs.state.running = True pod.status.container_statuses = [cs] mock_v1.list_namespaced_pod.return_value = MagicMock(items=[pod]) _setup_clients(core, v1=mock_v1) result = await core._wait_for_pod_container_started("sb1", "default", timeout_seconds=5) assert result == "sb1-abc" async def test_returns_none_on_timeout_no_pods(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.list_namespaced_pod.return_value = MagicMock(items=[]) _setup_clients(core, v1=mock_v1) elapsed = 0.0 def advancing_time(): nonlocal elapsed elapsed += 1.0 return elapsed with ( patch("k7.core.core.time.time", side_effect=advancing_time), patch("k7.core.core.asyncio.sleep", new_callable=AsyncMock), ): result = await core._wait_for_pod_container_started("sb1", "default", timeout_seconds=2) assert result is None async def test_waits_for_running_phase(self, core: K7Core): mock_v1 = AsyncMock() pending_pod = mock_pod(name="sb1-abc", phase="Pending") running_pod = mock_pod(name="sb1-abc", phase="Running") cs = MagicMock() cs.name = "sandbox" cs.state.running = True running_pod.status.container_statuses = [cs] mock_v1.list_namespaced_pod.side_effect = [ MagicMock(items=[pending_pod]), MagicMock(items=[running_pod]), ] _setup_clients(core, v1=mock_v1) with patch("k7.core.core.asyncio.sleep", new_callable=AsyncMock): result = await core._wait_for_pod_container_started("sb1", "default", timeout_seconds=60) assert result == "sb1-abc" async def test_returns_none_when_container_not_started(self, core: K7Core): mock_v1 = AsyncMock() pod = mock_pod(name="sb1-abc", phase="Running") cs = MagicMock() cs.name = "sandbox" cs.state.running = None cs.state.waiting = MagicMock() pod.status.container_statuses = [cs] mock_v1.list_namespaced_pod.return_value = MagicMock(items=[pod]) _setup_clients(core, v1=mock_v1) elapsed = 0.0 def advancing_time(): nonlocal elapsed elapsed += 1.0 return elapsed with ( patch("k7.core.core.time.time", side_effect=advancing_time), patch("k7.core.core.asyncio.sleep", new_callable=AsyncMock), ): result = await core._wait_for_pod_container_started("sb1", "default", timeout_seconds=2) assert result is None # --- _list_backends_per_node --- def _node(name: str, labels: dict[str, str], taints: list | None = None): n = MagicMock() n.metadata.name = name n.metadata.labels = labels n.spec.taints = taints or [] return n class TestListBackendsPerNode: async def test_extracts_backend_labels(self, core: K7Core): mock_v1 = AsyncMock() nodes = [ _node( "node1", { "k7.katakate.org/backend-kata-qemu-longhorn": "true", "k7.katakate.org/backend-kata-firecracker-devmapper": "true", "kubernetes.io/hostname": "node1", }, ), _node("node2", {"k7.katakate.org/backend-kata-qemu-longhorn": "true"}), _node("node3", {}), ] mock_v1.list_node.return_value = SimpleNamespace(items=nodes) _setup_clients(core, v1=mock_v1) result = await core._list_backends_per_node() assert result["node1"] == ["kata-firecracker-devmapper", "kata-qemu-longhorn"] assert result["node2"] == ["kata-qemu-longhorn"] assert result["node3"] == [] async def test_skips_label_with_false_value(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.list_node.return_value = SimpleNamespace( items=[_node("n", {"k7.katakate.org/backend-kata-qemu-longhorn": "false"})] ) _setup_clients(core, v1=mock_v1) result = await core._list_backends_per_node() assert result["n"] == [] class TestListClusterNodes: async def test_hostname_backends_and_tenant(self, core: K7Core): taint = SimpleNamespace(key=K7_TENANT_LABEL, value="acme", effect="NoSchedule") mock_v1 = AsyncMock() mock_v1.list_node.return_value = SimpleNamespace( items=[ _node( "k7-node-01", { "kubernetes.io/hostname": "k7-node-01", "k7.katakate.org/backend-k7d": "true", K7_TENANT_LABEL: "acme", }, taints=[taint], ), _node("k7-node-02", {"kubernetes.io/hostname": "k7-node-02"}), ] ) _setup_clients(core, v1=mock_v1) rows = await core.list_cluster_nodes() by_name = {r["name"]: r for r in rows} assert by_name["k7-node-01"]["hostname"] == "k7-node-01" assert by_name["k7-node-01"]["backends"] == ["k7d"] assert by_name["k7-node-01"]["tenant"] == "acme" assert by_name["k7-node-02"]["hostname"] == "k7-node-02" assert by_name["k7-node-02"]["backends"] == [] assert by_name["k7-node-02"]["tenant"] == "" # --- _check_scheduling --- def _event(reason: str, message: str): e = MagicMock() e.reason = reason e.message = message return e class TestCheckScheduling: async def test_running_pod_returns_success(self, core: K7Core): mock_v1 = AsyncMock() pod = mock_pod(name="sb1-x", phase="Running") mock_v1.list_namespaced_pod.return_value = MagicMock(items=[pod]) _setup_clients(core, v1=mock_v1) result = await core._check_scheduling("sb1", "default", "kata-qemu-longhorn", timeout_seconds=1) assert result.success is True async def test_failed_scheduling_returns_explanatory_error(self, core: K7Core): mock_v1 = AsyncMock() pod = mock_pod(name="sb1-x", phase="Pending") mock_v1.list_namespaced_pod.return_value = MagicMock(items=[pod]) events = SimpleNamespace( items=[ _event( "FailedScheduling", "0/2 nodes are available: 2 node(s) didn't match Pod's node affinity/selector.", ) ] ) mock_v1.list_namespaced_event.return_value = events # Two nodes labeled with FD only — no QL anywhere mock_v1.list_node.return_value = SimpleNamespace( items=[ _node("node1", {"k7.katakate.org/backend-kata-firecracker-devmapper": "true"}), _node("node2", {"k7.katakate.org/backend-kata-firecracker-devmapper": "true"}), ] ) _setup_clients(core, v1=mock_v1) result = await core._check_scheduling("sb1", "default", "kata-qemu-longhorn", timeout_seconds=1) assert result.success is False assert "kata-qemu-longhorn" in result.error assert "kata-firecracker-devmapper" in result.error assert "node1" in result.error assert "node2" in result.error async def test_no_pod_yet_eventually_returns_success(self, core: K7Core): mock_v1 = AsyncMock() mock_v1.list_namespaced_pod.return_value = MagicMock(items=[]) _setup_clients(core, v1=mock_v1) elapsed = 0.0 def advancing_time(): nonlocal elapsed elapsed += 1.0 return elapsed with ( patch("k7.core.core.time.time", side_effect=advancing_time), patch("k7.core.core.asyncio.sleep", new_callable=AsyncMock), ): result = await core._check_scheduling("sb1", "default", "kata-qemu-longhorn", timeout_seconds=2) assert result.success is True async def test_unrelated_event_does_not_trigger(self, core: K7Core): mock_v1 = AsyncMock() pod = mock_pod(name="sb1-x", phase="Pending") mock_v1.list_namespaced_pod.return_value = MagicMock(items=[pod]) mock_v1.list_namespaced_event.return_value = SimpleNamespace( items=[_event("Scheduled", "Successfully assigned default/sb1-x to node1")] ) _setup_clients(core, v1=mock_v1) elapsed = 0.0 def advancing_time(): nonlocal elapsed elapsed += 1.0 return elapsed with ( patch("k7.core.core.time.time", side_effect=advancing_time), patch("k7.core.core.asyncio.sleep", new_callable=AsyncMock), ): result = await core._check_scheduling("sb1", "default", "kata-qemu-longhorn", timeout_seconds=2) assert result.success is True # --- _cilium_available / _apply_cilium_egress_policy --- class TestCiliumAvailable: async def test_returns_true_when_crd_exists(self, core: K7Core): fake_api_ext = AsyncMock() fake_api_ext.read_custom_resource_definition.return_value = MagicMock() _setup_clients(core, apiext=fake_api_ext) assert await core._cilium_available() is True async def test_returns_false_when_crd_missing(self, core: K7Core): fake_api_ext = AsyncMock() fake_api_ext.read_custom_resource_definition.side_effect = ApiException(status=404) _setup_clients(core, apiext=fake_api_ext) assert await core._cilium_available() is False class TestApplyCiliumEgressPolicy: async def test_builds_policy_with_fqdns_and_cidrs(self, core: K7Core): mock_custom = AsyncMock() _setup_clients(core, custom=mock_custom) result = await core._apply_cilium_egress_policy( sandbox_name="sb1", namespace="default", cidrs=["10.0.0.0/8"], fqdns=["api.openai.com", "*.huggingface.co"], ) assert result.success is True args, kwargs = mock_custom.create_namespaced_custom_object.call_args body = kwargs["body"] assert body["kind"] == "CiliumNetworkPolicy" assert body["metadata"]["name"] == "sb1-egress" assert body["spec"]["endpointSelector"]["matchLabels"]["katakate.org/sandbox"] == "sb1" egress = body["spec"]["egress"] # DNS rule first, then FQDN rule, then CIDR rule assert any("toFQDNs" in r for r in egress) fqdn_rule = next(r for r in egress if "toFQDNs" in r) assert {"matchName": "api.openai.com"} in fqdn_rule["toFQDNs"] # Leading `*.` must be translated to Cilium's multi-label wildcard # `**.` — a single `*` never crosses label boundaries, silently # breaking nested subdomains like production.cloudfront.docker.com. assert {"matchPattern": "**.huggingface.co"} in fqdn_rule["toFQDNs"] assert any("toCIDR" in r and r["toCIDR"] == ["10.0.0.0/8"] for r in egress) # DNS proxy allowance is present assert any("toPorts" in r and any("rules" in p and "dns" in p["rules"] for p in r["toPorts"]) for r in egress) async def test_leading_wildcard_becomes_multilabel_pattern(self, core: K7Core): mock_custom = AsyncMock() _setup_clients(core, custom=mock_custom) result = await core._apply_cilium_egress_policy( sandbox_name="sb1", namespace="default", cidrs=[], fqdns=["*.docker.com", "**.already.com", "api-*.foo.com"], ) assert result.success is True body = mock_custom.create_namespaced_custom_object.call_args.kwargs["body"] fqdn_rule = next(r for r in body["spec"]["egress"] if "toFQDNs" in r) assert {"matchPattern": "**.docker.com"} in fqdn_rule["toFQDNs"] # Explicit `**.` is passed through untouched; mid-label wildcards too. assert {"matchPattern": "**.already.com"} in fqdn_rule["toFQDNs"] assert {"matchPattern": "api-*.foo.com"} in fqdn_rule["toFQDNs"] async def test_replaces_on_409(self, core: K7Core): mock_custom = AsyncMock() mock_custom.create_namespaced_custom_object.side_effect = ApiException(status=409) _setup_clients(core, custom=mock_custom) result = await core._apply_cilium_egress_policy("sb1", "default", [], ["api.openai.com"]) assert result.success is True mock_custom.replace_namespaced_custom_object.assert_called_once() async def test_non_conflict_error_propagates(self, core: K7Core): mock_custom = AsyncMock() mock_custom.create_namespaced_custom_object.side_effect = ApiException(status=500) _setup_clients(core, custom=mock_custom) result = await core._apply_cilium_egress_policy("sb1", "default", [], ["api.openai.com"]) assert result.success is False assert "CiliumNetworkPolicy" in result.error # --- kubernetes_asyncio ApiClient lifecycle --- class TestApiClientLifecycle: async def test_typed_clients_share_one_api_client(self, core: K7Core): api = MagicMock() api.close = AsyncMock() with ( patch("k7.core.core.client.ApiClient", return_value=api) as api_cls, patch("k7.core.core.client.CoreV1Api") as core_cls, patch("k7.core.core.client.AppsV1Api") as apps_cls, ): await core._get_core_v1_client() await core._get_apps_v1_client() api_cls.assert_called_once() core_cls.assert_called_once_with(api) apps_cls.assert_called_once_with(api) await core.aclose() api.close.assert_awaited_once() assert core._api_client is None assert core._core_v1_client is None assert core._apps_v1_client is None async def test_aclose_is_idempotent(self, core: K7Core): await core.aclose() await core.aclose() async def test_aclose_allows_a_new_session(self, core: K7Core): first = MagicMock() first.close = AsyncMock() second = MagicMock() second.close = AsyncMock() with ( patch("k7.core.core.client.ApiClient", side_effect=[first, second]), patch("k7.core.core.client.CoreV1Api"), ): await core._get_core_v1_client() await core.aclose() first.close.assert_awaited_once() await core._get_core_v1_client() await core.aclose() second.close.assert_awaited_once() class TestCreateSnapshotMessage: async def test_named_snapshot_keeps_create_message(self, core: K7Core): """Regression: quiesced success used to drop the VolumeSnapshot name. ``k7 snapshot create`` echoes ``result.message``; empty stdout failed ``TestSnapshotCrud`` with ``assert snap in cp.stdout``. """ dep = MagicMock() dep.metadata.annotations = {} dep.status.ready_replicas = 1 mock_apps = AsyncMock() mock_apps.read_namespaced_deployment.return_value = dep _setup_clients(core, apps=mock_apps) with ( patch.object(core, "_detect_backend", new=AsyncMock(return_value="kata-qemu-longhorn")), patch.object( core, "_create_volume_snapshot", new=AsyncMock( return_value=OperationResult(success=True, message="Snapshot snap1 created for PVC sb-root-lh") ), ), patch.object( core, "exec_command", new=AsyncMock(return_value=ExecResult(exit_code=0, stdout="", stderr="", duration_ms=0)), ), ): result = await core.create_snapshot("sb", "snap1") assert result.success, result.error assert "snap1" in result.message