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_default() -> None: assert load_host_agent_config({}) == HostAgentConfig( control_plane_url="https://amcp.home.jerryyan.top", enrollment_managed=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_disabled_and_do_not_trigger_validation() -> None: config = load_host_agent_config({}) assert config.console_enabled is False 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 def test_console_enabled_with_default_loopback_bind_passes() -> None: config = load_host_agent_config({"HOST_AGENT_CONSOLE_ENABLED": "true"}) assert config.console_enabled is True assert config.console_bind_host == "127.0.0.1" @pytest.mark.parametrize("bind_host", ["127.0.0.1", "localhost", "::1"]) def test_console_enabled_with_loopback_bind_host_passes(bind_host: str) -> None: config = load_host_agent_config( { "HOST_AGENT_CONSOLE_ENABLED": "true", "HOST_AGENT_CONSOLE_BIND_HOST": bind_host, } ) assert config.console_bind_host == bind_host def test_console_enabled_with_non_loopback_bind_without_opt_in_raises() -> None: with pytest.raises(HostAgentConfigurationError): load_host_agent_config( { "HOST_AGENT_CONSOLE_ENABLED": "true", "HOST_AGENT_CONSOLE_BIND_HOST": "0.0.0.0", } ) def test_console_enabled_with_non_loopback_bind_with_opt_in_succeeds() -> None: config = load_host_agent_config( { "HOST_AGENT_CONSOLE_ENABLED": "true", "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_disabled_with_non_loopback_bind_does_not_raise() -> None: config = load_host_agent_config({"HOST_AGENT_CONSOLE_BIND_HOST": "0.0.0.0"}) assert config.console_enabled is False assert config.console_bind_host == "0.0.0.0" def test_console_env_vars_parse_numeric_and_bool_fields() -> None: config = load_host_agent_config( { "HOST_AGENT_CONSOLE_ENABLED": "1", "HOST_AGENT_CONSOLE_PORT": "9001", "HOST_AGENT_CONSOLE_SESSION_TTL_SECONDS": "3600", "HOST_AGENT_CONSOLE_HISTORY_LIMIT": "50", } ) assert config.console_enabled is True 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)