agentic-mobile-control/tests/test_cloud_control_config.py

from __future__ import annotations

import pytest

from cloud.auth import (
    ConfiguredBearerAuthProvider,
    NullAuthProvider,
    create_auth_provider,
)
from cloud.control_config import (
    CloudConfigurationError,
    CloudControlConfig,
    load_control_config,
)


def test_load_control_config_uses_safe_local_defaults() -> None:
    assert load_control_config({}) == CloudControlConfig()


def test_load_control_config_parses_deployment_values() -> None:
    config = load_control_config(
        {
            "CLOUD_ENVIRONMENT": "production",
            "CLOUD_DATABASE_URL": "postgresql+psycopg://cloud@db/cloud",
            "CLOUD_SCHEDULER_INTERVAL_SECONDS": "2.5",
            "CLOUD_LEASE_REAPER_INTERVAL_SECONDS": "7",
            "CLOUD_LEASE_DURATION_SECONDS": "90",
            "CLOUD_MAX_TASK_ATTEMPTS": "5",
            "CLOUD_PUBLIC_CREDENTIALS_JSON": (
                '[{"principal_id":"sdk","token":"sdk-secret",'
                '"scopes":["tasks:submit","tasks:read"]}]'
            ),
            "CLOUD_HOST_CREDENTIALS_JSON": (
                '[{"principal_id":"agent-a","token":"host-secret",'
                '"host_id":"host-a","scopes":["host:agent"]}]'
            ),
        }
    )

    assert config.environment == "production"
    assert config.database_url.startswith("postgresql+psycopg://")
    assert config.scheduler_interval_seconds == 2.5
    assert config.max_task_attempts == 5
    assert len(config.credentials) == 2
    assert config.credentials[1].host_id == "host-a"
    assert "sdk-secret" not in repr(config)


def test_load_control_config_parses_enrollment_credentials() -> None:
    config = load_control_config(
        {
            "CLOUD_ENROLLMENT_TOKENS_JSON": (
                '[{"principal_id":"installer-a","token":"one-time-enrollment-secret"}]'
            )
        }
    )

    assert len(config.enrollment_credentials) == 1
    assert config.enrollment_credentials[0].principal_id == "installer-a"
    assert "one-time-enrollment-secret" not in repr(config)


@pytest.mark.parametrize(
    "environment,database_url",
    [("invalid", "sqlite:///test.db"), ("local", "mysql://db/cloud")],
)
def test_load_control_config_rejects_invalid_environment_or_database(
    environment: str,
    database_url: str,
) -> None:
    with pytest.raises(CloudConfigurationError):
        load_control_config(
            {
                "CLOUD_ENVIRONMENT": environment,
                "CLOUD_DATABASE_URL": database_url,
            }
        )


def test_load_control_config_rejects_unsafe_production_anonymous_mode() -> None:
    with pytest.raises(CloudConfigurationError, match="production"):
        load_control_config(
            {
                "CLOUD_ENVIRONMENT": "production",
                "CLOUD_DATABASE_URL": "postgresql://db/cloud",
                "CLOUD_ALLOW_INSECURE_ANONYMOUS": "true",
                "CLOUD_PUBLIC_CREDENTIALS_JSON": (
                    '[{"principal_id":"sdk","token":"secret","scopes":[]}]'
                ),
            }
        )


def test_load_control_config_rejects_missing_production_credentials() -> None:
    with pytest.raises(CloudConfigurationError, match="credential"):
        load_control_config(
            {
                "CLOUD_ENVIRONMENT": "production",
                "CLOUD_DATABASE_URL": "postgresql://db/cloud",
            }
        )


def test_load_control_config_parses_user_session_settings() -> None:
    config = load_control_config(
        {
            "CLOUD_USER_SESSION_IDLE_SECONDS": "600",
            "CLOUD_USER_SESSION_ABSOLUTE_SECONDS": "1200",
            "CLOUD_LOGIN_FAILURE_LIMIT": "3",
            "CLOUD_LOGIN_FAILURE_WINDOW_SECONDS": "60",
            "CLOUD_LOGIN_BLOCK_SECONDS": "90",
            "CLOUD_SESSION_COOKIE_SECURE": "true",
            "CLOUD_TRUST_PROXY_HEADERS": "true",
        }
    )

    assert config.user_session_idle_seconds == 600
    assert config.user_session_absolute_seconds == 1200
    assert config.login_failure_limit == 3
    assert config.login_block_seconds == 90
    assert config.session_cookie_secure is True
    assert config.trust_proxy_headers is True


def test_load_control_config_rejects_unsafe_user_session_ttls() -> None:
    with pytest.raises(CloudConfigurationError, match="ABSOLUTE"):
        load_control_config(
            {
                "CLOUD_USER_SESSION_IDLE_SECONDS": "1200",
                "CLOUD_USER_SESSION_ABSOLUTE_SECONDS": "600",
            }
        )


def test_production_requires_secure_user_session_cookie() -> None:
    with pytest.raises(CloudConfigurationError, match="secure user session"):
        load_control_config(
            {
                "CLOUD_ENVIRONMENT": "production",
                "CLOUD_DATABASE_URL": "postgresql://db/cloud",
                "CLOUD_PUBLIC_CREDENTIALS_JSON": (
                    '[{"principal_id":"sdk","token":"secret","scopes":[]}]'
                ),
                "CLOUD_SESSION_COOKIE_SECURE": "false",
            }
        )


@pytest.mark.parametrize(
    "name,value",
    [
        ("CLOUD_PUBLIC_CREDENTIALS_JSON", "not-json"),
        ("CLOUD_PUBLIC_CREDENTIALS_JSON", "{}"),
        (
            "CLOUD_HOST_CREDENTIALS_JSON",
            '[{"principal_id":"agent","token":"secret","scopes":[]}]',
        ),
        (
            "CLOUD_ENROLLMENT_TOKENS_JSON",
            '[{"principal_id":"installer","token":123}]',
        ),
    ],
)
def test_load_control_config_rejects_invalid_credentials(
    name: str,
    value: str,
) -> None:
    with pytest.raises(CloudConfigurationError, match="credential") as error:
        load_control_config({name: value})
    assert "secret" not in str(error.value)


def test_auth_provider_requires_explicit_anonymous_override() -> None:
    secure_provider = create_auth_provider([], allow_insecure_anonymous=False)
    insecure_provider = create_auth_provider([], allow_insecure_anonymous=True)

    assert isinstance(secure_provider, ConfiguredBearerAuthProvider)
    request = type("Request", (), {"headers": {}})()
    assert secure_provider.authenticate(request) is None
    assert isinstance(insecure_provider, NullAuthProvider)


@pytest.mark.parametrize(
    "name,value",
    [
        ("CLOUD_SCHEDULER_INTERVAL_SECONDS", "0"),
        ("CLOUD_LEASE_DURATION_SECONDS", "invalid"),
        ("CLOUD_MAX_TASK_ATTEMPTS", "-1"),
    ],
)
def test_load_control_config_rejects_invalid_numeric_values(
    name: str,
    value: str,
) -> None:
    with pytest.raises(CloudConfigurationError):
        load_control_config({name: value})