steven_li
29845657f5
新增ipaddress模块导入以支持IP地址处理， 添加DEPLOY_DIRECT_PUBLIC_HOST_BIND_IP环境变量配置， 实现IP地址验证、直接URL构建和端口分配功能， 当基础域名是IP地址时自动使用直接绑定模式， 支持IPv4和IPv6地址格式并添加相应参数传递
92 lines
3.5 KiB
Python
92 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
SERVER_PATH = Path(__file__).resolve().parents[1] / "server.py"
|
|
|
|
|
|
def _load_server_module():
|
|
spec = importlib.util.spec_from_file_location("deploy_control_server_public_url_tests", SERVER_PATH)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_create_instance_uses_direct_host_port_url_when_base_domain_is_ip(monkeypatch) -> None:
|
|
server = _load_server_module()
|
|
commands: list[list[str]] = []
|
|
record: dict[str, Any] = {
|
|
"instance_id": "urldebug",
|
|
"container_name": "app-instance-urldebug",
|
|
"host_port": 20005,
|
|
"public_url": "http://172.19.207.40:20005",
|
|
}
|
|
lookups = iter([None, None, record])
|
|
|
|
monkeypatch.setattr(server, "PUBLIC_BASE_DOMAIN", "172.19.207.40")
|
|
monkeypatch.setattr(server, "PUBLIC_PORT", 8088)
|
|
monkeypatch.setattr(server, "get_registry_record", lambda **_kwargs: next(lookups))
|
|
monkeypatch.setattr(server, "ensure_network", lambda: None)
|
|
monkeypatch.setattr(server, "ensure_proxy", lambda: None)
|
|
monkeypatch.setattr(server, "wait_for_backend", lambda _record: None)
|
|
monkeypatch.setattr(server, "pick_instance_host_port", lambda _instance_id: 20005)
|
|
|
|
def capture_command(args: list[str], **_kwargs: Any) -> str:
|
|
commands.append(args)
|
|
return ""
|
|
|
|
monkeypatch.setattr(server, "run_command", capture_command)
|
|
|
|
server.create_or_get_instance({
|
|
"username": "urldebug",
|
|
"password": "secret",
|
|
"instance_id": "urldebug",
|
|
})
|
|
|
|
create_command = commands[0]
|
|
assert create_command[create_command.index("--host-port") + 1] == "20005"
|
|
assert create_command[create_command.index("--host-bind-ip") + 1] == "0.0.0.0"
|
|
assert create_command[create_command.index("--public-url") + 1] == "http://172.19.207.40:20005"
|
|
assert create_command[create_command.index("--instance-host") + 1] == "urldebug.172.19.207.40"
|
|
|
|
|
|
def test_create_instance_keeps_router_url_when_base_domain_is_dns(monkeypatch) -> None:
|
|
server = _load_server_module()
|
|
commands: list[list[str]] = []
|
|
record: dict[str, Any] = {
|
|
"instance_id": "urldebug",
|
|
"container_name": "app-instance-urldebug",
|
|
"host_port": 20005,
|
|
"public_url": "https://urldebug.apps.example.com",
|
|
}
|
|
lookups = iter([None, None, record])
|
|
|
|
monkeypatch.setattr(server, "PUBLIC_SCHEME", "https")
|
|
monkeypatch.setattr(server, "PUBLIC_BASE_DOMAIN", "apps.example.com")
|
|
monkeypatch.setattr(server, "PUBLIC_PORT", 443)
|
|
monkeypatch.setattr(server, "get_registry_record", lambda **_kwargs: next(lookups))
|
|
monkeypatch.setattr(server, "ensure_network", lambda: None)
|
|
monkeypatch.setattr(server, "ensure_proxy", lambda: None)
|
|
monkeypatch.setattr(server, "wait_for_backend", lambda _record: None)
|
|
monkeypatch.setattr(server, "pick_instance_host_port", lambda _instance_id: 20005)
|
|
|
|
def capture_command(args: list[str], **_kwargs: Any) -> str:
|
|
commands.append(args)
|
|
return ""
|
|
|
|
monkeypatch.setattr(server, "run_command", capture_command)
|
|
|
|
server.create_or_get_instance({
|
|
"username": "urldebug",
|
|
"password": "secret",
|
|
"instance_id": "urldebug",
|
|
})
|
|
|
|
create_command = commands[0]
|
|
assert "--host-port" not in create_command
|
|
assert create_command[create_command.index("--public-url") + 1] == "https://urldebug.apps.example.com"
|