q792602257andClaude Opus 4.6
d00ada67a5
Acquire an exclusive, non-blocking filelock on the identity state directory as the first action of create_application(), before resolve_host_identity() or any enrollment/heartbeat side effect. A second process against the same identity_path exits immediately with InstanceAlreadyRunningError naming the lock path; the lock releases automatically on any process exit (including SIGKILL) via OS-level advisory locking, and explicitly during run_async()'s shutdown finally block. filelock is promoted from transitive to direct dependency (version unchanged at 3.29.7). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
149 lines
4.5 KiB
Python
149 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from host_agent import cli
|
|
from host_agent.instance_lock import InstanceAlreadyRunningError
|
|
from host_agent.local_account import LocalAccountStore
|
|
|
|
|
|
class _RecordingApplication:
|
|
def __init__(self) -> None:
|
|
self.ran = False
|
|
|
|
def run(self) -> None:
|
|
self.ran = True
|
|
|
|
|
|
def _base_env(tmp_path) -> dict[str, str]:
|
|
return {
|
|
"HOST_AGENT_CONTROL_PLANE_URL": "https://cloud.example",
|
|
"HOST_AGENT_LOCAL_ACCOUNT_PATH": str(tmp_path / "account.json"),
|
|
"HOST_AGENT_IDENTITY_PATH": str(tmp_path / "identity.json"),
|
|
}
|
|
|
|
|
|
def _set_env(monkeypatch, tmp_path) -> None:
|
|
for key, value in _base_env(tmp_path).items():
|
|
monkeypatch.setenv(key, value)
|
|
|
|
|
|
def _patch_create_application(monkeypatch) -> dict:
|
|
captured: dict = {}
|
|
|
|
def fake_create_application(*, config=None, **kwargs):
|
|
captured["config"] = config
|
|
app = _RecordingApplication()
|
|
captured["app"] = app
|
|
return app
|
|
|
|
monkeypatch.setattr(cli, "create_application", fake_create_application)
|
|
return captured
|
|
|
|
|
|
def test_existing_account_fast_path_skips_prompting(monkeypatch, tmp_path) -> None:
|
|
_set_env(monkeypatch, tmp_path)
|
|
LocalAccountStore(tmp_path / "account.json").create(
|
|
"operator", "correct horse battery staple"
|
|
)
|
|
|
|
def fail_input(prompt: str = "") -> str:
|
|
raise AssertionError("must not prompt when a local account already exists")
|
|
|
|
monkeypatch.setattr("builtins.input", fail_input)
|
|
captured = _patch_create_application(monkeypatch)
|
|
|
|
cli.main([])
|
|
|
|
assert captured["app"].ran is True
|
|
assert captured["config"].display_name == "operator"
|
|
|
|
|
|
def test_interactive_first_run_prompts_and_creates_account(
|
|
monkeypatch, tmp_path
|
|
) -> None:
|
|
_set_env(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
|
inputs = iter(["operator"])
|
|
monkeypatch.setattr("builtins.input", lambda prompt="": next(inputs))
|
|
passwords = iter(["hunter2", "hunter2"])
|
|
monkeypatch.setattr("getpass.getpass", lambda prompt="": next(passwords))
|
|
captured = _patch_create_application(monkeypatch)
|
|
|
|
cli.main([])
|
|
|
|
assert captured["app"].ran is True
|
|
account = LocalAccountStore(tmp_path / "account.json").load()
|
|
assert account is not None
|
|
assert account.username == "operator"
|
|
|
|
|
|
def test_non_interactive_without_account_exits_with_clear_error(
|
|
monkeypatch, tmp_path, capsys
|
|
) -> None:
|
|
_set_env(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(sys.stdin, "isatty", lambda: False)
|
|
captured = _patch_create_application(monkeypatch)
|
|
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
cli.main([])
|
|
|
|
assert exc_info.value.code == 1
|
|
assert "setup" in capsys.readouterr().err
|
|
assert "app" not in captured
|
|
assert LocalAccountStore(tmp_path / "account.json").load() is None
|
|
|
|
|
|
def test_setup_subcommand_creates_account(monkeypatch, tmp_path) -> None:
|
|
_set_env(monkeypatch, tmp_path)
|
|
inputs = iter(["operator"])
|
|
monkeypatch.setattr("builtins.input", lambda prompt="": next(inputs))
|
|
passwords = iter(["hunter2", "hunter2"])
|
|
monkeypatch.setattr("getpass.getpass", lambda prompt="": next(passwords))
|
|
captured = _patch_create_application(monkeypatch)
|
|
|
|
cli.main(["setup"])
|
|
|
|
assert "app" not in captured
|
|
account = LocalAccountStore(tmp_path / "account.json").load()
|
|
assert account is not None
|
|
assert account.username == "operator"
|
|
|
|
|
|
def test_setup_subcommand_refuses_overwrite_without_confirmation(
|
|
monkeypatch, tmp_path
|
|
) -> None:
|
|
_set_env(monkeypatch, tmp_path)
|
|
store = LocalAccountStore(tmp_path / "account.json")
|
|
original = store.create("operator", "original-password")
|
|
monkeypatch.setattr("builtins.input", lambda prompt="": "n")
|
|
|
|
cli.main(["setup"])
|
|
|
|
assert store.load() == original
|
|
|
|
|
|
def test_duplicate_instance_exits_with_clear_error(
|
|
monkeypatch, tmp_path, capsys
|
|
) -> None:
|
|
_set_env(monkeypatch, tmp_path)
|
|
LocalAccountStore(tmp_path / "account.json").create(
|
|
"operator", "correct horse battery staple"
|
|
)
|
|
lock_path = tmp_path / "identity_dir" / "host_agent.lock"
|
|
|
|
def raising_create_application(*, config=None, **kwargs):
|
|
raise InstanceAlreadyRunningError(lock_path)
|
|
|
|
monkeypatch.setattr(cli, "create_application", raising_create_application)
|
|
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
cli.main([])
|
|
|
|
assert exc_info.value.code == 1
|
|
err = capsys.readouterr().err
|
|
assert "another Host Agent instance" in err
|
|
assert str(lock_path) in err
|