Spaces:
Running
Running
File size: 4,400 Bytes
ca89449 0fa27d4 b7329cf ca89449 b7329cf ca89449 b7329cf ca89449 b7329cf ca89449 b7329cf ca89449 b7329cf ca89449 0fa27d4 b7329cf 0fa27d4 b7329cf 0fa27d4 b7329cf 0fa27d4 bf3d642 b7329cf bf3d642 b7329cf bf3d642 b7329cf bf3d642 b7329cf bf3d642 b7329cf bf3d642 b7329cf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | from __future__ import annotations
import click
import headroom.cli.wrap as wrap_cli
class _Manifest:
profile = "default"
preset = "persistent-service"
supervisor_kind = "service"
health_url = "http://127.0.0.1:8787/readyz"
def test_ensure_proxy_recovers_matching_persistent_deployment(monkeypatch) -> None:
calls: list[str] = []
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: False)
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: False)
monkeypatch.setattr(
"headroom.install.supervisors.start_supervisor",
lambda manifest: calls.append(f"start:{manifest.profile}"),
)
monkeypatch.setattr(
"headroom.install.runtime.wait_ready", lambda manifest, timeout_seconds=45: True
)
monkeypatch.setattr(
wrap_cli,
"_start_proxy",
lambda *args, **kwargs: (_ for _ in ()).throw(
AssertionError("ephemeral proxy should not start")
),
)
result = wrap_cli._ensure_proxy(8787, False)
assert result is None
assert calls == ["start:default"]
def test_ensure_proxy_recovers_persistent_deployment_when_socket_is_bound(monkeypatch) -> None:
calls: list[str] = []
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: True)
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: False)
monkeypatch.setattr(
"headroom.install.supervisors.start_supervisor",
lambda manifest: calls.append(f"start:{manifest.profile}"),
)
monkeypatch.setattr(
"headroom.install.runtime.wait_ready", lambda manifest, timeout_seconds=45: True
)
result = wrap_cli._ensure_proxy(8787, False)
assert result is None
assert calls == ["start:default"]
def test_ensure_proxy_rejects_unhealthy_persistent_deployment(monkeypatch) -> None:
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: True)
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: False)
monkeypatch.setattr(wrap_cli, "_recover_persistent_proxy", lambda port: False)
try:
wrap_cli._ensure_proxy(8787, False)
except click.ClickException as exc:
assert "is not healthy" in str(exc)
else:
raise AssertionError("expected unhealthy persistent deployment to raise")
def test_ensure_proxy_falls_back_when_persistent_manifest_is_stale(monkeypatch) -> None:
calls: list[str] = []
monkeypatch.setattr(wrap_cli, "_check_proxy", lambda port: False)
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: False)
monkeypatch.setattr(wrap_cli, "_recover_persistent_proxy", lambda port: False)
monkeypatch.setattr(wrap_cli, "_start_proxy", lambda *args, **kwargs: calls.append("start"))
result = wrap_cli._ensure_proxy(8787, False)
assert result is None
assert calls == ["start"]
def test_find_persistent_manifest_prefers_default_profile(monkeypatch) -> None:
class DefaultManifest:
profile = "default"
port = 8787
class OtherManifest:
profile = "custom"
port = 8787
monkeypatch.setattr(
"headroom.install.state.list_manifests",
lambda: [OtherManifest(), DefaultManifest()],
)
manifest = wrap_cli._find_persistent_manifest(8787)
assert manifest.profile == "default"
def test_recover_persistent_proxy_reuses_healthy_deployment(monkeypatch) -> None:
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: _Manifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: True)
assert wrap_cli._recover_persistent_proxy(8787) is True
def test_recover_persistent_proxy_warns_for_task_deployment(monkeypatch) -> None:
class TaskManifest(_Manifest):
supervisor_kind = "task"
monkeypatch.setattr(wrap_cli, "_find_persistent_manifest", lambda port: TaskManifest())
monkeypatch.setattr("headroom.install.health.probe_ready", lambda url: False)
assert wrap_cli._recover_persistent_proxy(8787) is False
|