81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from host_agent.client import HostAgentAPIError
|
|
from host_agent.config import HostAgentConfig
|
|
from host_agent.enrollment import resolve_host_identity
|
|
from host_agent.identity import HostIdentityStore
|
|
|
|
|
|
def _config(**overrides) -> HostAgentConfig:
|
|
values = {
|
|
"control_plane_url": "https://control.example",
|
|
}
|
|
values.update(overrides)
|
|
return HostAgentConfig(**values)
|
|
|
|
|
|
class _RecordingEnrollmentClient:
|
|
def __init__(self, *, host_id: str = "host-cloud-a") -> None:
|
|
self.host_id = host_id
|
|
self.calls: list[dict] = []
|
|
|
|
def enroll_host(self, **payload):
|
|
from cloud.internal_api.models import HostEnrollmentResponse
|
|
|
|
self.calls.append(payload)
|
|
return HostEnrollmentResponse(host_id=self.host_id)
|
|
|
|
|
|
class _RejectingEnrollmentClient:
|
|
def enroll_host(self, **payload):
|
|
raise HostAgentAPIError(401, "unauthorized")
|
|
|
|
|
|
def test_fresh_install_directly_enrolls(tmp_path) -> None:
|
|
identity_store = HostIdentityStore(tmp_path / "identity.json")
|
|
client = _RecordingEnrollmentClient()
|
|
|
|
resolved = resolve_host_identity(
|
|
_config(),
|
|
identity_store=identity_store,
|
|
client=client,
|
|
)
|
|
|
|
assert resolved.host_id == "host-cloud-a"
|
|
assert len(client.calls) == 1
|
|
assert identity_store.load().host_id == "host-cloud-a"
|
|
|
|
|
|
def test_self_service_rejection_propagates_as_api_error(tmp_path) -> None:
|
|
identity_store = HostIdentityStore(tmp_path / "identity.json")
|
|
client = _RejectingEnrollmentClient()
|
|
|
|
try:
|
|
resolve_host_identity(
|
|
_config(),
|
|
identity_store=identity_store,
|
|
client=client,
|
|
)
|
|
except HostAgentAPIError as exc:
|
|
assert exc.status_code == 401
|
|
else:
|
|
raise AssertionError("expected HostAgentAPIError to propagate")
|
|
assert identity_store.load().host_id is None
|
|
|
|
|
|
def test_existing_cached_identity_skips_enrollment(tmp_path) -> None:
|
|
identity_store = HostIdentityStore(tmp_path / "identity.json")
|
|
identity_store.complete(identity_store.load_or_create(), "host-cloud-a")
|
|
|
|
class ExplodingClient:
|
|
def enroll_host(self, **payload):
|
|
raise AssertionError("cached identity must skip enrollment")
|
|
|
|
resolved = resolve_host_identity(
|
|
_config(),
|
|
identity_store=identity_store,
|
|
client=ExplodingClient(), # type: ignore[arg-type]
|
|
)
|
|
|
|
assert resolved.host_id == "host-cloud-a"
|