Disable sandbox mesh for K8S regions

koyeb/sandbox/test_utils.py

@@ -1,6 +1,14 @@
 import unittest
+from unittest.mock import patch
 
-from koyeb.sandbox.utils import create_docker_source
+from koyeb.api.models.deployment_definition_type import DeploymentDefinitionType
+from koyeb.api.models.deployment_mesh import DeploymentMesh
+from koyeb.api.models.docker_source import DockerSource
+from koyeb.sandbox.utils import (
+    build_env_vars,
+    create_deployment_definition,
+    create_docker_source,
+)
 
 
 class TestCreateDockerSource(unittest.TestCase):
@@ -60,5 +68,41 @@ def test_privileged_and_registry_secret_still_work(self):
         self.assertEqual(ds.command, "serve")
 
 
+class TestCreateDeploymentDefinition(unittest.TestCase):
+    """Tests for sandbox deployment definition mesh configuration."""
+
+    def _create_definition(self, enable_mesh=None):
+        return create_deployment_definition(
+            name="sandbox",
+            docker_source=DockerSource(image="myimage"),
+            env_vars=build_env_vars({}),
+            instance_type="micro",
+            enable_mesh=enable_mesh,
+        )
+
+    def test_mesh_auto_by_default(self):
+        definition = self._create_definition()
+        self.assertEqual(definition.type, DeploymentDefinitionType.SANDBOX)
+        self.assertEqual(definition.mesh, DeploymentMesh.DEPLOYMENT_MESH_AUTO)
+
+    def test_mesh_enabled_when_requested(self):
+        definition = self._create_definition(enable_mesh=True)
+        self.assertEqual(definition.mesh, DeploymentMesh.DEPLOYMENT_MESH_ENABLED)
+
+    def test_mesh_disabled_when_requested(self):
+        definition = self._create_definition(enable_mesh=False)
+        self.assertEqual(definition.mesh, DeploymentMesh.DEPLOYMENT_MESH_DISABLED)
+
+    @patch.dict("os.environ", {"KOYEB_K8S_REGION": "par"})
+    def test_koyeb_k8s_region_disables_mesh(self):
+        definition = self._create_definition(enable_mesh=True)
+        self.assertEqual(definition.mesh, DeploymentMesh.DEPLOYMENT_MESH_DISABLED)
+
+    @patch.dict("os.environ", {"KOYEB_K8S_REGION": ""})
+    def test_empty_koyeb_k8s_region_disables_mesh(self):
+        definition = self._create_definition(enable_mesh=True)
+        self.assertEqual(definition.mesh, DeploymentMesh.DEPLOYMENT_MESH_DISABLED)
+
+
 if __name__ == "__main__":
     unittest.main()

koyeb/sandbox/utils.py

@@ -9,7 +9,7 @@
 import os
 import shlex
 from dataclasses import dataclass
-from typing import Any, Callable, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
 
 from koyeb.api import ApiClient, Configuration
 from koyeb.api.api import (
@@ -38,6 +38,9 @@
 from koyeb.api.models.docker_source import DockerSource
 from koyeb.api.models.proxy_port_protocol import ProxyPortProtocol
 
+if TYPE_CHECKING:
+    from .executor_client import ConnectionInfo
+
 # Setup logging
 logger = logging.getLogger(__name__)
 
@@ -385,12 +388,13 @@ def create_deployment_definition(
     scalings = [DeploymentScaling(min=min_scale, max=1, targets=targets)]
 
     # Set mesh configuration
-    mesh = DeploymentMesh.DEPLOYMENT_MESH_AUTO
-    if enable_mesh is None:
+    if os.getenv("KOYEB_K8S_REGION") is not None:
+        mesh = DeploymentMesh.DEPLOYMENT_MESH_DISABLED
+    elif enable_mesh is None:
         mesh = DeploymentMesh.DEPLOYMENT_MESH_AUTO
     elif not enable_mesh:
         mesh = DeploymentMesh.DEPLOYMENT_MESH_DISABLED
-    elif enable_mesh:
+    else:
         mesh = DeploymentMesh.DEPLOYMENT_MESH_ENABLED
 
     return DeploymentDefinition(
@@ -518,7 +522,7 @@ async def wrapper(self, *args, **kwargs):
 
 
 def create_sandbox_client(
-    conn_info: Optional['ConnectionInfo'],
+    conn_info: Optional["ConnectionInfo"],
     existing_client: Optional[Any] = None,
 ) -> Any:
     """