254 lines
7.6 KiB
Python
254 lines
7.6 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from host_agent.config import (
|
|
HostAgentConfigurationError,
|
|
HostAgentConfig,
|
|
load_host_agent_config,
|
|
)
|
|
|
|
|
|
def test_load_host_agent_config_uses_managed_cloud_defaults() -> None:
|
|
config = load_host_agent_config({})
|
|
|
|
assert config == HostAgentConfig(
|
|
control_plane_url="https://amcp.home.jerryyan.top",
|
|
enrollment_managed=True,
|
|
)
|
|
assert config.ai_planner_transport == "cloud"
|
|
|
|
|
|
def test_load_host_agent_config_allows_explicit_direct_planner_transport() -> None:
|
|
config = load_host_agent_config({"AI_PLANNER_TRANSPORT": "direct"})
|
|
|
|
assert config.ai_planner_transport == "direct"
|
|
|
|
|
|
def test_load_host_agent_config_supports_local_mode() -> None:
|
|
config = load_host_agent_config({"HOST_AGENT_MODE": "local"})
|
|
|
|
assert config.mode == "local"
|
|
assert config.control_plane_url == ""
|
|
assert config.enrollment_managed is False
|
|
assert config.ai_planner_transport == "direct"
|
|
assert config.dependency_supervisor_enabled is True
|
|
assert config.appium_supervised is True
|
|
|
|
|
|
def test_load_host_agent_config_parses_poll_and_retry_values() -> None:
|
|
config = load_host_agent_config(
|
|
{
|
|
"HOST_AGENT_CONTROL_PLANE_URL": "https://cloud.example/v1/",
|
|
"HOST_AGENT_HEARTBEAT_INTERVAL_SECONDS": "10",
|
|
"HOST_AGENT_POLL_TIMEOUT_SECONDS": "15",
|
|
"HOST_AGENT_RETRY_BACKOFF_SECONDS": "2",
|
|
"HOST_AGENT_MAX_RETRY_BACKOFF_SECONDS": "20",
|
|
}
|
|
)
|
|
|
|
assert config.control_plane_url == "https://cloud.example/v1"
|
|
assert config.poll_timeout_seconds == 15
|
|
assert config.max_retry_backoff_seconds == 20
|
|
|
|
|
|
def test_load_host_agent_config_uses_direct_enrollment(tmp_path) -> None:
|
|
identity_path = tmp_path / "host_identity.json"
|
|
config = load_host_agent_config(
|
|
{
|
|
"HOST_AGENT_CONTROL_PLANE_URL": "https://cloud.example",
|
|
"HOST_AGENT_IDENTITY_PATH": str(identity_path),
|
|
"HOST_AGENT_DISPLAY_NAME": "Edge Mac",
|
|
}
|
|
)
|
|
|
|
assert config.host_id == ""
|
|
assert config.token == ""
|
|
assert config.identity_path == identity_path
|
|
assert config.enrollment_managed is True
|
|
assert config.display_name == "Edge Mac"
|
|
|
|
|
|
def test_static_credential_environment_variables_are_ignored() -> None:
|
|
config = load_host_agent_config(
|
|
{
|
|
"HOST_AGENT_HOST_ID": "legacy-host",
|
|
"HOST_AGENT_TOKEN": "legacy-token",
|
|
"HOST_AGENT_ENROLLMENT_TOKEN": "legacy-enrollment-token",
|
|
}
|
|
)
|
|
|
|
assert config.host_id == ""
|
|
assert config.token == ""
|
|
assert config.enrollment_managed is True
|
|
|
|
|
|
def test_fresh_install_defaults_local_account_path() -> None:
|
|
config = load_host_agent_config(
|
|
{"HOST_AGENT_CONTROL_PLANE_URL": "https://cloud.example"}
|
|
)
|
|
|
|
assert config.enrollment_managed is True
|
|
assert config.local_account_path == Path("tasks/host_local_account.json")
|
|
|
|
|
|
def test_local_account_path_can_be_overridden(tmp_path) -> None:
|
|
account_path = tmp_path / "account.json"
|
|
config = load_host_agent_config(
|
|
{
|
|
"HOST_AGENT_CONTROL_PLANE_URL": "https://cloud.example",
|
|
"HOST_AGENT_LOCAL_ACCOUNT_PATH": str(account_path),
|
|
}
|
|
)
|
|
|
|
assert config.local_account_path == account_path
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"overrides",
|
|
[
|
|
{"HOST_AGENT_CONTROL_PLANE_URL": "ftp://cloud.example"},
|
|
{"HOST_AGENT_POLL_TIMEOUT_SECONDS": "0"},
|
|
{
|
|
"HOST_AGENT_RETRY_BACKOFF_SECONDS": "10",
|
|
"HOST_AGENT_MAX_RETRY_BACKOFF_SECONDS": "5",
|
|
},
|
|
],
|
|
)
|
|
def test_load_host_agent_config_rejects_invalid_values(
|
|
overrides: dict[str, str],
|
|
) -> None:
|
|
with pytest.raises(HostAgentConfigurationError):
|
|
load_host_agent_config(overrides)
|
|
|
|
|
|
def test_console_defaults_are_loopback_bound() -> None:
|
|
config = load_host_agent_config({})
|
|
|
|
assert config.console_bind_host == "127.0.0.1"
|
|
assert config.console_port == 8765
|
|
assert config.console_allow_non_loopback is False
|
|
assert config.console_session_ttl_seconds == 43200.0
|
|
assert config.console_history_limit == 200
|
|
|
|
|
|
@pytest.mark.parametrize("bind_host", ["127.0.0.1", "localhost", "::1"])
|
|
def test_console_loopback_bind_host_passes(bind_host: str) -> None:
|
|
config = load_host_agent_config(
|
|
{
|
|
"HOST_AGENT_CONSOLE_BIND_HOST": bind_host,
|
|
}
|
|
)
|
|
|
|
assert config.console_bind_host == bind_host
|
|
|
|
|
|
def test_console_non_loopback_bind_without_opt_in_raises() -> None:
|
|
with pytest.raises(HostAgentConfigurationError):
|
|
load_host_agent_config(
|
|
{
|
|
"HOST_AGENT_CONSOLE_BIND_HOST": "0.0.0.0",
|
|
}
|
|
)
|
|
|
|
|
|
def test_console_non_loopback_bind_with_opt_in_succeeds() -> None:
|
|
config = load_host_agent_config(
|
|
{
|
|
"HOST_AGENT_CONSOLE_BIND_HOST": "0.0.0.0",
|
|
"HOST_AGENT_CONSOLE_ALLOW_NON_LOOPBACK": "true",
|
|
}
|
|
)
|
|
|
|
assert config.console_bind_host == "0.0.0.0"
|
|
assert config.console_allow_non_loopback is True
|
|
|
|
|
|
def test_console_env_vars_parse_numeric_and_bool_fields() -> None:
|
|
config = load_host_agent_config(
|
|
{
|
|
"HOST_AGENT_CONSOLE_PORT": "9001",
|
|
"HOST_AGENT_CONSOLE_SESSION_TTL_SECONDS": "3600",
|
|
"HOST_AGENT_CONSOLE_HISTORY_LIMIT": "50",
|
|
}
|
|
)
|
|
|
|
assert config.console_port == 9001
|
|
assert config.console_session_ttl_seconds == 3600
|
|
assert config.console_history_limit == 50
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"overrides",
|
|
[
|
|
{"HOST_AGENT_CONSOLE_PORT": "0"},
|
|
{"HOST_AGENT_CONSOLE_SESSION_TTL_SECONDS": "-1"},
|
|
{"HOST_AGENT_CONSOLE_HISTORY_LIMIT": "0"},
|
|
],
|
|
)
|
|
def test_console_numeric_fields_reject_invalid_values(
|
|
overrides: dict[str, str],
|
|
) -> None:
|
|
with pytest.raises(HostAgentConfigurationError):
|
|
load_host_agent_config(overrides)
|
|
|
|
|
|
def test_dependency_supervisor_defaults_to_disabled() -> None:
|
|
config = load_host_agent_config({})
|
|
|
|
assert config.dependency_supervisor_enabled is False
|
|
assert config.appium_supervised is False
|
|
assert config.appium_host == "127.0.0.1"
|
|
assert config.appium_port == 4723
|
|
assert config.dependency_restart_max_attempts == 5
|
|
|
|
|
|
def test_dependency_supervisor_env_vars_parse_bool_and_numeric_fields() -> None:
|
|
config = load_host_agent_config(
|
|
{
|
|
"HOST_AGENT_DEPENDENCY_SUPERVISOR_ENABLED": "true",
|
|
"HOST_AGENT_APPIUM_SUPERVISED": "1",
|
|
"HOST_AGENT_APPIUM_HOST": "0.0.0.0",
|
|
"HOST_AGENT_APPIUM_PORT": "4724",
|
|
"HOST_AGENT_DEPENDENCY_RESTART_MAX_ATTEMPTS": "8",
|
|
}
|
|
)
|
|
|
|
assert config.dependency_supervisor_enabled is True
|
|
assert config.appium_supervised is True
|
|
assert config.appium_host == "0.0.0.0"
|
|
assert config.appium_port == 4724
|
|
assert config.dependency_restart_max_attempts == 8
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"overrides",
|
|
[
|
|
{"HOST_AGENT_APPIUM_PORT": "0"},
|
|
{"HOST_AGENT_DEPENDENCY_RESTART_MAX_ATTEMPTS": "-1"},
|
|
],
|
|
)
|
|
def test_dependency_supervisor_numeric_fields_reject_invalid_values(
|
|
overrides: dict[str, str],
|
|
) -> None:
|
|
with pytest.raises(HostAgentConfigurationError):
|
|
load_host_agent_config(overrides)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"setting,value",
|
|
[
|
|
("HOST_AGENT_RUNTIME_SUPERVISED", "true"),
|
|
("HOST_AGENT_RUNTIME_HOST", "127.0.0.1"),
|
|
("HOST_AGENT_RUNTIME_PORT", "8000"),
|
|
],
|
|
)
|
|
def test_removed_runtime_supervision_settings_are_rejected(
|
|
setting: str,
|
|
value: str,
|
|
) -> None:
|
|
with pytest.raises(HostAgentConfigurationError, match="standalone Runtime service"):
|
|
load_host_agent_config({setting: value})
|