hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

test: keep the invariant tests for the config-read caches

Browse Source 
Trim the picked test files to the tests that pin an invariant a regression
would break: 115070 keeps the shared-object-mutation-poisons-cache test and
one signature-invalidation test (drops 4 that re-test yaml parse/missing
behaviour or duplicate invalidation); 117440 keeps the blocked-read-completes
test and the hook-timeout-reads-config-once test (drops 7 restating the
clamp/fallback contract already covered by test_plugins.py). Adds one test
for the tuple memo: under a writer thread flipping config.yaml, a lock-free
reader never sees a sig paired with another publish's value.
This commit is contained in:
kshitijk4poor
2026-09-22 14:25:58 +05:30
committed by kshitij
parent aeafa28342
commit bd6ff8b4a8
2 changed files with 25 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
tests/hermes_cli/test_config_lock_free_cache_hit.py
View File

@@ -79,108 +79,6 @@ def test_cached_read_completes_while_another_thread_holds_the_config_lock(config
)
def test_deepcopy_read_also_completes_under_a_held_lock(config_home):
from hermes_cli import config as cfgmod
cfgmod.load_config()
holding = threading.Event()
release = threading.Event()
def _hold():
with cfgmod._CONFIG_LOCK:
holding.set()
release.wait(timeout=30)
thread = threading.Thread(target=_hold, daemon=True)
thread.start()
try:
assert holding.wait(timeout=10)
started = time.perf_counter()
cfg = cfgmod.load_config()
elapsed = time.perf_counter() - started
finally:
release.set()
thread.join(timeout=10)
assert cfg["agent"]["max_turns"] == 500
assert elapsed < MAX_BLOCKED_READ_SECS
def test_fast_path_still_sees_a_changed_config_file(config_home):
"""Freshness is not sacrificed for the lock-free read."""
from hermes_cli import config as cfgmod
assert cfgmod.load_config_readonly()["agent"]["max_turns"] == 500
path = config_home / "config.yaml"
path.write_text(
path.read_text(encoding="utf-8").replace("max_turns: 500", "max_turns: 123"),
encoding="utf-8",
)
_bump_mtime(path)
assert cfgmod.load_config_readonly()["agent"]["max_turns"] == 123
def test_load_config_still_returns_an_isolated_object(config_home):
"""The deepcopy contract must survive the fast path."""
from hermes_cli import config as cfgmod
first = cfgmod.load_config()
first["agent"]["max_turns"] = -1
assert cfgmod.load_config()["agent"]["max_turns"] == 500
assert cfgmod.load_config_readonly()["agent"]["max_turns"] == 500
def test_concurrent_readers_and_a_writer_never_see_a_torn_config(config_home):
"""The fast path reads a tuple that is replaced wholesale, never mutated."""
from hermes_cli import config as cfgmod
cfgmod.load_config_readonly()
stop = threading.Event()
errors: list[BaseException] = []
seen: list[int] = []
def _writer():
try:
while not stop.is_set():
cfg = cfgmod.load_config()
cfg.setdefault("agent", {})["max_turns"] = 500
cfgmod.save_config(cfg)
except BaseException as exc: # noqa: BLE001
errors.append(exc)
def _reader():
try:
for _ in range(300):
cfg = cfgmod.load_config_readonly()
# Every observation must be a complete, well-formed config.
assert isinstance(cfg, dict)
assert isinstance(cfg.get("agent"), dict)
seen.append(int(cfg["agent"]["max_turns"]))
except BaseException as exc: # noqa: BLE001
errors.append(exc)
writer = threading.Thread(target=_writer, daemon=True)
readers = [threading.Thread(target=_reader, daemon=True) for _ in range(4)]
writer.start()
for r in readers:
r.start()
for r in readers:
r.join(timeout=60)
stop.set()
writer.join(timeout=30)
assert not errors, f"torn/failed read under concurrency: {errors[0]!r}"
assert seen, "readers observed nothing"
# ---------------------------------------------------------------------------
# Hook-callback timeout memoization.
# ---------------------------------------------------------------------------
def test_hook_timeout_does_not_read_config_on_every_invocation(config_home):
from hermes_cli import config as cfgmod
from hermes_cli import plugins as pluginsmod
@@ -209,73 +107,33 @@ def test_hook_timeout_does_not_read_config_on_every_invocation(config_home):
)
def test_hook_timeout_picks_up_a_changed_config(config_home):
def test_hook_timeout_memo_never_pairs_a_new_sig_with_an_old_value(config_home):
"""The memo is published as ONE ``(sig, value)`` tuple: with a writer thread flipping the config
between two timeouts, every lock-free read must see sig and value from the SAME publish."""
from hermes_cli import plugins as pluginsmod
from hermes_cli.config import _load_config_cache_sig, get_config_path
pluginsmod._reset_hook_callback_timeout_cache()
assert pluginsmod._resolve_hook_callback_timeout() == 30.0
def write(v):
(config_home / "config.yaml").write_text(f"plugins:\n hook_callback_timeout: {v}\n", encoding="utf-8")
_bump_mtime(get_config_path())
return pluginsmod._resolve_hook_callback_timeout(), _load_config_cache_sig(get_config_path())[1]
path = config_home / "config.yaml"
path.write_text(
path.read_text(encoding="utf-8").replace(
"hook_callback_timeout: 30", "hook_callback_timeout: 45"
),
encoding="utf-8",
)
_bump_mtime(path)
sig_of = {}
for v in (5, 7):
value, sig = write(v)
assert value == float(v)
sig_of[sig] = value
stop = threading.Event()
assert pluginsmod._resolve_hook_callback_timeout() == 45.0
def writer():
while not stop.is_set():
write(5); write(7)
@pytest.mark.parametrize(
"raw,expected_attr",
[
("9999", "_MAX_HOOK_CALLBACK_TIMEOUT_SECS"),
("-5", "_HOOK_CALLBACK_TIMEOUT_SECS"),
("'not-a-number'", "_HOOK_CALLBACK_TIMEOUT_SECS"),
],
)
def test_hook_timeout_validation_survives_memoization(config_home, raw, expected_attr):
"""Clamping and the fallback must not be lost to the cache."""
import re
from hermes_cli import plugins as pluginsmod
path = config_home / "config.yaml"
path.write_text(
re.sub(
r" hook_callback_timeout: .*",
f" hook_callback_timeout: {raw}",
path.read_text(encoding="utf-8"),
),
encoding="utf-8",
)
_bump_mtime(path)
pluginsmod._reset_hook_callback_timeout_cache()
assert pluginsmod._resolve_hook_callback_timeout() == getattr(
pluginsmod, expected_attr
)
def test_hook_timeout_zero_disables_and_is_cached(config_home):
"""``<= 0`` means 'no threaded timeout' -- a falsy value the memo must
still return rather than treating as 'unresolved'."""
import re
from hermes_cli import plugins as pluginsmod
path = config_home / "config.yaml"
path.write_text(
re.sub(
r" hook_callback_timeout: .*",
" hook_callback_timeout: 0",
path.read_text(encoding="utf-8"),
),
encoding="utf-8",
)
_bump_mtime(path)
pluginsmod._reset_hook_callback_timeout_cache()
assert pluginsmod._resolve_hook_callback_timeout() == 0.0
assert pluginsmod._resolve_hook_callback_timeout() == 0.0
t = threading.Thread(target=writer, daemon=True); t.start()
try:
for _ in range(2000):
sig, value = pluginsmod._HOOK_TIMEOUT_CACHE
if sig in sig_of:
assert sig_of[sig] == value, "memo published a new sig with a stale value"
finally:
stop.set(); t.join(timeout=5)

53
tests/test_load_yaml_file_readonly.py
View File

@@ -1,7 +1,6 @@
"""``load_yaml_file_readonly`` re-parses only when the file signature changes."""
import os
import pytest
from utils import load_yaml_file_readonly
@@ -20,42 +19,6 @@ def test_cache_hit_returns_same_object_and_invalidates_on_rewrite(tmp_path):
assert second is not first
def test_parse_error_propagates_and_is_not_cached(tmp_path):
path = tmp_path / "config.yaml"
path.write_text("secrets: [unclosed\n")
with pytest.raises(Exception):
load_yaml_file_readonly(path)
path.write_text("secrets: {}\n")
os.utime(path, ns=(os.stat(path).st_mtime_ns + 1_000_000,) * 2)
assert load_yaml_file_readonly(path) == {"secrets": {}}
def test_missing_file_raises(tmp_path):
with pytest.raises(FileNotFoundError):
load_yaml_file_readonly(tmp_path / "nope.yaml")
def test_atomic_replace_invalidates_even_with_identical_mtime_and_size(tmp_path):
"""``save_config`` writes via ``atomic_yaml_write`` → a new inode; the signature must change even
when a writer pins mtime and keeps the size (the case mtime+size alone would miss)."""
from utils import atomic_replace
path = tmp_path / "config.yaml"
path.write_text("terminal: {backend: local}\n")
st = os.stat(path)
first = load_yaml_file_readonly(path)
tmp = tmp_path / "config.yaml.tmp"
tmp.write_text("terminal: {backend: lokal}\n") # same size, different content
os.utime(tmp, ns=(st.st_atime_ns, st.st_mtime_ns))
atomic_replace(tmp, path)
after = os.stat(path)
assert (after.st_mtime_ns, after.st_size) == (st.st_mtime_ns, st.st_size)
assert load_yaml_file_readonly(path) == {"terminal": {"backend": "lokal"}}
assert load_yaml_file_readonly(path) is not first
def test_callers_do_not_mutate_the_shared_cached_object(tmp_path, monkeypatch):
"""Both callers only read one section; the cached mapping must survive them byte-for-byte,
otherwise a later reader of the same file would observe another caller's edits."""
@@ -82,19 +45,3 @@ def test_callers_do_not_mutate_the_shared_cached_object(tmp_path, monkeypatch):
assert secrets == {"onepassword": {"enabled": False}}
assert cached == snapshot
assert load_yaml_file_readonly(path) is cached
def test_profile_config_edit_is_visible_on_next_scope_build(tmp_path):
"""A config.yaml edit reaches the next ``build_profile_terminal_scope`` — the cache never pins
a stale policy across the edit (the "config changes stop applying" upgrade risk)."""
from tools.terminal_scope import build_profile_terminal_scope
home = tmp_path / "profiles" / "work"
home.mkdir(parents=True)
path = home / "config.yaml"
path.write_text("terminal:\n backend: local\n")
assert build_profile_terminal_scope(home)["TERMINAL_ENV"] == "local"
path.write_text("terminal:\n backend: docker\n")
os.utime(path, ns=(os.stat(path).st_mtime_ns + 1_000_000,) * 2)
assert build_profile_terminal_scope(home)["TERMINAL_ENV"] == "docker"