1  # Copyright 2026 Alibaba Group Holding Ltd.
  2  #
  3  # Licensed under the Apache License, Version 2.0 (the "License");
  4  # you may not use this file except in compliance with the License.
  5  # You may obtain a copy of the License at
  6  #
  7  #     http://www.apache.org/licenses/LICENSE-2.0
  8  #
  9  # Unless required by applicable law or agreed to in writing, software
 10  # distributed under the License is distributed on an "AS IS" BASIS,
 11  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12  # See the License for the specific language governing permissions and
 13  # limitations under the License.
 14  
 15  from unittest.mock import patch
 16  
 17  import pytest
 18  from fastapi import HTTPException
 19  
 20  from opensandbox_server.api.schema import PlatformSpec
 21  from opensandbox_server.services.docker_windows_profile import (
 22      apply_windows_runtime_host_config_defaults,
 23      inject_windows_resource_limits_env,
 24      inject_windows_user_ports,
 25      resolve_docker_platform,
 26      validate_windows_resource_limits,
 27      validate_windows_runtime_prerequisites,
 28  )
 29  
 30  
 31  def test_apply_windows_runtime_host_config_defaults_injects_required_defaults():
 32      host_cfg = {"network_mode": "host"}
 33      updated = apply_windows_runtime_host_config_defaults(host_cfg, "sbx-1")
 34  
 35      assert updated["devices"] == ["/dev/kvm", "/dev/net/tun"]
 36      assert "NET_ADMIN" in updated["cap_add"]
 37      assert "NET_RAW" in updated["cap_add"]
 38      assert "opensandbox-win-oem-sbx-1:/oem:rw" in updated["binds"]
 39  
 40  
 41  def test_apply_windows_runtime_host_config_defaults_removes_net_admin_from_cap_drop():
 42      host_cfg = {
 43          "cap_drop": ["NET_ADMIN", "NET_RAW", "SYS_ADMIN"],
 44          "binds": ["/tmp/data:/data:rw"],
 45      }
 46      updated = apply_windows_runtime_host_config_defaults(host_cfg, "sbx-2")
 47  
 48      assert "NET_ADMIN" not in (updated.get("cap_drop") or [])
 49      assert "NET_RAW" not in (updated.get("cap_drop") or [])
 50      assert "SYS_ADMIN" in (updated.get("cap_drop") or [])
 51      assert "/tmp/data:/data:rw" in updated["binds"]
 52  
 53  
 54  def test_validate_windows_runtime_prerequisites_reports_missing_devices():
 55      with patch("opensandbox_server.services.docker_windows_profile.os.path.exists", return_value=False):
 56          with pytest.raises(HTTPException) as exc_info:
 57              validate_windows_runtime_prerequisites()
 58  
 59      assert exc_info.value.status_code == 400
 60      assert "/dev/kvm" in exc_info.value.detail["message"]
 61      assert "/dev/net/tun" in exc_info.value.detail["message"]
 62  
 63  
 64  def test_resolve_docker_platform_skips_windows_profile():
 65      assert (
 66          resolve_docker_platform(PlatformSpec(os="windows", arch="amd64"))
 67          is None
 68      )
 69  
 70  
 71  def test_resolve_docker_platform_preserves_linux_profile():
 72      assert (
 73          resolve_docker_platform(PlatformSpec(os="linux", arch="arm64"))
 74          == "linux/arm64"
 75      )
 76  
 77  
 78  def test_inject_windows_user_ports_appends_when_missing():
 79      env = ["VERSION=11"]
 80      updated = inject_windows_user_ports(env, ["44772", "8080/tcp"])
 81      assert "VERSION=11" in updated
 82      assert "USER_PORTS=44772,8080" in updated
 83  
 84  
 85  def test_inject_windows_user_ports_merges_existing_value():
 86      env = ["USER_PORTS=3389,44772", "VERSION=11"]
 87      updated = inject_windows_user_ports(env, ["44772", "8080"])
 88      assert "USER_PORTS=3389,44772,8080" in updated
 89  
 90  
 91  def test_inject_windows_resource_limits_env_from_resource_limits():
 92      env = ["VERSION=11"]
 93      updated = inject_windows_resource_limits_env(
 94          env,
 95          {
 96              "cpu": "4",
 97              "memory": "8G",
 98              "disk": "64G",
 99          },
100      )
101      assert "VERSION=11" in updated
102      assert "CPU_CORES=4" in updated
103      assert "RAM_SIZE=8G" in updated
104      assert "DISK_SIZE=64G" in updated
105  
106  
107  def test_inject_windows_resource_limits_env_overrides_existing_env():
108      env = ["CPU_CORES=1", "RAM_SIZE=1G", "DISK_SIZE=20G", "VERSION=11"]
109      updated = inject_windows_resource_limits_env(
110          env,
111          {
112              "cpu": "2500m",
113              "memory": "8192Mi",
114              "storage": "100Gi",
115          },
116      )
117      assert "CPU_CORES=3" in updated
118      assert "RAM_SIZE=8G" in updated
119      assert "DISK_SIZE=100G" in updated
120      assert "CPU_CORES=1" not in updated
121      assert "RAM_SIZE=1G" not in updated
122      assert "DISK_SIZE=20G" not in updated
123  
124  
125  def test_validate_windows_resource_limits_accepts_minimum_values():
126      validate_windows_resource_limits(
127          {
128              "cpu": "2",
129              "memory": "4G",
130              "disk": "64G",
131          }
132      )
133  
134  
135  @pytest.mark.parametrize(
136      ("resource_limits", "expected_message_fragment"),
137      [
138          ({"cpu": "1500m"}, "resourceLimits.cpu >= 2"),
139          ({"memory": "3G"}, "resourceLimits.memory >= 4G"),
140          ({"disk": "63G"}, "resourceLimits.disk"),
141      ],
142  )
143  def test_validate_windows_resource_limits_rejects_values_below_minimum(
144      resource_limits, expected_message_fragment
145  ):
146      with pytest.raises(HTTPException) as exc_info:
147          validate_windows_resource_limits(resource_limits)
148      assert exc_info.value.status_code == 400
149      assert expected_message_fragment in exc_info.value.detail["message"]
150  
151