opensandbox-group--opensandbox/server/tests/k8s/test_kubernetes_service.py

# Copyright 2025 Alibaba Group Holding Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import pytest
from types import SimpleNamespace
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock, patch
from fastapi import HTTPException

from opensandbox_server.services.k8s.kubernetes_service import KubernetesSandboxService
from opensandbox_server.services.constants import (
    OPEN_SANDBOX_EGRESS_AUTH_HEADER,
    OPEN_SANDBOX_SECURE_ACCESS_HEADER,
    SANDBOX_EGRESS_AUTH_TOKEN_METADATA_KEY,
    SANDBOX_SECURE_ACCESS_TOKEN_METADATA_KEY,
    SANDBOX_ID_LABEL,
    SANDBOX_MANAGED_VOLUMES_LABEL,
    SANDBOX_MANUAL_CLEANUP_LABEL,
    SandboxErrorCodes,
)
from opensandbox_server.api.schema import (
    CredentialProxyConfig,
    ImageAuth,
    ListSandboxesRequest,
    NetworkPolicy,
    PlatformSpec,
)
from opensandbox_server.config import (
    EGRESS_MODE_DNS,
    EGRESS_MODE_DNS_NFT,
    EgressConfig,
    GatewayConfig,
    GatewayRouteModeConfig,
    IngressConfig,
    SecureAccessConfig,
    SecureAccessKey,
)
from opensandbox_server.api.schema import Endpoint

class TestKubernetesSandboxServiceInit:
    
    def test_init_with_valid_config_succeeds(self, k8s_app_config):
        with patch('opensandbox_server.services.k8s.kubernetes_service.K8sClient') as mock_k8s_client, \
             patch('opensandbox_server.services.k8s.kubernetes_service.create_workload_provider') as mock_create_provider:
            
            mock_provider = MagicMock()
            mock_create_provider.return_value = mock_provider
            
            service = KubernetesSandboxService(k8s_app_config)
            
            assert service.namespace == k8s_app_config.kubernetes.namespace
            assert service.execd_image == k8s_app_config.runtime.execd_image
            mock_k8s_client.assert_called_once_with(k8s_app_config.kubernetes)
            mock_create_provider.assert_called_once()
    
    def test_init_without_kubernetes_config_raises_error(self, app_config_no_k8s):
        # app_config_no_k8s still has kubernetes config, just without kubeconfig
        # This will cause K8sClient initialization to fail and raise HTTPException
        with pytest.raises(HTTPException) as exc_info:
            KubernetesSandboxService(app_config_no_k8s)
        
        assert exc_info.value.status_code == 503
        assert exc_info.value.detail["code"] == SandboxErrorCodes.K8S_INITIALIZATION_ERROR
    
    def test_init_with_wrong_runtime_type_raises_error(self, app_config_docker):
        with pytest.raises(ValueError, match="requires runtime.type = 'kubernetes'"):
            KubernetesSandboxService(app_config_docker)
    
    def test_init_with_k8s_client_failure_raises_http_exception(self, k8s_app_config):
        with patch('opensandbox_server.services.k8s.kubernetes_service.K8sClient') as mock_k8s_client:
            mock_k8s_client.side_effect = Exception("Failed to load kubeconfig")
            
            with pytest.raises(HTTPException) as exc_info:
                KubernetesSandboxService(k8s_app_config)
            
            assert exc_info.value.status_code == 503
            assert "code" in exc_info.value.detail
            assert exc_info.value.detail["code"] == SandboxErrorCodes.K8S_INITIALIZATION_ERROR

class TestKubernetesSandboxServiceCreate:

    def test_credential_proxy_requires_dns_nft_mode(
        self, k8s_service, create_sandbox_request
    ):
        create_sandbox_request.network_policy = NetworkPolicy(default_action="deny", egress=[])
        create_sandbox_request.credential_proxy = CredentialProxyConfig(enabled=True)
        k8s_service.app_config.egress = EgressConfig(
            image="opensandbox/egress:v1.1.4", mode=EGRESS_MODE_DNS
        )

        with pytest.raises(HTTPException) as exc_info:
            k8s_service._ensure_network_policy_support(create_sandbox_request)

        assert exc_info.value.status_code == 400
        assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PARAMETER
        assert "dns+nft" in exc_info.value.detail["message"]

    @pytest.mark.asyncio
    async def test_create_sandbox_with_valid_request_succeeds(
        self, k8s_service, create_sandbox_request, mock_workload
    ):
        # Mock workload provider
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-sandbox-123",
            "uid": "abc-123",
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Pod is running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = "10.244.0.5:8080"
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)
        
        response = await k8s_service.create_sandbox(create_sandbox_request)
        
        # CreateSandboxResponse uses 'id' field
        assert response.id is not None
        assert response.status.state == "Running"
        k8s_service.workload_provider.create_workload.assert_called_once()

    @pytest.mark.asyncio
    async def test_create_sandbox_response_restores_extensions_from_workload(
        self, k8s_service, create_sandbox_request, mock_workload
    ):
        create_sandbox_request.extensions = {
            "opensandbox.extensions.custom-label": "中文数据",
        }
        mock_workload["metadata"]["annotations"] = {
            "opensandbox.io/extensions.custom-label": "中文数据",
        }
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-sandbox-123",
            "uid": "abc-123",
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Pod is running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)

        response = await k8s_service.create_sandbox(create_sandbox_request)

        assert response.extensions == {
            "opensandbox.extensions.custom-label": "中文数据",
        }

    @pytest.mark.asyncio
    async def test_create_sandbox_response_ignores_non_dict_workload_annotations(
        self, k8s_service, create_sandbox_request
    ):
        workload = SimpleNamespace(
            metadata=SimpleNamespace(annotations=["not", "a", "mapping"]),
            spec=SimpleNamespace(),
        )
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-sandbox-123",
            "uid": "abc-123",
        }
        k8s_service.workload_provider.get_workload.return_value = workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Pod is running",
            "last_transition_at": datetime.now(timezone.utc),
        }

        response = await k8s_service.create_sandbox(create_sandbox_request)

        assert response.extensions is None

    def test_list_sandboxes_restores_extensions_from_workloads(self, k8s_service, mock_workload):
        mock_workload["metadata"]["annotations"] = {
            "opensandbox.io/extensions.custom-label": "中文数据",
        }
        k8s_service.workload_provider.list_workloads.return_value = [mock_workload]
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)

        from opensandbox_server.api.schema import PaginationRequest
        request = ListSandboxesRequest(pagination=PaginationRequest(page=1, page_size=20))
        response = k8s_service.list_sandboxes(request)

        assert response.items[0].extensions == {
            "opensandbox.extensions.custom-label": "中文数据",
        }

    @pytest.mark.asyncio
    async def test_create_sandbox_normalizes_allocated_status_to_running(
        self, k8s_service, create_sandbox_request, mock_workload
    ):
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-sandbox-123",
            "uid": "abc-123",
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Allocated",
            "reason": "IP_ASSIGNED",
            "message": "Pod has IP assigned but not ready",
            "last_transition_at": datetime.now(timezone.utc),
        }

        response = await k8s_service.create_sandbox(create_sandbox_request)

        assert response.status.state == "Running"
        assert response.status.reason == "IP_ASSIGNED"
        assert response.status.message == "Pod has IP assigned and sandbox is ready for requests"

    @pytest.mark.asyncio
    async def test_create_sandbox_uses_configured_timeout_and_poll_interval(
        self, k8s_service, create_sandbox_request, mock_workload
    ):

        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-sandbox-123",
            "uid": "abc-123",
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Pod is running",
            "last_transition_at": datetime.now(timezone.utc),
        }

        # Override config values
        k8s_service.app_config.kubernetes.sandbox_create_timeout_seconds = 120
        k8s_service.app_config.kubernetes.sandbox_create_poll_interval_seconds = 0.5

        with patch.object(k8s_service, "_wait_for_sandbox_ready", wraps=k8s_service._wait_for_sandbox_ready) as mock_wait:
            await k8s_service.create_sandbox(create_sandbox_request)

        mock_wait.assert_called_once()
        _, kwargs = mock_wait.call_args
        assert kwargs["timeout_seconds"] == 120
        assert kwargs["poll_interval_seconds"] == 0.5

    @pytest.mark.asyncio
    async def test_create_sandbox_rejects_image_auth_when_provider_not_supported(
        self, k8s_service, create_sandbox_request
    ):
        k8s_service.workload_provider.supports_image_auth.return_value = False
        create_sandbox_request.image.auth = ImageAuth(
            username="registry-user",
            password="registry-pass",
        )

        with pytest.raises(HTTPException) as exc_info:
            await k8s_service.create_sandbox(create_sandbox_request)

        assert exc_info.value.status_code == 400
        assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PARAMETER
        k8s_service.workload_provider.create_workload.assert_not_called()

    @pytest.mark.asyncio
    async def test_create_sandbox_allows_image_auth_when_provider_supported(
        self, k8s_service, create_sandbox_request
    ):
        k8s_service.workload_provider.supports_image_auth.return_value = True
        create_sandbox_request.image.auth = ImageAuth(
            username="registry-user",
            password="registry-pass",
        )
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id", "uid": "uid-1"
        }
        k8s_service.workload_provider.get_workload.return_value = MagicMock()
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running", "reason": "", "message": "",
            "last_transition_at": datetime.now(timezone.utc),
        }

        # Should not raise
        await k8s_service.create_sandbox(create_sandbox_request)
        k8s_service.workload_provider.create_workload.assert_called_once()

    @pytest.mark.asyncio
    async def test_create_sandbox_with_no_timeout_calls_provider_with_expires_at_none_and_manual_cleanup_label(
        self, k8s_service, create_sandbox_request
    ):
        """When timeout is None (manual cleanup), provider receives expires_at=None and manual-cleanup label."""
        create_sandbox_request.timeout = None
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id", "uid": "uid-1"
        }
        k8s_service.workload_provider.get_workload.return_value = MagicMock()
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running", "reason": "", "message": "",
            "last_transition_at": datetime.now(timezone.utc),
        }

        await k8s_service.create_sandbox(create_sandbox_request)

        k8s_service.workload_provider.create_workload.assert_called_once()
        _, kwargs = k8s_service.workload_provider.create_workload.call_args
        assert kwargs["expires_at"] is None
        assert kwargs["labels"].get(SANDBOX_MANUAL_CLEANUP_LABEL) == "true"

    @pytest.mark.asyncio
    async def test_create_sandbox_with_network_policy_passes_egress_token_and_annotations(
        self, k8s_service, create_sandbox_request
    ):
        create_sandbox_request.network_policy = NetworkPolicy(default_action="deny", egress=[])
        k8s_service.app_config.egress = EgressConfig(image="opensandbox/egress:v1.1.4")
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id", "uid": "uid-1"
        }
        k8s_service.workload_provider.get_workload.return_value = MagicMock()
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running", "reason": "", "message": "",
            "last_transition_at": datetime.now(timezone.utc),
        }

        with patch(
            "opensandbox_server.services.k8s.kubernetes_service.generate_egress_token",
            return_value="egress-token",
        ):
            await k8s_service.create_sandbox(create_sandbox_request)

        _, kwargs = k8s_service.workload_provider.create_workload.call_args
        assert kwargs["egress_auth_token"] == "egress-token"
        assert kwargs["egress_mode"] == EGRESS_MODE_DNS
        assert kwargs["annotations"][SANDBOX_EGRESS_AUTH_TOKEN_METADATA_KEY] == "egress-token"

    @pytest.mark.asyncio
    async def test_create_sandbox_with_secure_access_passes_annotations(
        self, k8s_service, create_sandbox_request
    ):
        create_sandbox_request.secure_access = True
        k8s_service.app_config.ingress = IngressConfig(
            mode="gateway",
            gateway=GatewayConfig(
                address="gateway.example.com",
                route=GatewayRouteModeConfig(mode="header"),
            ),
        )
        k8s_service.ingress_config = k8s_service.app_config.ingress
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id", "uid": "uid-1"
        }
        k8s_service.workload_provider.get_workload.return_value = MagicMock()
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running", "reason": "", "message": "",
            "last_transition_at": datetime.now(timezone.utc),
        }

        with patch(
            "opensandbox_server.services.k8s.kubernetes_service.generate_secure_access_token",
            return_value="secure-token",
        ):
            await k8s_service.create_sandbox(create_sandbox_request)

        _, kwargs = k8s_service.workload_provider.create_workload.call_args
        assert kwargs["annotations"][SANDBOX_SECURE_ACCESS_TOKEN_METADATA_KEY] == "secure-token"

    @pytest.mark.asyncio
    async def test_create_sandbox_rejects_secure_access_without_gateway_ingress(
        self, k8s_service, create_sandbox_request
    ):
        create_sandbox_request.secure_access = True
        k8s_service.app_config.ingress = IngressConfig(mode="direct")
        k8s_service.ingress_config = k8s_service.app_config.ingress

        with pytest.raises(HTTPException) as exc_info:
            await k8s_service.create_sandbox(create_sandbox_request)

        assert exc_info.value.status_code == 400
        assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PARAMETER
        assert "ingress.mode='gateway'" in exc_info.value.detail["message"]
        k8s_service.workload_provider.create_workload.assert_not_called()

    @pytest.mark.asyncio
    async def test_create_sandbox_with_network_policy_passes_egress_mode_dns_nft_from_config(
        self, k8s_service, create_sandbox_request
    ):
        create_sandbox_request.network_policy = NetworkPolicy(default_action="deny", egress=[])
        k8s_service.app_config.egress = EgressConfig(
            image="opensandbox/egress:v1.1.4",
            mode=EGRESS_MODE_DNS_NFT,
        )
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id", "uid": "uid-1"
        }
        k8s_service.workload_provider.get_workload.return_value = MagicMock()
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running", "reason": "", "message": "",
            "last_transition_at": datetime.now(timezone.utc),
        }

        with patch(
            "opensandbox_server.services.k8s.kubernetes_service.generate_egress_token",
            return_value="egress-token",
        ):
            await k8s_service.create_sandbox(create_sandbox_request)

        _, kwargs = k8s_service.workload_provider.create_workload.call_args
        assert kwargs["egress_mode"] == EGRESS_MODE_DNS_NFT

    @pytest.mark.asyncio
    async def test_create_sandbox_passes_platform_to_workload_provider(
        self, k8s_service, create_sandbox_request
    ):
        create_sandbox_request.platform = PlatformSpec(os="linux", arch="arm64")
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id", "uid": "uid-1"
        }
        k8s_service.workload_provider.get_workload.return_value = {
            "spec": {
                "template": {
                    "spec": {
                        "nodeSelector": {
                            "kubernetes.io/os": "linux",
                            "kubernetes.io/arch": "arm64",
                        }
                    }
                }
            }
        }
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running", "reason": "", "message": "",
            "last_transition_at": datetime.now(timezone.utc),
        }

        response = await k8s_service.create_sandbox(create_sandbox_request)

        _, kwargs = k8s_service.workload_provider.create_workload.call_args
        assert kwargs["platform"] == create_sandbox_request.platform
        assert response.platform is not None
        assert response.platform.os == "linux"
        assert response.platform.arch == "arm64"

    @pytest.mark.asyncio
    async def test_create_sandbox_rejects_unsupported_platform(self, k8s_service, create_sandbox_request):
        create_sandbox_request.platform = PlatformSpec(os="darwin", arch="arm64")

        with pytest.raises(HTTPException) as exc_info:
            await k8s_service.create_sandbox(create_sandbox_request)

        assert exc_info.value.status_code == 400
        assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PARAMETER
        k8s_service.workload_provider.create_workload.assert_not_called()

    @pytest.mark.asyncio
    async def test_create_sandbox_derives_platform_from_node_affinity(
        self, k8s_service, create_sandbox_request
    ):
        create_sandbox_request.platform = None
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id", "uid": "uid-1"
        }
        k8s_service.workload_provider.get_workload.return_value = {
            "spec": {
                "template": {
                    "spec": {
                        "affinity": {
                            "nodeAffinity": {
                                "requiredDuringSchedulingIgnoredDuringExecution": {
                                    "nodeSelectorTerms": [
                                        {
                                            "matchExpressions": [
                                                {
                                                    "key": "kubernetes.io/os",
                                                    "operator": "In",
                                                    "values": ["linux"],
                                                },
                                                {
                                                    "key": "kubernetes.io/arch",
                                                    "operator": "In",
                                                    "values": ["arm64"],
                                                },
                                            ]
                                        }
                                    ]
                                }
                            }
                        }
                    }
                }
            }
        }
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running", "reason": "", "message": "",
            "last_transition_at": datetime.now(timezone.utc),
        }

        response = await k8s_service.create_sandbox(create_sandbox_request)

        assert response.platform is not None
        assert response.platform.os == "linux"
        assert response.platform.arch == "arm64"

    @pytest.mark.asyncio
    async def test_create_sandbox_uses_template_platform_constraints(
        self, k8s_service, create_sandbox_request
    ):
        create_sandbox_request.platform = PlatformSpec(os="linux", arch="arm64")
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id", "uid": "uid-1"
        }
        k8s_service.workload_provider.get_workload.return_value = {
            "spec": {
                "template": {
                    "spec": {
                        "nodeSelector": {
                            "kubernetes.io/os": "linux",
                            "kubernetes.io/arch": "arm64",
                        }
                    }
                }
            }
        }
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running", "reason": "", "message": "",
            "last_transition_at": datetime.now(timezone.utc),
        }
        response = await k8s_service.create_sandbox(create_sandbox_request)

        assert response.platform is not None
        assert response.platform.os == "linux"
        assert response.platform.arch == "arm64"

    def test_get_endpoint_resolve_internal_uses_pod_ip_even_in_gateway_mode(
        self, k8s_service
    ):
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {
                "annotations": {
                    "sandbox.opensandbox.io/endpoints": '["10.0.0.1"]',
                }
            }
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = Endpoint(
            endpoint="gateway.example.com",
            headers={"OpenSandbox-Ingress-To": "sbx-123-44772"},
        )

        endpoint = k8s_service.get_endpoint("sbx-123", 44772, resolve_internal=True)

        assert endpoint.endpoint == "10.0.0.1:44772"
        assert endpoint.headers is None


    def test_get_endpoint_keeps_instance_egress_auth_header_private_for_workload_ports(
        self, k8s_service
    ):
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {
                "annotations": {
                    SANDBOX_EGRESS_AUTH_TOKEN_METADATA_KEY: "egress-token",
                }
            }
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = Endpoint(
            endpoint="gateway.example.com",
            headers={"OpenSandbox-Ingress-To": "sbx-123-44772"},
        )

        endpoint = k8s_service.get_endpoint("sbx-123", 44772)

        assert endpoint.endpoint == "gateway.example.com"
        assert endpoint.headers == {
            "OpenSandbox-Ingress-To": "sbx-123-44772",
        }

    def test_get_endpoint_merges_egress_auth_header_for_egress_api_port(
        self, k8s_service
    ):
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {
                "annotations": {
                    SANDBOX_EGRESS_AUTH_TOKEN_METADATA_KEY: "egress-token",
                }
            }
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = Endpoint(
            endpoint="gateway.example.com",
            headers={"OpenSandbox-Ingress-To": "sbx-123-18080"},
        )

        endpoint = k8s_service.get_endpoint("sbx-123", 18080)

        assert endpoint.endpoint == "gateway.example.com"
        assert endpoint.headers == {
            "OpenSandbox-Ingress-To": "sbx-123-18080",
            OPEN_SANDBOX_EGRESS_AUTH_HEADER: "egress-token",
        }

    def test_get_execd_endpoint_merges_secure_access_header_from_instance_metadata(
        self, k8s_service
    ):
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {
                "annotations": {
                    SANDBOX_SECURE_ACCESS_TOKEN_METADATA_KEY: "secure-token",
                }
            }
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = Endpoint(
            endpoint="gateway.example.com",
            headers={"OpenSandbox-Ingress-To": "sbx-123-44772"},
        )

        endpoint = k8s_service.get_endpoint("sbx-123", 44772)

        assert endpoint.endpoint == "gateway.example.com"
        assert endpoint.headers == {
            "OpenSandbox-Ingress-To": "sbx-123-44772",
            OPEN_SANDBOX_SECURE_ACCESS_HEADER: "secure-token",
        }

    def test_get_user_endpoint_also_merges_secure_access_header(
        self, k8s_service
    ):
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {
                "annotations": {
                    SANDBOX_SECURE_ACCESS_TOKEN_METADATA_KEY: "secure-token",
                }
            }
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = Endpoint(
            endpoint="gateway.example.com",
            headers={"OpenSandbox-Ingress-To": "sbx-123-8080"},
        )

        endpoint = k8s_service.get_endpoint("sbx-123", 8080)

        assert endpoint.headers == {
            "OpenSandbox-Ingress-To": "sbx-123-8080",
            OPEN_SANDBOX_SECURE_ACCESS_HEADER: "secure-token",
        }

    @pytest.mark.asyncio
    async def test_create_sandbox_rejects_timeout_above_configured_maximum(
        self, k8s_service, create_sandbox_request
    ):
        k8s_service.app_config.server.max_sandbox_timeout_seconds = 3600
        create_sandbox_request.timeout = 7200

        with pytest.raises(HTTPException) as exc_info:
            await k8s_service.create_sandbox(create_sandbox_request)

        assert exc_info.value.status_code == 400
        assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PARAMETER
        assert "configured maximum of 3600s" in exc_info.value.detail["message"]
        k8s_service.workload_provider.create_workload.assert_not_called()

    @pytest.mark.asyncio
    async def test_create_sandbox_pool_mode_missing_pool_fails_fast(
        self, k8s_service
    ):
        """Pool mode validates poolRef before creating the BatchSandbox."""
        from opensandbox_server.api.schema import CreateSandboxRequest

        pool_request = CreateSandboxRequest(
            extensions={"poolRef": "does-not-exist-pool"},
        )
        k8s_service.k8s_client.get_custom_object.return_value = None

        with pytest.raises(HTTPException) as exc_info:
            await k8s_service.create_sandbox(pool_request)

        assert exc_info.value.status_code == 404
        assert exc_info.value.detail == {
            "code": SandboxErrorCodes.K8S_POOL_NOT_FOUND,
            "message": "Pool 'does-not-exist-pool' not found.",
        }
        k8s_service.k8s_client.get_custom_object.assert_called_once_with(
            group="sandbox.opensandbox.io",
            version="v1alpha1",
            namespace=k8s_service.namespace,
            plural="pools",
            name="does-not-exist-pool",
        )
        k8s_service.workload_provider.create_workload.assert_not_called()
        k8s_service.workload_provider.delete_workload.assert_not_called()

    @pytest.mark.asyncio
    async def test_create_sandbox_pool_mode_auto_assign_skips_pool_lookup(
        self, k8s_service, mock_workload
    ):
        """Pool auto-assignment uses '*' and must be handled by the controller."""
        from opensandbox_server.api.schema import CreateSandboxRequest

        pool_request = CreateSandboxRequest(
            extensions={"poolRef": "*"},
        )

        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-sandbox-pool-auto",
            "uid": "pool-auto-123",
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Pod is running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = "10.244.0.5:8080"
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)

        response = await k8s_service.create_sandbox(pool_request)

        assert response.id is not None
        k8s_service.k8s_client.get_custom_object.assert_not_called()
        k8s_service.workload_provider.create_workload.assert_called_once()
        assert k8s_service.workload_provider.create_workload.call_args.kwargs["extensions"] == {"poolRef": "*"}

    @pytest.mark.asyncio
    async def test_create_sandbox_pool_mode_skips_image_and_entrypoint_validation(
        self, k8s_service, mock_workload
    ):
        """Pool mode: poolRef only, no image/entrypoint/resourceLimits — should succeed."""
        from opensandbox_server.api.schema import CreateSandboxRequest

        pool_request = CreateSandboxRequest(
            extensions={"poolRef": "my-pool"},
        )

        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-sandbox-pool",
            "uid": "pool-123",
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Pod is running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = "10.244.0.5:8080"
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)

        response = await k8s_service.create_sandbox(pool_request)

        assert response.id is not None
        assert response.status.state == "Running"
        k8s_service.workload_provider.create_workload.assert_called_once()

    @pytest.mark.asyncio
    async def test_create_sandbox_pool_mode_image_auth_guard_no_error(
        self, k8s_service, mock_workload
    ):
        """Pool mode with image=None should not raise AttributeError in _ensure_image_auth_support."""
        from opensandbox_server.api.schema import CreateSandboxRequest

        pool_request = CreateSandboxRequest(
            extensions={"poolRef": "my-pool"},
        )
        assert pool_request.image is None

        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-sandbox-pool2",
            "uid": "pool-456",
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Pod is running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = "10.244.0.6:8080"
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)

        # Should not raise AttributeError on None.auth
        response = await k8s_service.create_sandbox(pool_request)
        assert response.id is not None

class TestWaitForSandboxReady:
    """_wait_for_sandbox_ready method tests"""
    
    @pytest.mark.asyncio
    async def test_wait_for_running_pod_succeeds(self, k8s_service, mock_workload):
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Pod is running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        
        result = await k8s_service._wait_for_sandbox_ready("test-sandbox-id", timeout_seconds=10)
        
        assert result == mock_workload
    
    @pytest.mark.asyncio
    async def test_wait_for_pending_then_running_succeeds(self, k8s_service, mock_workload):
        # Mock state transition: Pending -> Allocated -> Running
        status_sequence = [
            {"state": "Pending", "reason": "", "message": "Pending", "last_transition_at": datetime.now(timezone.utc)},
            {"state": "Allocated", "reason": "IP_ASSIGNED", "message": "IP assigned", "last_transition_at": datetime.now(timezone.utc)},
            {"state": "Running", "reason": "", "message": "Running", "last_transition_at": datetime.now(timezone.utc)},
        ]
        
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.side_effect = status_sequence
        
        result = await k8s_service._wait_for_sandbox_ready("test-sandbox-id", timeout_seconds=10, poll_interval_seconds=0.1)
        
        assert result == mock_workload
        assert k8s_service.workload_provider.get_status.call_count == 2
    
    @pytest.mark.asyncio
    async def test_wait_for_allocated_pod_returns_immediately(self, k8s_service, mock_workload):
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Allocated",
            "reason": "IP_ASSIGNED",
            "message": "Pod has IP assigned",
            "last_transition_at": datetime.now(timezone.utc),
        }
        
        result = await k8s_service._wait_for_sandbox_ready("test-sandbox-id", timeout_seconds=10)
        
        assert result == mock_workload
    
    @pytest.mark.asyncio
    async def test_wait_timeout_raises_exception(self, k8s_service, mock_workload):
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Pending",
            "reason": "",
            "message": "Still pending",
            "last_transition_at": datetime.now(timezone.utc),
        }
        
        with pytest.raises(HTTPException) as exc_info:
            await k8s_service._wait_for_sandbox_ready("test-sandbox-id", timeout_seconds=1, poll_interval_seconds=0.5)
        
        assert exc_info.value.status_code == 504  # Gateway Timeout
        assert "timeout" in exc_info.value.detail["message"].lower()

    @pytest.mark.asyncio
    async def test_wait_returns_400_when_scheduler_marks_platform_unschedulable(self, k8s_service, mock_workload):
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Failed",
            "reason": "POD_PLATFORM_UNSCHEDULABLE",
            "message": "0/1 nodes are available: 1 node(s) didn't match Pod's node affinity.",
            "last_transition_at": datetime.now(timezone.utc),
        }

        with pytest.raises(HTTPException) as exc_info:
            await k8s_service._wait_for_sandbox_ready(
                "test-sandbox-id",
                timeout_seconds=10,
                poll_interval_seconds=0.1,
            )

        assert exc_info.value.status_code == 400
        assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PARAMETER
        assert "unschedulable" in exc_info.value.detail["message"].lower()

    @pytest.mark.asyncio
    async def test_wait_keeps_polling_for_generic_failed_scheduling(self, k8s_service, mock_workload):
        status_sequence = [
            {
                "state": "Pending",
                "reason": "FailedScheduling",
                "message": "0/1 nodes are available: 1 Insufficient cpu.",
                "last_transition_at": datetime.now(timezone.utc),
            },
            {
                "state": "Running",
                "reason": "",
                "message": "Running",
                "last_transition_at": datetime.now(timezone.utc),
            },
        ]
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.side_effect = status_sequence

        result = await k8s_service._wait_for_sandbox_ready(
            "test-sandbox-id",
            timeout_seconds=10,
            poll_interval_seconds=0.1,
        )

        assert result == mock_workload

class TestKubernetesSandboxServiceRenew:
    def test_renew_expiration_rejects_manual_cleanup_sandbox(self, k8s_service):
        k8s_service.workload_provider.get_workload.return_value = MagicMock()
        k8s_service.workload_provider.get_expiration.return_value = None
        request = MagicMock(expires_at=datetime.now(timezone.utc) + timedelta(hours=1))

        with pytest.raises(HTTPException) as exc_info:
            k8s_service.renew_expiration("test-sandbox-id", request)

        assert exc_info.value.status_code == 409
        assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_EXPIRATION
        assert (
            exc_info.value.detail["message"]
            == "Sandbox test-sandbox-id does not have automatic expiration enabled."
        )

class TestGetSandbox:
    """get_sandbox method tests"""
    
    def test_get_existing_sandbox_succeeds(self, k8s_service, mock_workload):
        mock_workload["spec"] = {
            "template": {
                "spec": {
                    "nodeSelector": {
                        "kubernetes.io/os": "linux",
                        "kubernetes.io/arch": "amd64",
                    }
                }
            }
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = "10.0.0.1:8080"
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)
        
        # Use sandbox_id from mock_workload
        sandbox = k8s_service.get_sandbox("test-sandbox-123")
        
        # Sandbox uses 'id' field
        assert sandbox.id == "test-sandbox-123"
        assert sandbox.status.state == "Running"
        assert sandbox.platform is not None
        assert sandbox.platform.os == "linux"
        assert sandbox.platform.arch == "amd64"
    
    def test_get_nonexistent_sandbox_raises_404(self, k8s_service):
        k8s_service.workload_provider.get_workload.return_value = None
        
        with pytest.raises(HTTPException) as exc_info:
            k8s_service.get_sandbox("nonexistent-id")
        
        assert exc_info.value.status_code == 404
        assert "not found" in exc_info.value.detail["message"].lower()

    def test_get_sandbox_extracts_platform_from_affinity(self, k8s_service, mock_workload):
        mock_workload["spec"] = {
            "template": {
                "spec": {
                    "affinity": {
                        "nodeAffinity": {
                            "requiredDuringSchedulingIgnoredDuringExecution": {
                                "nodeSelectorTerms": [
                                    {
                                        "matchExpressions": [
                                            {
                                                "key": "kubernetes.io/os",
                                                "operator": "In",
                                                "values": ["linux"],
                                            },
                                            {
                                                "key": "kubernetes.io/arch",
                                                "operator": "In",
                                                "values": ["amd64"],
                                            },
                                        ]
                                    }
                                ]
                            }
                        }
                    }
                }
            }
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)

        sandbox = k8s_service.get_sandbox("test-sandbox-123")

        assert sandbox.platform is not None
        assert sandbox.platform.os == "linux"
        assert sandbox.platform.arch == "amd64"

    def test_get_sandbox_uses_template_platform_constraints(self, k8s_service, mock_workload):
        mock_workload["spec"] = {
            "template": {
                "spec": {
                    "nodeSelector": {
                        "kubernetes.io/os": "linux",
                        "kubernetes.io/arch": "arm64",
                    }
                }
            }
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)
        sandbox = k8s_service.get_sandbox("test-sandbox-123")

        assert sandbox.platform is not None
        assert sandbox.platform.os == "linux"
        assert sandbox.platform.arch == "arm64"

    def test_get_sandbox_merges_selector_and_affinity_platform_constraints(self, k8s_service, mock_workload):
        mock_workload["spec"] = {
            "template": {
                "spec": {
                    "nodeSelector": {
                        "kubernetes.io/os": "linux",
                    },
                    "affinity": {
                        "nodeAffinity": {
                            "requiredDuringSchedulingIgnoredDuringExecution": {
                                "nodeSelectorTerms": [
                                    {
                                        "matchExpressions": [
                                            {
                                                "key": "kubernetes.io/arch",
                                                "operator": "In",
                                                "values": ["arm64"],
                                            }
                                        ]
                                    }
                                ]
                            }
                        }
                    },
                }
            }
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)
        sandbox = k8s_service.get_sandbox("test-sandbox-123")

        assert sandbox.platform is not None
        assert sandbox.platform.os == "linux"
        assert sandbox.platform.arch == "arm64"

    def test_get_sandbox_returns_null_platform_for_default_scheduling(self, k8s_service, mock_workload):
        mock_workload["spec"] = {
            "template": {
                "spec": {
                    # no nodeSelector/nodeAffinity constraints
                }
            }
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)
        sandbox = k8s_service.get_sandbox("test-sandbox-123")

        assert sandbox.platform is None

class TestDeleteSandbox:
    """delete_sandbox method tests"""
    
    def test_delete_existing_sandbox_succeeds(self, k8s_service, mock_workload):
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.delete_workload.return_value = None

        k8s_service.delete_sandbox("test-sandbox-id")
        
        k8s_service.workload_provider.delete_workload.assert_called_once_with(
            sandbox_id="test-sandbox-id",
            namespace=k8s_service.namespace
        )
    
    def test_delete_nonexistent_sandbox_raises_404(self, k8s_service):
        # Mock delete_workload to raise exception containing "not found"
        k8s_service.workload_provider.delete_workload.side_effect = Exception("Sandbox not found")

        with pytest.raises(HTTPException) as exc_info:
            k8s_service.delete_sandbox("nonexistent-id")

        assert exc_info.value.status_code == 404

    def test_delete_sandbox_removes_managed_pvcs(self, k8s_service, mock_workload):
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.delete_workload.return_value = None

        pvc = MagicMock()
        pvc.metadata = MagicMock(name="metadata")
        pvc.metadata.name = "auto-data"
        k8s_service.k8s_client.list_pvcs.return_value = [pvc]

        k8s_service.delete_sandbox("test-sandbox-id")

        selector = (
            f"{SANDBOX_MANAGED_VOLUMES_LABEL}=server,"
            f"{SANDBOX_ID_LABEL}=test-sandbox-id"
        )
        k8s_service.k8s_client.list_pvcs.assert_called_once_with(
            k8s_service.namespace,
            label_selector=selector,
        )
        k8s_service.k8s_client.delete_pvc.assert_called_once_with(
            k8s_service.namespace,
            "auto-data",
        )

    def test_delete_sandbox_skips_pvc_cleanup_on_transient_error(self, k8s_service):
        # A non-404 delete failure must NOT trigger PVC deletion: the workload
        # is still using them.
        k8s_service.workload_provider.delete_workload.side_effect = Exception(
            "kubernetes api server unavailable"
        )

        with pytest.raises(HTTPException):
            k8s_service.delete_sandbox("flaky-id")

        k8s_service.k8s_client.list_pvcs.assert_not_called()
        k8s_service.k8s_client.delete_pvc.assert_not_called()

    def test_delete_sandbox_sweeps_orphaned_pvcs_on_404(self, k8s_service):
        # If the workload is already gone, we still sweep orphaned managed PVCs
        # so a retry after a partial cleanup heals state.
        k8s_service.workload_provider.delete_workload.side_effect = Exception(
            "Sandbox not found"
        )

        pvc = MagicMock()
        pvc.metadata = MagicMock(name="metadata")
        pvc.metadata.name = "orphan-pvc"
        k8s_service.k8s_client.list_pvcs.return_value = [pvc]

        with pytest.raises(HTTPException) as exc_info:
            k8s_service.delete_sandbox("ghost-id")
        assert exc_info.value.status_code == 404

        k8s_service.k8s_client.delete_pvc.assert_called_once_with(
            k8s_service.namespace,
            "orphan-pvc",
        )

    def test_delete_sandbox_pvc_cleanup_is_best_effort(self, k8s_service, mock_workload):
        # list_pvcs failure (e.g. 403) must not break sandbox deletion.
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.delete_workload.return_value = None
        k8s_service.k8s_client.list_pvcs.side_effect = Exception("forbidden")

        # Should not raise — workload deletion already succeeded.
        k8s_service.delete_sandbox("test-sandbox-id")
        k8s_service.k8s_client.delete_pvc.assert_not_called()


class TestEnsurePvcVolumes:
    """_ensure_pvc_volumes labels auto-created PVCs for later cleanup."""

    def _make_volume(
        self,
        claim_name: str,
        *,
        create_if_not_exists: bool = True,
        delete_on_sandbox_termination: bool = True,
    ):
        vol = MagicMock()
        vol.pvc = MagicMock()
        vol.pvc.claim_name = claim_name
        vol.pvc.create_if_not_exists = create_if_not_exists
        vol.pvc.delete_on_sandbox_termination = delete_on_sandbox_termination
        vol.pvc.storage = None
        vol.pvc.access_modes = None
        vol.pvc.storage_class = None
        return vol

    def test_opt_in_pvc_carries_managed_and_id_labels(self, k8s_service):
        # deleteOnSandboxTermination=true: PVC is labeled so cleanup will sweep it.
        k8s_service.k8s_client.get_pvc.return_value = None  # doesn't exist
        k8s_service.k8s_client.create_pvc.return_value = None

        k8s_service._ensure_pvc_volumes(
            [self._make_volume("data-claim", delete_on_sandbox_termination=True)],
            sandbox_id="sandbox-xyz",
        )

        k8s_service.k8s_client.create_pvc.assert_called_once()
        _ns, body = k8s_service.k8s_client.create_pvc.call_args.args
        labels = body.metadata.labels or {}
        assert labels.get(SANDBOX_MANAGED_VOLUMES_LABEL) == "server"
        assert labels.get(SANDBOX_ID_LABEL) == "sandbox-xyz"

    def test_opt_out_pvc_is_not_labeled_for_cleanup(self, k8s_service):
        # deleteOnSandboxTermination=false (default): PVC is created but not
        # labeled, so _cleanup_managed_pvcs will leave it alone.
        k8s_service.k8s_client.get_pvc.return_value = None
        k8s_service.k8s_client.create_pvc.return_value = None

        k8s_service._ensure_pvc_volumes(
            [self._make_volume("durable-claim", delete_on_sandbox_termination=False)],
            sandbox_id="sandbox-xyz",
        )

        k8s_service.k8s_client.create_pvc.assert_called_once()
        _ns, body = k8s_service.k8s_client.create_pvc.call_args.args
        labels = body.metadata.labels or {}
        assert SANDBOX_MANAGED_VOLUMES_LABEL not in labels
        assert SANDBOX_ID_LABEL not in labels

    def test_existing_pvc_is_not_labeled_or_modified(self, k8s_service):
        # Pre-existing PVC: get returns non-None, create must not be called.
        existing = MagicMock()
        existing.metadata.labels = {}  # unlabeled = user-managed, allow reuse
        k8s_service.k8s_client.get_pvc.return_value = existing

        k8s_service._ensure_pvc_volumes(
            [self._make_volume("existing-claim")],
            sandbox_id="sandbox-xyz",
        )

        k8s_service.k8s_client.create_pvc.assert_not_called()

    def test_existing_pvc_managed_by_other_sandbox_is_rejected_409(self, k8s_service):
        # If a PVC is already labeled as managed by another sandbox, allowing
        # this sandbox to mount it would let the owner's cleanup (label sweep
        # or ownerReference GC on CR delete/TTL) delete the volume out from
        # under us. Reject 409 instead.
        from opensandbox_server.services.constants import (
            SANDBOX_ID_LABEL,
            SANDBOX_MANAGED_VOLUMES_LABEL,
        )

        existing = MagicMock()
        existing.metadata.labels = {
            SANDBOX_MANAGED_VOLUMES_LABEL: "server",
            SANDBOX_ID_LABEL: "other-sandbox",
        }
        k8s_service.k8s_client.get_pvc.return_value = existing

        with pytest.raises(HTTPException) as exc_info:
            k8s_service._ensure_pvc_volumes(
                [self._make_volume("shared-claim", delete_on_sandbox_termination=True)],
                sandbox_id="sandbox-xyz",
            )

        assert exc_info.value.status_code == 409
        k8s_service.k8s_client.create_pvc.assert_not_called()

    def test_existing_pvc_owned_by_other_is_rejected_even_without_create_if_not_exists(
        self, k8s_service
    ):
        # ``createIfNotExists=false`` mounts a user-named PVC. If that PVC
        # happens to be labeled as managed by another sandbox, the new
        # sandbox would still be exposed to the owner's cleanup. The pre-pass
        # must catch this for ALL PVC mounts, not just create-if-not-exists.
        from opensandbox_server.services.constants import (
            SANDBOX_ID_LABEL,
            SANDBOX_MANAGED_VOLUMES_LABEL,
        )

        existing = MagicMock()
        existing.metadata.labels = {
            SANDBOX_MANAGED_VOLUMES_LABEL: "server",
            SANDBOX_ID_LABEL: "other-sandbox",
        }
        k8s_service.k8s_client.get_pvc.return_value = existing

        with pytest.raises(HTTPException) as exc_info:
            k8s_service._ensure_pvc_volumes(
                [self._make_volume("borrowed-claim", create_if_not_exists=False)],
                sandbox_id="sandbox-xyz",
            )

        assert exc_info.value.status_code == 409
        k8s_service.k8s_client.create_pvc.assert_not_called()

    def test_pre_pass_get_pvc_403_fails_closed(self, k8s_service):
        # Without 'get' on PVCs the ownership pre-pass cannot verify whether
        # the claim is already labeled by another sandbox. Silently skipping
        # would let a request with createIfNotExists=false mount someone
        # else's managed PVC, exposing it to the owner's cleanup. Refuse
        # with 500 — RBAC misconfig is a persistent server-side error, not
        # something retrying will fix.
        from kubernetes.client import ApiException

        k8s_service.k8s_client.get_pvc.side_effect = ApiException(
            status=403, reason="Forbidden"
        )

        with pytest.raises(HTTPException) as exc_info:
            k8s_service._ensure_pvc_volumes(
                [self._make_volume("unknown-owner", create_if_not_exists=False)],
                sandbox_id="sandbox-xyz",
            )

        assert exc_info.value.status_code == 500
        k8s_service.k8s_client.create_pvc.assert_not_called()

    def test_create_race_409_fails_closed_when_winner_pvc_already_gone(
        self, k8s_service
    ):
        # Winner created the PVC, our create raced and 409'd, then the
        # winner's own failure cleanup deleted the PVC before our re-fetch
        # could observe it. Proceeding would let the workload reference a
        # missing claim and fail at pod scheduling; raise 503 so the caller
        # retries and their fresh pre-pass sees the absent PVC.
        from kubernetes.client import ApiException

        # Pre-pass sees absent; create_pvc loses race with 409; re-fetch returns None.
        k8s_service.k8s_client.get_pvc.side_effect = [None, None]
        k8s_service.k8s_client.create_pvc.side_effect = ApiException(status=409, reason="AlreadyExists")

        with pytest.raises(HTTPException) as exc_info:
            k8s_service._ensure_pvc_volumes(
                [self._make_volume("contested-claim", delete_on_sandbox_termination=True)],
                sandbox_id="sandbox-b",
            )

        assert exc_info.value.status_code == 503

    def test_create_race_409_fails_closed_when_re_fetch_fails(self, k8s_service):
        # If the post-409 ownership re-check itself fails (transient/403),
        # we cannot confirm the winner didn't label the PVC for another
        # sandbox. Silently proceeding could mount hostile storage; raise
        # 503 instead so the caller can retry.
        from kubernetes.client import ApiException

        # Pre-pass sees absent; create_pvc loses race with 409; re-fetch fails.
        k8s_service.k8s_client.get_pvc.side_effect = [None, Exception("network blip")]
        k8s_service.k8s_client.create_pvc.side_effect = ApiException(status=409, reason="AlreadyExists")

        with pytest.raises(HTTPException) as exc_info:
            k8s_service._ensure_pvc_volumes(
                [self._make_volume("contested-claim", delete_on_sandbox_termination=True)],
                sandbox_id="sandbox-b",
            )

        assert exc_info.value.status_code == 503

    def test_create_race_409_re_checks_ownership_and_rejects_other_winner(
        self, k8s_service
    ):
        # Two concurrent creates: both see the PVC absent, A wins the create
        # and labels it for sandbox A, B's create raises 409. B must re-fetch
        # and apply the same managed-by-other guard, otherwise B silently
        # mounts A's labeled PVC and A's cleanup will reap it under B.
        from kubernetes.client import ApiException
        from opensandbox_server.services.constants import (
            SANDBOX_ID_LABEL,
            SANDBOX_MANAGED_VOLUMES_LABEL,
        )

        racer = MagicMock()
        racer.metadata.labels = {
            SANDBOX_MANAGED_VOLUMES_LABEL: "server",
            SANDBOX_ID_LABEL: "sandbox-a",
        }
        # First get_pvc (pre-pass) sees nothing; second (post-409) sees A's labeled PVC.
        k8s_service.k8s_client.get_pvc.side_effect = [None, racer]
        k8s_service.k8s_client.create_pvc.side_effect = ApiException(status=409, reason="AlreadyExists")

        with pytest.raises(HTTPException) as exc_info:
            k8s_service._ensure_pvc_volumes(
                [self._make_volume("contested-claim", delete_on_sandbox_termination=True)],
                sandbox_id="sandbox-b",
            )

        assert exc_info.value.status_code == 409

    def test_existing_pvc_labeled_by_same_sandbox_is_allowed(self, k8s_service):
        # Idempotent retry: same sandbox_id label on the existing PVC means
        # the previous create_sandbox attempt got partway and the user is
        # retrying. Don't 409 ourselves.
        from opensandbox_server.services.constants import (
            SANDBOX_ID_LABEL,
            SANDBOX_MANAGED_VOLUMES_LABEL,
        )

        existing = MagicMock()
        existing.metadata.labels = {
            SANDBOX_MANAGED_VOLUMES_LABEL: "server",
            SANDBOX_ID_LABEL: "sandbox-xyz",
        }
        k8s_service.k8s_client.get_pvc.return_value = existing

        k8s_service._ensure_pvc_volumes(
            [self._make_volume("retried-claim", delete_on_sandbox_termination=True)],
            sandbox_id="sandbox-xyz",
        )

        k8s_service.k8s_client.create_pvc.assert_not_called()

    def test_conflicting_delete_flag_is_rejected_400(self, k8s_service):
        # Same claim mounted twice with different delete flags must fail
        # 400 before any side effects: otherwise the first occurrence wins
        # and either leaks (opt-in then opt-out) or unexpectedly deletes
        # (opt-out then opt-in).
        with pytest.raises(HTTPException) as exc_info:
            k8s_service._ensure_pvc_volumes(
                [
                    self._make_volume("shared", delete_on_sandbox_termination=True),
                    self._make_volume("shared", delete_on_sandbox_termination=False),
                ],
                sandbox_id="sandbox-xyz",
            )

        assert exc_info.value.status_code == 400
        k8s_service.k8s_client.create_pvc.assert_not_called()

    def test_conflicting_create_flag_is_rejected_400(self, k8s_service):
        with pytest.raises(HTTPException) as exc_info:
            k8s_service._ensure_pvc_volumes(
                [
                    self._make_volume("shared", create_if_not_exists=True),
                    self._make_volume("shared", create_if_not_exists=False),
                ],
                sandbox_id="sandbox-xyz",
            )

        assert exc_info.value.status_code == 400
        k8s_service.k8s_client.create_pvc.assert_not_called()

    def test_duplicate_claim_with_matching_flags_is_allowed(self, k8s_service):
        # Mounting the same claim at multiple paths is legitimate as long as
        # the provisioning flags agree.
        k8s_service.k8s_client.get_pvc.return_value = None
        k8s_service.k8s_client.create_pvc.return_value = None

        k8s_service._ensure_pvc_volumes(
            [
                self._make_volume("shared", delete_on_sandbox_termination=True),
                self._make_volume("shared", delete_on_sandbox_termination=True),
            ],
            sandbox_id="sandbox-xyz",
        )

        # Only one PVC is created despite two mounts (dedup by claim).
        k8s_service.k8s_client.create_pvc.assert_called_once()


class TestCreateSandboxPvcFailureCleanup:
    """Auto-created PVCs are swept when sandbox creation fails before returning."""

    def _request_with_pvc(self):
        from opensandbox_server.api.schema import (
            CreateSandboxRequest,
            ImageSpec,
            PVC,
            ResourceLimits,
            Volume,
        )

        return CreateSandboxRequest(
            image=ImageSpec(uri="python:3.11"),
            entrypoint=["/bin/bash", "-c", "sleep 3600"],
            timeout=3600,
            resourceLimits=ResourceLimits(root={"cpu": "1", "memory": "1Gi"}),
            volumes=[
                Volume(
                    name="data",
                    mountPath="/data",
                    pvc=PVC(
                        claimName="auto-data",
                        createIfNotExists=True,
                        deleteOnSandboxTermination=True,
                    ),
                ),
            ],
        )

    @pytest.mark.asyncio
    async def test_pool_ref_with_volumes_rejected_before_pvc_creation(
        self, k8s_service
    ):
        # poolRef + volumes is invalid. The provider also rejects this but
        # raises after _ensure_pvc_volumes has already auto-created and
        # labeled PVCs — combined with the pessimistic workload_left_alive
        # flag, the finally would skip cleanup and leak storage. The outer
        # service must validate up-front, before any side effects.
        from opensandbox_server.api.schema import (
            CreateSandboxRequest,
            ImageSpec,
            PVC,
            ResourceLimits,
            Volume,
        )

        request = CreateSandboxRequest(
            image=ImageSpec(uri="python:3.11"),
            entrypoint=["/bin/bash", "-c", "sleep 3600"],
            timeout=3600,
            resourceLimits=ResourceLimits(root={"cpu": "1", "memory": "1Gi"}),
            extensions={"poolRef": "my-pool"},
            volumes=[
                Volume(
                    name="data",
                    mountPath="/data",
                    pvc=PVC(
                        claimName="auto-data",
                        createIfNotExists=True,
                        deleteOnSandboxTermination=True,
                    ),
                ),
            ],
        )

        with pytest.raises(HTTPException) as exc_info:
            await k8s_service.create_sandbox(request)

        assert exc_info.value.status_code == 400
        # No PVC was created, no workload was attempted, no cleanup needed.
        k8s_service.k8s_client.create_pvc.assert_not_called()
        k8s_service.workload_provider.create_workload.assert_not_called()

    @pytest.mark.asyncio
    async def test_shared_pvc_mixed_read_only_rejected_before_pvc_creation(
        self, k8s_service
    ):
        from opensandbox_server.api.schema import (
            CreateSandboxRequest,
            ImageSpec,
            PVC,
            ResourceLimits,
            Volume,
        )

        request = CreateSandboxRequest(
            image=ImageSpec(uri="python:3.11"),
            entrypoint=["/bin/bash", "-c", "sleep 3600"],
            timeout=3600,
            resourceLimits=ResourceLimits(root={"cpu": "1", "memory": "1Gi"}),
            volumes=[
                Volume(
                    name="shared-rw",
                    mountPath="/data/rw",
                    readOnly=False,
                    pvc=PVC(
                        claimName="auto-data",
                        createIfNotExists=True,
                        deleteOnSandboxTermination=True,
                    ),
                ),
                Volume(
                    name="shared-ro",
                    mountPath="/data/ro",
                    readOnly=True,
                    pvc=PVC(
                        claimName="auto-data",
                        createIfNotExists=True,
                        deleteOnSandboxTermination=True,
                    ),
                ),
            ],
        )

        with pytest.raises(HTTPException) as exc_info:
            await k8s_service.create_sandbox(request)

        assert exc_info.value.status_code == 400
        assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PARAMETER
        assert "mixed read_only values" in exc_info.value.detail["message"]
        k8s_service.k8s_client.create_pvc.assert_not_called()
        k8s_service.workload_provider.create_workload.assert_not_called()

    @pytest.mark.asyncio
    async def test_pvc_cleanup_runs_when_wait_for_ready_rollback_reports_not_found(
        self, k8s_service, mock_workload
    ):
        # The CR can disappear between create_workload returning and the
        # readiness rollback (e.g. controller deletes it on a short sandbox
        # ``timeout`` before ``sandbox_create_timeout_seconds``). The rollback
        # then gets "not found" — same safe state as a successful delete,
        # so PVCs must still be swept.
        k8s_service.k8s_client.get_pvc.return_value = None
        k8s_service.k8s_client.create_pvc.return_value = None
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id",
            "uid": "abc",
        }
        k8s_service.workload_provider.delete_workload.side_effect = Exception(
            "BatchSandbox 'test-id' not found"
        )

        async def _raise_ready(*args, **kwargs):
            raise HTTPException(
                status_code=500,
                detail={"code": "X", "message": "pod never ready"},
            )

        with patch.object(k8s_service, "_wait_for_sandbox_ready", _raise_ready):
            with patch.object(k8s_service, "_cleanup_managed_pvcs") as mock_cleanup:
                with pytest.raises(HTTPException):
                    await k8s_service.create_sandbox(self._request_with_pvc())

        mock_cleanup.assert_called_once()

    @pytest.mark.asyncio
    async def test_pvc_cleanup_runs_when_wait_for_ready_fails(
        self, k8s_service, mock_workload
    ):
        # PVC creation succeeds, workload is created, but the pod never
        # becomes ready and _wait_for_sandbox_ready raises. The create API
        # returns no id, so the caller cannot trigger delete_sandbox — the
        # server must sweep the labeled PVC itself.
        k8s_service.k8s_client.get_pvc.return_value = None
        k8s_service.k8s_client.create_pvc.return_value = None
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id",
            "uid": "abc",
        }

        async def _raise_ready(*args, **kwargs):
            raise HTTPException(
                status_code=500,
                detail={"code": "X", "message": "pod never ready"},
            )

        with patch.object(k8s_service, "_wait_for_sandbox_ready", _raise_ready):
            with patch.object(k8s_service, "_cleanup_managed_pvcs") as mock_cleanup:
                with pytest.raises(HTTPException):
                    await k8s_service.create_sandbox(self._request_with_pvc())

        mock_cleanup.assert_called_once()

    @pytest.mark.asyncio
    async def test_pvc_cleanup_skipped_when_rollback_delete_workload_fails(
        self, k8s_service, mock_workload
    ):
        # _wait_for_sandbox_ready raises after create_workload succeeded, then
        # the rollback delete_workload also raises (e.g. transient API error).
        # The CR is still alive in the cluster — sweeping its PVCs would leave
        # a live workload referencing missing storage. ownerReference GC will
        # reclaim them once the CR is eventually deleted.
        k8s_service.k8s_client.get_pvc.return_value = None
        k8s_service.k8s_client.create_pvc.return_value = None
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id",
            "uid": "abc",
            "apiVersion": "sandbox.opensandbox.io/v1alpha1",
            "kind": "BatchSandbox",
        }
        k8s_service.workload_provider.delete_workload.side_effect = Exception(
            "transient api error"
        )

        async def _raise_ready(*args, **kwargs):
            raise HTTPException(
                status_code=500,
                detail={"code": "X", "message": "pod never ready"},
            )

        with patch.object(k8s_service, "_wait_for_sandbox_ready", _raise_ready):
            with patch.object(k8s_service, "_cleanup_managed_pvcs") as mock_cleanup:
                with pytest.raises(HTTPException):
                    await k8s_service.create_sandbox(self._request_with_pvc())

        mock_cleanup.assert_not_called()

    @pytest.mark.asyncio
    async def test_pvc_cleanup_runs_when_create_workload_raises_and_rollback_succeeds(
        self, k8s_service
    ):
        # ``create_workload`` may have created the CR before raising; on any
        # non-ValueError we attempt rollback delete_workload. When that
        # rollback succeeds (or the CR never existed), the CR is gone and
        # the labeled PVCs can be safely swept.
        k8s_service.k8s_client.get_pvc.return_value = None
        k8s_service.k8s_client.create_pvc.return_value = None
        k8s_service.workload_provider.create_workload.side_effect = Exception(
            "workload api 500"
        )
        k8s_service.workload_provider.delete_workload.return_value = None

        with patch.object(k8s_service, "_cleanup_managed_pvcs") as mock_cleanup:
            with pytest.raises(HTTPException):
                await k8s_service.create_sandbox(self._request_with_pvc())

        # Rollback ran, then PVCs were swept.
        k8s_service.workload_provider.delete_workload.assert_called_once()
        mock_cleanup.assert_called_once()

    @pytest.mark.asyncio
    async def test_pvc_cleanup_runs_when_rollback_reports_not_found(
        self, k8s_service
    ):
        # ``create_workload`` raised after the provider's internal rollback
        # already deleted the CR (e.g. BatchSandboxProvider deletes the CR
        # itself after imagePullSecret fails, then re-raises). Our outer
        # rollback then gets "not found" — that's the same safe state as a
        # successful delete, so PVCs must still be swept.
        k8s_service.k8s_client.get_pvc.return_value = None
        k8s_service.k8s_client.create_pvc.return_value = None
        k8s_service.workload_provider.create_workload.side_effect = Exception(
            "workload api 500"
        )
        k8s_service.workload_provider.delete_workload.side_effect = Exception(
            "BatchSandbox 'sandbox-x' not found"
        )

        with patch.object(k8s_service, "_cleanup_managed_pvcs") as mock_cleanup:
            with pytest.raises(HTTPException):
                await k8s_service.create_sandbox(self._request_with_pvc())

        mock_cleanup.assert_called_once()

    @pytest.mark.asyncio
    async def test_pvc_cleanup_skipped_when_create_workload_raises_and_rollback_fails(
        self, k8s_service
    ):
        # ``create_workload`` raised and the rollback delete_workload also
        # raised — CR may still be alive. Skip the PVC sweep so we don't
        # yank storage from a live workload; the user's delete_sandbox
        # retry path will reclaim PVCs once the CR is confirmed gone.
        k8s_service.k8s_client.get_pvc.return_value = None
        k8s_service.k8s_client.create_pvc.return_value = None
        k8s_service.workload_provider.create_workload.side_effect = Exception(
            "workload api 500"
        )
        k8s_service.workload_provider.delete_workload.side_effect = Exception(
            "rollback also failed"
        )

        with patch.object(k8s_service, "_cleanup_managed_pvcs") as mock_cleanup:
            with pytest.raises(HTTPException):
                await k8s_service.create_sandbox(self._request_with_pvc())

        mock_cleanup.assert_not_called()

    @pytest.mark.asyncio
    async def test_pvc_cleanup_runs_when_create_workload_raises_value_error(
        self, k8s_service
    ):
        # Provider convention: ValueError == preflight failure with no CR
        # touched (e.g. AgentSandboxProvider rejecting windows platform).
        # PVCs we just labeled are orphans we must sweep, without needing
        # rollback delete_workload.
        k8s_service.k8s_client.get_pvc.return_value = None
        k8s_service.k8s_client.create_pvc.return_value = None
        k8s_service.workload_provider.create_workload.side_effect = ValueError(
            "platform.os=windows not supported by this provider"
        )

        with patch.object(k8s_service, "_cleanup_managed_pvcs") as mock_cleanup:
            with pytest.raises(HTTPException) as exc_info:
                await k8s_service.create_sandbox(self._request_with_pvc())

        assert exc_info.value.status_code == 400
        # No rollback needed (no CR), but PVC cleanup still runs.
        k8s_service.workload_provider.delete_workload.assert_not_called()
        mock_cleanup.assert_called_once()

    @pytest.mark.asyncio
    async def test_pvc_cleanup_does_not_run_on_success(
        self, k8s_service, mock_workload
    ):
        # On the success path the caller owns the sandbox and any auto-created
        # PVCs; the server must not double-sweep them.
        k8s_service.k8s_client.get_pvc.return_value = None
        k8s_service.k8s_client.create_pvc.return_value = None
        k8s_service.workload_provider.create_workload.return_value = {
            "name": "test-id",
            "uid": "abc",
        }
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = "10.0.0.1:8080"
        k8s_service.workload_provider.get_expiration.return_value = (
            datetime.now(timezone.utc) + timedelta(hours=1)
        )

        with patch.object(k8s_service, "_cleanup_managed_pvcs") as mock_cleanup:
            await k8s_service.create_sandbox(self._request_with_pvc())

        mock_cleanup.assert_not_called()


class TestAttachPvcOwnerReferences:
    """OwnerReferences let K8s GC sweep PVCs on controller-driven CR deletion (TTL)."""

    def test_patches_each_managed_pvc_with_owner_ref(self, k8s_service):
        k8s_service._attach_pvc_owner_references(
            ["pvc-a", "pvc-b"],
            {
                "name": "sandbox-1",
                "uid": "owner-uid",
                "apiVersion": "sandbox.opensandbox.io/v1alpha1",
                "kind": "BatchSandbox",
            },
        )

        assert k8s_service.k8s_client.patch_pvc.call_count == 2
        for call in k8s_service.k8s_client.patch_pvc.call_args_list:
            ns, name, body = call.args
            assert ns == k8s_service.namespace
            owner_ref = body["metadata"]["ownerReferences"][0]
            assert owner_ref["uid"] == "owner-uid"
            assert owner_ref["name"] == "sandbox-1"
            assert owner_ref["kind"] == "BatchSandbox"
            assert owner_ref["apiVersion"] == "sandbox.opensandbox.io/v1alpha1"
            # blockOwnerDeletion=False so PVC delete failures don't stall CR delete.
            assert owner_ref["blockOwnerDeletion"] is False

    def test_no_patch_when_no_managed_pvcs(self, k8s_service):
        k8s_service._attach_pvc_owner_references([], {"name": "s", "uid": "u", "apiVersion": "g/v", "kind": "K"})
        k8s_service.k8s_client.patch_pvc.assert_not_called()

    def test_patch_failure_is_best_effort(self, k8s_service):
        # Patch failure must not propagate — the sandbox is already created.
        k8s_service.k8s_client.patch_pvc.side_effect = Exception("forbidden")

        # No exception expected
        k8s_service._attach_pvc_owner_references(
            ["pvc-a"],
            {"name": "s", "uid": "u", "apiVersion": "g/v", "kind": "K"},
        )

    def test_skips_when_owner_info_incomplete(self, k8s_service):
        # If the workload provider didn't return full owner info, we must not
        # produce a malformed ownerReference. Label-based cleanup remains.
        k8s_service._attach_pvc_owner_references(
            ["pvc-a"],
            {"name": "s", "uid": "u"},  # missing apiVersion/kind
        )
        k8s_service.k8s_client.patch_pvc.assert_not_called()


class TestListSandboxes:
    """list_sandboxes method tests"""
    
    def test_list_all_sandboxes_succeeds(self, k8s_service, mock_workload):
        k8s_service.workload_provider.list_workloads.return_value = [mock_workload]
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = "10.0.0.1:8080"
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)
        
        from opensandbox_server.api.schema import PaginationRequest
        request = ListSandboxesRequest(pagination=PaginationRequest(page=1, page_size=20))
        response = k8s_service.list_sandboxes(request)
        
        # Sandbox in items uses 'id' field
        assert len(response.items) == 1
        assert response.items[0].id == "test-sandbox-123"
        assert response.pagination.total_items == 1
    
    def test_list_sandboxes_with_pagination(self, k8s_service, mock_workload):
        # Create multiple mock workloads using mock_workload as template
        workloads = []
        for i in range(10):
            workload = {
                "metadata": {
                    "name": f"sandbox-{i}",
                    "uid": f"uid-{i}",
                    "labels": {
                        "opensandbox.io/id": f"sandbox-{i}",
                    },
                    "annotations": mock_workload["metadata"]["annotations"].copy(),
                    "creationTimestamp": datetime.now(timezone.utc).isoformat(),
                },
                "spec": {},
                "status": {},
            }
            workloads.append(workload)
        
        k8s_service.workload_provider.list_workloads.return_value = workloads
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = "10.0.0.1:8080"
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)
        
        from opensandbox_server.api.schema import PaginationRequest
        request = ListSandboxesRequest(pagination=PaginationRequest(page=1, page_size=5))
        response = k8s_service.list_sandboxes(request)
        
        assert len(response.items) == 5
        assert response.pagination.page == 1
        assert response.pagination.page_size == 5
        assert response.pagination.total_items == 10
        assert response.pagination.total_pages == 2
    
    def test_list_sandboxes_sorted_by_creation_time(self, k8s_service, mock_workload):
        # Create workloads with different creation times
        base_time = datetime.now(timezone.utc)
        workloads = []
        
        # Create sandboxes with specific creation times
        # We'll create them in random order to verify sorting works
        creation_times = [
            base_time - timedelta(hours=5),  # Oldest
            base_time - timedelta(hours=2),
            base_time - timedelta(hours=1),
            base_time - timedelta(minutes=30),
            base_time,  # Newest
        ]
        
        for i, created_at in enumerate(creation_times):
            workload = {
                "metadata": {
                    "name": f"sandbox-{i}",
                    "uid": f"uid-{i}",
                    "labels": {
                        "opensandbox.io/id": f"sandbox-{i}",
                    },
                    "annotations": mock_workload["metadata"]["annotations"].copy(),
                    "creationTimestamp": created_at.isoformat(),
                },
                "spec": {},
                "status": {},
            }
            workloads.append(workload)
        
        k8s_service.workload_provider.list_workloads.return_value = workloads
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = "10.0.0.1:8080"
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)
        
        from opensandbox_server.api.schema import PaginationRequest
        request = ListSandboxesRequest(pagination=PaginationRequest(page=1, page_size=10))
        response = k8s_service.list_sandboxes(request)
        
        # Verify all items are returned
        assert len(response.items) == 5
        
        # Verify they are sorted by creation time (newest first)
        # The order should be: index 4 (newest), 3, 2, 1, 0 (oldest)
        assert response.items[0].id == "sandbox-4"  # Newest
        assert response.items[1].id == "sandbox-3"
        assert response.items[2].id == "sandbox-2"
        assert response.items[3].id == "sandbox-1"
        assert response.items[4].id == "sandbox-0"  # Oldest
        
        # Also verify the creation times are in descending order
        for i in range(len(response.items) - 1):
            assert response.items[i].created_at >= response.items[i + 1].created_at

    def test_list_sandboxes_returns_null_platform_for_default_scheduling(self, k8s_service):
        workloads = [
            {
                "metadata": {
                    "name": "sandbox-1",
                    "uid": "uid-1",
                    "labels": {"opensandbox.io/id": "sandbox-1"},
                    "annotations": {},
                    "creationTimestamp": datetime.now(timezone.utc).isoformat(),
                },
                "spec": {
                    "template": {
                        "spec": {
                            # Default scheduling: no nodeSelector/nodeAffinity constraints.
                        }
                    }
                },
                "status": {},
            }
        ]
        k8s_service.workload_provider.list_workloads.return_value = workloads
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": "",
            "message": "Running",
            "last_transition_at": datetime.now(timezone.utc),
        }
        k8s_service.workload_provider.get_expiration.return_value = datetime.now(timezone.utc) + timedelta(hours=1)
        from opensandbox_server.api.schema import PaginationRequest
        request = ListSandboxesRequest(pagination=PaginationRequest(page=1, page_size=10))
        response = k8s_service.list_sandboxes(request)

        assert len(response.items) == 1
        assert response.items[0].platform is None

class TestRenewExpiration:
    """renew_sandbox_expiration method tests"""
    
    def test_renew_expiration_succeeds(self, k8s_service, mock_workload):
        new_expiration = datetime.now(timezone.utc) + timedelta(hours=2)
        
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        k8s_service.workload_provider.update_expiration.return_value = None
        k8s_service.workload_provider.get_expiration.return_value = new_expiration
        
        from opensandbox_server.api.schema import RenewSandboxExpirationRequest
        request = RenewSandboxExpirationRequest(expires_at=new_expiration)
        
        response = k8s_service.renew_expiration("test-sandbox-id", request)
        
        assert response.expires_at == new_expiration
        k8s_service.workload_provider.update_expiration.assert_called_once_with(
            sandbox_id="test-sandbox-id",
            namespace=k8s_service.namespace,
            expires_at=new_expiration
        )
    
    def test_renew_with_past_time_raises_error(self, k8s_service, mock_workload):
        past_time = datetime.now(timezone.utc) - timedelta(hours=1)
        
        k8s_service.workload_provider.get_workload.return_value = mock_workload
        
        from opensandbox_server.api.schema import RenewSandboxExpirationRequest
        request = RenewSandboxExpirationRequest(expires_at=past_time)
        
        with pytest.raises(HTTPException) as exc_info:
            k8s_service.renew_expiration("test-sandbox-id", request)
        
        assert exc_info.value.status_code == 400

    def test_renew_returns_409_when_sandbox_has_no_expiration(self, k8s_service):
        """Renew is rejected with 409 when sandbox has no TTL (manual cleanup)."""
        k8s_service.workload_provider.get_workload.return_value = MagicMock()
        k8s_service.workload_provider.get_expiration.return_value = None
        from opensandbox_server.api.schema import RenewSandboxExpirationRequest
        request = RenewSandboxExpirationRequest(
            expires_at=datetime.now(timezone.utc) + timedelta(hours=1)
        )

        with pytest.raises(HTTPException) as exc_info:
            k8s_service.renew_expiration("no-ttl-sandbox", request)

        assert exc_info.value.status_code == 409
        assert "does not have automatic expiration" in exc_info.value.detail["message"]
        k8s_service.workload_provider.update_expiration.assert_not_called()


class TestSignedEndpoint:
    """Test signed route token generation in get_endpoint."""

    BASE64_SECRET = "bXktdGVzdC1zZWNyZXQ="  # "my-test-secret"

    def _setup_gateway_with_secure_access(self, k8s_service, route_mode="wildcard"):
        """Helper to configure ingress gateway with secure_access on the service."""
        address = "*.sandbox.example.com" if route_mode == "wildcard" else "gateway.sandbox.example.com"
        k8s_service.ingress_config = IngressConfig(
            mode="gateway",
            gateway=GatewayConfig(
                address=address,
                route=GatewayRouteModeConfig(mode=route_mode),
            ),
            secure_access=SecureAccessConfig(
                active_key="a",
                keys=[
                    SecureAccessKey(key_id="a", key=self.BASE64_SECRET),
                ],
            ),
        )

    def test_signed_endpoint_embeds_token_in_url(self, k8s_service):
        self._setup_gateway_with_secure_access(k8s_service)
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {"annotations": {}},
        }

        endpoint = k8s_service.get_endpoint("sbx-001", 8080, expires=2000000000)

        assert endpoint.endpoint.startswith("sbx-001-8080-")
        assert endpoint.endpoint.endswith(".sandbox.example.com")
        # The signature should be embedded in the endpoint URL (right-split for hyphenated sandbox_id)
        parts = endpoint.endpoint.split(".")[0].rsplit("-", 3)
        assert len(parts) == 4  # sandbox_id-port-b36-signature

    def test_signed_endpoint_omits_secure_access_header(self, k8s_service):
        """Signed endpoint must NOT include the static SecureAccessToken header."""
        self._setup_gateway_with_secure_access(k8s_service)
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {
                "annotations": {
                    SANDBOX_SECURE_ACCESS_TOKEN_METADATA_KEY: "static-token",
                }
            },
        }

        endpoint = k8s_service.get_endpoint("sbx-001", 8080, expires=2000000000)

        if endpoint.headers:
            assert OPEN_SANDBOX_SECURE_ACCESS_HEADER not in endpoint.headers, (
                "Signed endpoint should not carry the static SecureAccessToken header"
            )

    def test_unsigned_endpoint_attaches_secure_access_header(self, k8s_service):
        """Unsigned endpoint must include the static SecureAccessToken header."""
        self._setup_gateway_with_secure_access(k8s_service)
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {
                "annotations": {
                    SANDBOX_SECURE_ACCESS_TOKEN_METADATA_KEY: "static-token",
                }
            },
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = Endpoint(
            endpoint="sbx-001-8080.sandbox.example.com",
        )

        endpoint = k8s_service.get_endpoint("sbx-001", 8080)

        assert endpoint.headers is not None
        assert endpoint.headers[OPEN_SANDBOX_SECURE_ACCESS_HEADER] == "static-token"

    def test_signed_endpoint_header_mode(self, k8s_service):
        self._setup_gateway_with_secure_access(k8s_service, route_mode="header")
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {"annotations": {}},
        }

        endpoint = k8s_service.get_endpoint("sbx-001", 8080, expires=2000000000)

        assert endpoint.endpoint == "gateway.sandbox.example.com"
        assert endpoint.headers is not None
        ingress_val = endpoint.headers.get("OpenSandbox-Ingress-To", "")
        # Header value should contain the signed route: {sid}-{port}-{b36}-{sig}
        parts = ingress_val.rsplit("-", 3)
        assert len(parts) == 4

    def test_signed_endpoint_uri_mode(self, k8s_service):
        self._setup_gateway_with_secure_access(k8s_service, route_mode="uri")
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {"annotations": {}},
        }

        endpoint = k8s_service.get_endpoint("sbx-001", 8080, expires=2000000000)

        # URI mode: endpoint is {addr}/{sid}/{port}/{b36}/{sig}
        assert "x2qxvk" in endpoint.endpoint
        assert "0ff8cd39a" in endpoint.endpoint

    def test_expires_negative_rejected(self, k8s_service):
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {"annotations": {}},
        }

        with pytest.raises(HTTPException) as exc:
            k8s_service.get_endpoint("sbx-001", 8080, expires=-1)

        assert exc.value.status_code == 400

    def test_expires_in_past_rejected(self, k8s_service):
        """A timestamp in the past must be rejected."""
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {"annotations": {}},
        }

        with pytest.raises(HTTPException) as exc:
            k8s_service.get_endpoint("sbx-001", 8080, expires=1000000)

        assert exc.value.status_code == 400
        assert "must be greater than current time" in exc.value.detail["message"]

    def test_expires_without_gateway_rejected(self, k8s_service):
        """No ingress config at all."""
        k8s_service.ingress_config = None
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {"annotations": {}},
        }

        with pytest.raises(HTTPException) as exc:
            k8s_service.get_endpoint("sbx-001", 8080, expires=2000000000)

        assert exc.value.status_code == 400
        assert "gateway" in exc.value.detail["message"].lower()

    def test_expires_without_secure_access_keys_rejected(self, k8s_service):
        """Gateway configured but no secure_access keys block."""
        k8s_service.ingress_config = IngressConfig(
            mode="gateway",
            gateway=GatewayConfig(
                address="*.example.com",
                route=GatewayRouteModeConfig(mode="wildcard"),
            ),
        )
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {"annotations": {}},
        }

        with pytest.raises(HTTPException) as exc:
            k8s_service.get_endpoint("sbx-001", 8080, expires=2000000000)

        assert exc.value.status_code == 400
        assert "secure_access" in exc.value.detail["message"].lower()

    def test_unsigned_endpoint_no_expires(self, k8s_service):
        """Without expires, the unsigned endpoint should be returned."""
        self._setup_gateway_with_secure_access(k8s_service)
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {"annotations": {}},
        }
        k8s_service.workload_provider.get_endpoint_info.return_value = Endpoint(
            endpoint="sbx-001-8080.sandbox.example.com",
        )

        endpoint = k8s_service.get_endpoint("sbx-001", 8080)

        assert endpoint.endpoint == "sbx-001-8080.sandbox.example.com"

    def test_signed_endpoint_different_expires_produces_different_endpoints(self, k8s_service):
        self._setup_gateway_with_secure_access(k8s_service)
        k8s_service.workload_provider.get_workload.return_value = {
            "metadata": {"annotations": {}},
        }

        ep1 = k8s_service.get_endpoint("sbx-001", 8080, expires=2000000000)
        ep2 = k8s_service.get_endpoint("sbx-001", 8080, expires=2000000500)

        assert ep1.endpoint != ep2.endpoint


class TestPatchSandboxMetadata:
    """Verify patch_sandbox_metadata builds the JSON merge-patch body correctly
    and uses the API server's PATCH response (not a cache-prone re-fetch)."""

    @staticmethod
    def _workload(labels: dict) -> dict:
        return {
            "metadata": {
                "name": "sandbox-sbx-001",
                "labels": dict(labels),
                "creationTimestamp": datetime(2026, 1, 1, tzinfo=timezone.utc),
            },
            "spec": {},
            "status": {"conditions": []},
        }

    @staticmethod
    def _stub_provider_status(k8s_service) -> None:
        k8s_service.workload_provider.get_status.return_value = {
            "state": "Running",
            "reason": None,
            "message": None,
            "last_transition_at": None,
        }
        k8s_service.workload_provider.get_expiration.return_value = None

    def test_patch_body_sends_null_for_deleted_keys(self, k8s_service):
        initial = {"opensandbox.io/id": "sbx-001", "team": "infra", "env": "dev"}
        patched = {"opensandbox.io/id": "sbx-001", "env": "stage"}

        k8s_service.workload_provider.get_workload.return_value = self._workload(initial)
        k8s_service.workload_provider.patch_labels.return_value = self._workload(patched)
        self._stub_provider_status(k8s_service)

        k8s_service.patch_sandbox_metadata("sbx-001", {"env": "stage", "team": None})

        k8s_service.workload_provider.patch_labels.assert_called_once()
        body_labels = k8s_service.workload_provider.patch_labels.call_args.kwargs["labels"]
        assert body_labels["env"] == "stage"
        assert body_labels["team"] is None
        assert body_labels["opensandbox.io/id"] == "sbx-001"

    def test_returns_sandbox_from_patch_response(self, k8s_service):
        """The PATCH response is authoritative; re-reading via get_workload
        could hit a stale informer cache."""
        initial = {"opensandbox.io/id": "sbx-001", "env": "dev"}
        patched = {"opensandbox.io/id": "sbx-001", "env": "stage"}

        k8s_service.workload_provider.get_workload.return_value = self._workload(initial)
        k8s_service.workload_provider.patch_labels.return_value = self._workload(patched)
        self._stub_provider_status(k8s_service)

        sandbox = k8s_service.patch_sandbox_metadata("sbx-001", {"env": "stage"})

        assert sandbox.metadata == {"env": "stage"}
        # Pre-patch read only; no second get_workload after patch_labels.
        assert k8s_service.workload_provider.get_workload.call_count == 1