135 lines
4.3 KiB
Python
135 lines
4.3 KiB
Python
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)
|
|
|
|
|
|
@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",
|
|
}
|
|
)
|
|
|
|
|
|
@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":[]}]',
|
|
),
|
|
],
|
|
)
|
|
def test_load_control_config_rejects_invalid_credentials(
|
|
name: str,
|
|
value: str,
|
|
) -> None:
|
|
with pytest.raises(CloudConfigurationError, match="credentials") 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})
|