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>
81 lines
2.2 KiB
Python
81 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from host_agent.instance_lock import (
|
|
InstanceAlreadyRunningError,
|
|
InstanceLock,
|
|
)
|
|
|
|
|
|
def test_acquire_succeeds_when_free(tmp_path) -> None:
|
|
lock = InstanceLock(tmp_path)
|
|
lock.acquire()
|
|
assert lock.lock_path == tmp_path / "host_agent.lock"
|
|
lock.release()
|
|
|
|
|
|
def test_acquire_raises_when_already_held(tmp_path) -> None:
|
|
first = InstanceLock(tmp_path)
|
|
first.acquire()
|
|
try:
|
|
second = InstanceLock(tmp_path)
|
|
with pytest.raises(InstanceAlreadyRunningError) as info:
|
|
second.acquire()
|
|
assert info.value.lock_path == tmp_path / "host_agent.lock"
|
|
assert "another Host Agent instance" in str(info.value)
|
|
finally:
|
|
first.release()
|
|
|
|
|
|
def test_release_then_reacquire_succeeds(tmp_path) -> None:
|
|
lock = InstanceLock(tmp_path)
|
|
lock.acquire()
|
|
lock.release()
|
|
# Same handle can reacquire after release.
|
|
lock.acquire()
|
|
lock.release()
|
|
# A fresh handle against the same path also succeeds.
|
|
other = InstanceLock(tmp_path)
|
|
other.acquire()
|
|
other.release()
|
|
|
|
|
|
def test_release_is_idempotent_and_safe_before_acquire(tmp_path) -> None:
|
|
lock = InstanceLock(tmp_path)
|
|
# Releasing before acquire must be a no-op, not raise.
|
|
lock.release()
|
|
lock.acquire()
|
|
lock.release()
|
|
lock.release()
|
|
|
|
|
|
def test_independent_paths_never_contend(tmp_path) -> None:
|
|
dir_a = tmp_path / "identity-a"
|
|
dir_b = tmp_path / "identity-b"
|
|
lock_a = InstanceLock(dir_a)
|
|
lock_a.acquire()
|
|
try:
|
|
lock_b = InstanceLock(dir_b)
|
|
lock_b.acquire()
|
|
lock_b.release()
|
|
finally:
|
|
lock_a.release()
|
|
|
|
|
|
def test_context_manager_releases_on_exit(tmp_path) -> None:
|
|
with InstanceLock(tmp_path):
|
|
second = InstanceLock(tmp_path)
|
|
with pytest.raises(InstanceAlreadyRunningError):
|
|
second.acquire()
|
|
# After the context exits, a fresh handle can acquire.
|
|
InstanceLock(tmp_path).acquire()
|
|
|
|
|
|
def test_state_directory_is_created_if_missing(tmp_path) -> None:
|
|
nested = tmp_path / "nested" / "state"
|
|
lock = InstanceLock(nested)
|
|
assert nested.exists()
|
|
lock.acquire()
|
|
lock.release()
|