test(hclib): update tests for moved/removed symbols
This commit is contained in:
@@ -192,7 +192,7 @@ class TestModelStateIncludesNamedProviders:
|
||||
)
|
||||
acp_agent = HermesACPAgent(session_manager=manager)
|
||||
|
||||
with patch("hermes_cli.models.curated_models_for_provider", return_value=[]), patch(
|
||||
with patch(
|
||||
"acp_adapter.server._named_custom_provider_catalogs",
|
||||
return_value=[("custom:ollama", "Ollama", [])],
|
||||
):
|
||||
@@ -215,9 +215,6 @@ class TestModelStateIncludesNamedProviders:
|
||||
acp_agent = HermesACPAgent(session_manager=manager)
|
||||
|
||||
with patch(
|
||||
"hermes_cli.models.curated_models_for_provider",
|
||||
return_value=[("gpt-5.4", "recommended")],
|
||||
), patch(
|
||||
"acp_adapter.server._named_custom_provider_catalogs",
|
||||
return_value=[
|
||||
(
|
||||
|
||||
@@ -84,13 +84,6 @@ class TestOpenCodeFreeModelLists:
|
||||
curated = set(_PROVIDER_MODELS["opencode-free"])
|
||||
assert set(_DEFAULT_PROVIDER_MODELS["opencode-free"]) <= curated
|
||||
|
||||
def test_every_curated_model_is_keyless(self):
|
||||
"""Every model in the opencode-free catalog must satisfy the keyless
|
||||
predicate — a paid slug here would route with no auth and 401."""
|
||||
from hermes_cli.models import _PROVIDER_MODELS, is_opencode_zen_free_model
|
||||
for mid in _PROVIDER_MODELS["opencode-free"]:
|
||||
assert is_opencode_zen_free_model(mid), mid
|
||||
|
||||
def test_delisted_ox_alpha_not_in_floor(self):
|
||||
"""x-preview-f-free was delisted by the relay 2026-08-26 (401s keyless);
|
||||
the offline floor must not offer it (#95914)."""
|
||||
|
||||
@@ -23,7 +23,6 @@ from hermes_cli.focus_view import (
|
||||
FOCUS_CONFIG_KEY,
|
||||
FOCUS_STATUSBAR_LABEL,
|
||||
FOCUS_TOOL_PROGRESS_MODE,
|
||||
effective_tool_progress_mode,
|
||||
focus_statusbar_segment,
|
||||
format_focus_status,
|
||||
format_focus_toggle_message,
|
||||
@@ -69,15 +68,6 @@ class TestComposesWithVerboseModes:
|
||||
def test_focus_on_snaps_to_the_existing_off_mode(self):
|
||||
# Focus view must reuse the tool_progress "off" path, not invent a mode.
|
||||
assert FOCUS_TOOL_PROGRESS_MODE == "off"
|
||||
for configured in ("off", "new", "all", "verbose"):
|
||||
assert effective_tool_progress_mode(True, configured) == "off"
|
||||
|
||||
@pytest.mark.parametrize("configured", ["off", "new", "all", "verbose"])
|
||||
def test_focus_off_leaves_the_configured_verbose_mode_untouched(self, configured):
|
||||
assert effective_tool_progress_mode(False, configured) == configured
|
||||
|
||||
|
||||
|
||||
|
||||
def test_new_mode_skips_consecutive_repeats_like_the_renderer(self):
|
||||
assert would_display_tool_line("new", "terminal", "terminal") is False
|
||||
|
||||
@@ -7,20 +7,10 @@ import pytest
|
||||
import hermes_cli.auth as auth
|
||||
from hermes_cli.auth import (
|
||||
NOUS_BILLING_MANAGE_SCOPE,
|
||||
nous_token_has_billing_scope,
|
||||
step_up_nous_billing_scope,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# nous_token_has_billing_scope
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# step_up_nous_billing_scope
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -28,7 +28,6 @@ from hermes_cli.local_runtime.catalog import (
|
||||
PLEASANT_FLOOR_TOK_S,
|
||||
predicted_decode_tok_s,
|
||||
recommended_entry,
|
||||
recommended_id,
|
||||
select_variant,
|
||||
)
|
||||
from hermes_cli.local_runtime.estimator import HardwareBudget
|
||||
@@ -143,7 +142,7 @@ def test_unified_never_recommends_a_below_floor_dense_model():
|
||||
predicted decode is below the pleasant floor while a resident
|
||||
alternative clears it."""
|
||||
budget = _unified(128)
|
||||
pick = recommended_id(budget)
|
||||
pick = recommended_entry(budget)[0].id
|
||||
assert pick is not None
|
||||
entry = next(e for e in CATALOG if e.id == pick)
|
||||
choice = select_variant(entry, budget)
|
||||
@@ -162,7 +161,7 @@ def test_quality_decides_where_speed_permits():
|
||||
the pick must be the highest-quality fitting entry — the axis that
|
||||
justifies carrying an editorial field at all."""
|
||||
budget = _discrete(512)
|
||||
pick = recommended_id(budget)
|
||||
pick = recommended_entry(budget)[0].id
|
||||
resident = [
|
||||
e for e in CATALOG
|
||||
if (c := select_variant(e, budget)) is not None and c.zero_spill
|
||||
|
||||
@@ -299,18 +299,6 @@ def test_ensure_model_ready_unknown_model_raises(stub_server, tmp_path):
|
||||
sup.ensure_model_ready("absent")
|
||||
|
||||
|
||||
def test_model_failures_surface_exit_code_not_retry(stub_server, tmp_path):
|
||||
"""Design: child failures surface, never auto-retry."""
|
||||
port, handler = stub_server
|
||||
handler.models = {"data": [
|
||||
{"id": "ok", "status": {"value": "loaded"}},
|
||||
{"id": "dead", "status": {"value": "failed", "exit_code": -1073741819}},
|
||||
]}
|
||||
sup = _make_supervisor(tmp_path, port)
|
||||
failures = sup.model_failures()
|
||||
assert failures == {"dead": -1073741819}
|
||||
|
||||
|
||||
def test_is_idle_requires_no_busy_slots_and_zero_processing(stub_server, tmp_path):
|
||||
port, handler = stub_server
|
||||
sup = _make_supervisor(tmp_path, port)
|
||||
|
||||
@@ -113,7 +113,7 @@ def test_a_stamp_can_name_every_managed_system(
|
||||
install_tree = tmp_path / "install"
|
||||
install_tree.mkdir()
|
||||
|
||||
config_mod.stamp_install_method(managed_value, project_root=install_tree)
|
||||
(install_tree / ".install_method").write_text(managed_value + "\n", encoding="utf-8")
|
||||
|
||||
assert config_mod.detect_install_method(install_tree) == managed_value
|
||||
|
||||
|
||||
@@ -5,12 +5,10 @@ from hermes_cli.moa_config import (
|
||||
DEFAULT_MOA_AGGREGATOR,
|
||||
DEFAULT_MOA_PRESET_NAME,
|
||||
DEFAULT_MOA_REFERENCE_MODELS,
|
||||
build_moa_turn_prompt,
|
||||
decode_moa_turn,
|
||||
exact_moa_preset_name,
|
||||
normalize_moa_config,
|
||||
resolve_moa_preset,
|
||||
set_active_moa_preset,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -7,7 +7,6 @@ from hermes_cli.models import (
|
||||
azure_foundry_model_api_mode,
|
||||
copilot_model_api_mode,
|
||||
fetch_github_model_catalog,
|
||||
curated_models_for_provider,
|
||||
fetch_api_models,
|
||||
fetch_lmstudio_models,
|
||||
github_model_reasoning_efforts,
|
||||
@@ -57,25 +56,6 @@ class TestParseModelInput:
|
||||
assert model == "anthropic/claude-sonnet-4.5"
|
||||
|
||||
|
||||
# -- curated_models_for_provider ---------------------------------------------
|
||||
|
||||
class TestCuratedModelsForProvider:
|
||||
def test_openrouter_returns_curated_list(self):
|
||||
with patch(
|
||||
"hermes_cli.models.fetch_openrouter_models",
|
||||
return_value=[
|
||||
("anthropic/claude-opus-4.6", "recommended"),
|
||||
("qwen/qwen3.6-plus", ""),
|
||||
],
|
||||
):
|
||||
models = curated_models_for_provider("openrouter")
|
||||
assert len(models) > 0
|
||||
assert any("claude" in m[0] for m in models)
|
||||
|
||||
def test_unknown_provider_returns_empty(self):
|
||||
assert curated_models_for_provider("totally-unknown") == []
|
||||
|
||||
|
||||
# -- normalize_provider ------------------------------------------------------
|
||||
|
||||
class TestNormalizeProvider:
|
||||
|
||||
@@ -8,7 +8,7 @@ from unittest.mock import patch, MagicMock
|
||||
from hermes_cli.nous_account import NousPortalAccountInfo
|
||||
from hermes_cli.models import (
|
||||
OPENROUTER_MODELS, fetch_openrouter_models, model_ids, detect_provider_for_model,
|
||||
is_nous_free_tier, partition_nous_models_by_tier,
|
||||
partition_nous_models_by_tier,
|
||||
check_nous_free_tier, _FREE_TIER_CACHE_TTL,
|
||||
union_with_portal_free_recommendations,
|
||||
union_with_portal_paid_recommendations,
|
||||
@@ -163,23 +163,6 @@ class TestDetectProviderForModel:
|
||||
|
||||
|
||||
|
||||
class TestIsNousFreeTier:
|
||||
"""Tests for is_nous_free_tier — account tier detection."""
|
||||
|
||||
def test_paid_service_access_allowed_true_is_not_free(self):
|
||||
assert is_nous_free_tier({"paid_service_access": {"allowed": True}}) is False
|
||||
|
||||
|
||||
def test_empty_subscription_not_free(self):
|
||||
"""Empty subscription dict defaults to not-free (don't block users)."""
|
||||
assert is_nous_free_tier({"subscription": {}}) is False
|
||||
|
||||
|
||||
def test_empty_response_not_free(self):
|
||||
"""Completely empty response defaults to not-free."""
|
||||
assert is_nous_free_tier({}) is False
|
||||
|
||||
|
||||
class TestPartitionNousModelsByTier:
|
||||
"""Tests for partition_nous_models_by_tier — free vs paid tier model split."""
|
||||
|
||||
|
||||
@@ -96,13 +96,17 @@ class TestCallSiteWiring:
|
||||
)
|
||||
|
||||
def test_validator_wired_at_all_known_call_sites(self):
|
||||
"""All 2 known auth.py NETWORK sites use the validator. If this count
|
||||
drops, someone removed protection; if it grows, audit the new
|
||||
site to be sure validation is appropriate."""
|
||||
"""All 2 known auth.py NETWORK refresh sites route the Portal-returned
|
||||
inference URL through ``_healed_nous_inference_url`` (which applies the
|
||||
validator and heals to the default). If this count drops, someone removed
|
||||
protection; if it grows, audit the new site to be sure validation is
|
||||
appropriate."""
|
||||
source = self._read_auth_source()
|
||||
refresh_count = source.count(
|
||||
'_validate_nous_inference_url_from_network(refreshed.get("inference_base_url"))'
|
||||
)
|
||||
assert (
|
||||
source.count('_validate_nous_inference_url_from_network(refreshed.get("inference_base_url"))')
|
||||
== 1
|
||||
), "the validator must be applied exactly once, inside _healed_nous_inference_url"
|
||||
refresh_count = source.count("_healed_nous_inference_url(refreshed)")
|
||||
mint_count = source.count(
|
||||
'_validate_nous_inference_url_from_network(mint_payload.get("inference_base_url"))'
|
||||
)
|
||||
|
||||
@@ -9,13 +9,12 @@ The Go relay doesn't serve the free tier at all ("Model x is not supported").
|
||||
These tests pin the keyless routing added for the community report where the
|
||||
free Ox Alpha model failed under an OpenCode subscription:
|
||||
|
||||
1. ``is_opencode_zen_free_model`` recognizes free slugs (bare + prefixed).
|
||||
2. ``opencode_zen_free_runtime`` pins free slugs to the Zen relay with the
|
||||
1. ``opencode_zen_free_runtime`` pins free slugs to the Zen relay with the
|
||||
keyless placeholder + empty-Authorization headers, for BOTH family
|
||||
providers (Go selections heal to Zen).
|
||||
3. ``resolve_runtime_provider`` routes free slugs keylessly with no
|
||||
2. ``resolve_runtime_provider`` routes free slugs keylessly with no
|
||||
OPENCODE_* credential present, and still fails closed for paid models.
|
||||
4. The keyless placeholder never reaches the wire: client default_headers
|
||||
3. The keyless placeholder never reaches the wire: client default_headers
|
||||
carry ``Authorization: ""`` overriding the SDK bearer.
|
||||
"""
|
||||
|
||||
@@ -26,42 +25,11 @@ import pytest
|
||||
|
||||
from hermes_cli.models import (
|
||||
OPENCODE_ZEN_FREE_KEYLESS_PLACEHOLDER,
|
||||
is_opencode_zen_free_model,
|
||||
opencode_zen_free_headers,
|
||||
opencode_zen_free_runtime,
|
||||
)
|
||||
|
||||
|
||||
class TestFreeSlugDetection:
|
||||
def test_bare_free_slug(self):
|
||||
assert is_opencode_zen_free_model("x-preview-f-free")
|
||||
|
||||
def test_provider_prefixed_slug(self):
|
||||
assert is_opencode_zen_free_model("opencode-zen/x-preview-f-free")
|
||||
|
||||
def test_other_free_tier_slugs(self):
|
||||
for slug in (
|
||||
"hy3-free",
|
||||
"laguna-s-2.1-free",
|
||||
"mimo-v2.5-free",
|
||||
"nemotron-3-ultra-free",
|
||||
):
|
||||
assert is_opencode_zen_free_model(slug), slug
|
||||
|
||||
def test_paid_models_not_free(self):
|
||||
for slug in ("claude-sonnet-5", "glm-5.2", "kimi-k3", "gpt-5.6-sol"):
|
||||
assert not is_opencode_zen_free_model(slug), slug
|
||||
|
||||
def test_empty_and_none(self):
|
||||
assert not is_opencode_zen_free_model("")
|
||||
assert not is_opencode_zen_free_model(None)
|
||||
|
||||
def test_freedom_like_names_not_swept(self):
|
||||
# suffix match must be exact "-free", not substring "free"
|
||||
assert not is_opencode_zen_free_model("freeform-1")
|
||||
assert not is_opencode_zen_free_model("model-freedom")
|
||||
|
||||
|
||||
class TestFreeRuntime:
|
||||
def test_zen_provider_free_model(self):
|
||||
rt = opencode_zen_free_runtime("opencode-zen", "hy3-free")
|
||||
|
||||
@@ -30,23 +30,6 @@ def test_code_scoped_stamp_wins_over_home_stamp(tmp_path):
|
||||
assert detect_install_method(project_root=code) == "git"
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def test_stamp_install_method_writes_code_scoped(tmp_path):
|
||||
"""stamp_install_method writes next to the code, not into $HERMES_HOME."""
|
||||
code = tmp_path / "code"
|
||||
home = tmp_path / "home"
|
||||
code.mkdir()
|
||||
home.mkdir()
|
||||
with patch("hermes_cli.config.get_hermes_home", return_value=home):
|
||||
from hermes_cli.config import stamp_install_method
|
||||
stamp_install_method("git", project_root=code)
|
||||
assert (code / ".install_method").read_text().strip() == "git"
|
||||
assert not (home / ".install_method").exists()
|
||||
|
||||
|
||||
def test_container_without_stamp_is_not_docker(tmp_path):
|
||||
"""An unstamped install in a generic container must NOT be flagged as docker.
|
||||
|
||||
|
||||
@@ -11,7 +11,7 @@ The live profile on disk must stay untouched.
|
||||
|
||||
import tarfile
|
||||
|
||||
from hermes_cli.profiles import export_profile, _DEFAULT_EXPORT_EXCLUDE_ROOT
|
||||
from hermes_cli.profiles import export_profile
|
||||
|
||||
# Long enough to match agent.redact prefix patterns (sk- + 10+ chars).
|
||||
_LEAKED_KEY = "sk-or-v1-reallyLongSecretKeyValue12345678"
|
||||
@@ -25,11 +25,6 @@ def _patch_named_profile(monkeypatch, profiles_root, profile_dir):
|
||||
|
||||
class TestCredentialExclusion:
|
||||
|
||||
def test_auth_json_in_default_exclude_set(self):
|
||||
"""auth.json must be in the default export exclusion set."""
|
||||
assert "auth.json" in _DEFAULT_EXPORT_EXCLUDE_ROOT
|
||||
|
||||
|
||||
def test_named_profile_export_excludes_auth(self, tmp_path, monkeypatch):
|
||||
"""Named profile export must not contain auth.json or .env."""
|
||||
profiles_root = tmp_path / "profiles"
|
||||
|
||||
@@ -40,7 +40,6 @@ from hermes_cli.profiles import (
|
||||
_get_profiles_root,
|
||||
_get_default_hermes_home,
|
||||
seed_profile_skills,
|
||||
has_bundled_skills_opt_out,
|
||||
NO_BUNDLED_SKILLS_MARKER,
|
||||
backfill_profile_envs,
|
||||
profiles_to_serve,
|
||||
@@ -210,9 +209,6 @@ class TestNoSkillsOptOut:
|
||||
assert marker.is_file(), "expected .no-bundled-skills marker in profile root"
|
||||
assert "--no-skills" in marker.read_text()
|
||||
|
||||
# has_bundled_skills_opt_out() agrees
|
||||
assert has_bundled_skills_opt_out(profile_dir) is True
|
||||
|
||||
# skills/ dir exists (profile bootstrapping still creates the dir) but
|
||||
# contains nothing yet because create_profile itself doesn't seed.
|
||||
assert (profile_dir / "skills").is_dir()
|
||||
@@ -231,7 +227,7 @@ class TestNoSkillsOptOut:
|
||||
import subprocess as _sp
|
||||
|
||||
profile_dir = create_profile("orchestrator", no_alias=True, no_skills=True)
|
||||
assert has_bundled_skills_opt_out(profile_dir) is True
|
||||
assert (profile_dir / NO_BUNDLED_SKILLS_MARKER).is_file()
|
||||
|
||||
# Marker present: the subprocess still runs (essential-only seeding
|
||||
# happens inside sync_skills) and its skipped_opt_out flag surfaces.
|
||||
@@ -254,7 +250,6 @@ class TestNoSkillsOptOut:
|
||||
|
||||
# Delete marker → next call is a normal full sync.
|
||||
(profile_dir / NO_BUNDLED_SKILLS_MARKER).unlink()
|
||||
assert has_bundled_skills_opt_out(profile_dir) is False
|
||||
r2 = seed_profile_skills(profile_dir, quiet=True)
|
||||
assert r2 == {"copied": []}
|
||||
assert len(called) == 2
|
||||
|
||||
@@ -214,20 +214,6 @@ class TestRendering:
|
||||
assert fake_advisory.summary in body
|
||||
|
||||
|
||||
def test_render_doctor_section_with_unacked_hit(
|
||||
self, fake_advisory, monkeypatch
|
||||
):
|
||||
monkeypatch.setattr(adv, "get_acked_ids", lambda: set())
|
||||
hit = adv.AdvisoryHit(
|
||||
advisory=fake_advisory,
|
||||
package="fake-malicious-pkg",
|
||||
installed_version="6.6.6",
|
||||
)
|
||||
has_problems, lines = adv.render_doctor_section([hit])
|
||||
assert has_problems is True
|
||||
body = "\n".join(lines)
|
||||
assert fake_advisory.title in body
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"""Tests for the state.db integrity guard used by the update flow (#68474).
|
||||
|
||||
Exercises ``verify_sqlite_integrity`` and ``copy_db_and_verify`` against REAL
|
||||
Exercises ``verify_sqlite_integrity`` against REAL
|
||||
SQLite files (valid, zeroed, truncated) — the exact corruption signature from
|
||||
issue #68474 (file kept at original size, 100% null bytes, header gone).
|
||||
"""
|
||||
@@ -9,7 +9,7 @@ import sqlite3
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli.backup import copy_db_and_verify, verify_sqlite_integrity
|
||||
from hermes_cli.backup import verify_sqlite_integrity
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
|
||||
@@ -6,16 +6,11 @@ from pathlib import PureWindowsPath
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from hermes_cli.telegram_managed_bot import (
|
||||
DEFAULT_MANAGER_BOT,
|
||||
TELEGRAM_ONBOARDING_URL_ENV,
|
||||
TelegramBotSetupResult,
|
||||
TelegramPairing,
|
||||
create_pairing,
|
||||
generate_bot_username,
|
||||
generate_deep_link,
|
||||
generate_pairing_nonce,
|
||||
poll_for_setup_result,
|
||||
poll_for_token,
|
||||
print_qr_code,
|
||||
render_qr_terminal,
|
||||
)
|
||||
@@ -39,31 +34,6 @@ class TestGenerateBotUsername:
|
||||
assert len(names) == 20
|
||||
|
||||
|
||||
class TestGenerateDeepLink:
|
||||
def test_basic_format(self):
|
||||
link = generate_deep_link(
|
||||
manager_bot="TestBot",
|
||||
suggested_username="my_bot",
|
||||
)
|
||||
assert link == "https://t.me/newbot/TestBot/my_bot"
|
||||
|
||||
|
||||
def test_name_url_encoded(self):
|
||||
link = generate_deep_link(
|
||||
manager_bot="Bot",
|
||||
suggested_username="test_bot",
|
||||
suggested_name="Hermes & Friends",
|
||||
)
|
||||
assert "Hermes+%26+Friends" in link
|
||||
|
||||
|
||||
class TestPairingNonce:
|
||||
|
||||
def test_hex_chars(self):
|
||||
nonce = generate_pairing_nonce()
|
||||
assert all(c in "0123456789abcdef" for c in nonce)
|
||||
|
||||
|
||||
class TestQRCode:
|
||||
def test_render_returns_string(self):
|
||||
result = render_qr_terminal("https://example.com")
|
||||
@@ -157,11 +127,11 @@ class TestPollForToken:
|
||||
"hermes_cli.telegram_managed_bot.httpx.get", return_value=mock_resp
|
||||
) as get:
|
||||
with patch("hermes_cli.telegram_managed_bot.time.sleep"):
|
||||
token = poll_for_token(
|
||||
result = poll_for_setup_result(
|
||||
"https://api.example.com", self.pairing(), timeout=5
|
||||
)
|
||||
|
||||
assert token == VALID_TOKEN
|
||||
assert result is not None and result.token == VALID_TOKEN
|
||||
assert (
|
||||
get.call_args.args[0]
|
||||
== "https://api.example.com/v1/telegram/pairings/abcdefghijklmnop"
|
||||
@@ -193,10 +163,10 @@ class TestPollForToken:
|
||||
|
||||
with patch("hermes_cli.telegram_managed_bot.httpx.get", side_effect=fake_get):
|
||||
with patch("hermes_cli.telegram_managed_bot.time.sleep"):
|
||||
token = poll_for_token(
|
||||
result = poll_for_setup_result(
|
||||
"https://api.example.com", self.pairing(), timeout=30
|
||||
)
|
||||
assert token == SECOND_VALID_TOKEN
|
||||
assert result is not None and result.token == SECOND_VALID_TOKEN
|
||||
|
||||
|
||||
class TestSetupTelegramAuto:
|
||||
|
||||
@@ -10,7 +10,7 @@ import io
|
||||
|
||||
import yaml
|
||||
|
||||
from utils import fast_safe_load, _get_fast_yaml_loader
|
||||
from utils import fast_safe_load, _fast_yaml_loader
|
||||
|
||||
|
||||
_DOCS = [
|
||||
@@ -39,7 +39,7 @@ def test_empty_document_returns_none():
|
||||
|
||||
|
||||
def test_prefers_c_loader_when_available():
|
||||
loader = _get_fast_yaml_loader()
|
||||
loader = _fast_yaml_loader
|
||||
# If libyaml is compiled in, we must be using the C loader; otherwise the
|
||||
# pure-Python SafeLoader is an acceptable fallback. Either way it must be a
|
||||
# safe loader (never the unsafe full Loader).
|
||||
|
||||
@@ -27,7 +27,6 @@ from hermes_cli.gitlock import (
|
||||
LOCK_NAMES,
|
||||
STALE_LOCK_MIN_AGE_SECONDS,
|
||||
clear_stale_git_locks,
|
||||
is_ancestor_of_head,
|
||||
)
|
||||
|
||||
|
||||
@@ -114,26 +113,3 @@ def test_clear_noop_on_non_repo(tmp_path: Path) -> None:
|
||||
|
||||
def test_clear_noop_with_no_locks(repo: Path) -> None:
|
||||
assert clear_stale_git_locks(repo) == []
|
||||
|
||||
|
||||
def test_is_ancestor_true_for_first_commit(repo: Path) -> None:
|
||||
first = subprocess.run(
|
||||
["git", "rev-list", "--max-parents=0", "HEAD"],
|
||||
cwd=repo, capture_output=True, text=True, check=True,
|
||||
).stdout.strip()
|
||||
assert is_ancestor_of_head(repo, first) is True
|
||||
|
||||
|
||||
def test_is_ancestor_true_for_head_itself(repo: Path) -> None:
|
||||
head = subprocess.run(
|
||||
["git", "rev-parse", "HEAD"], cwd=repo, capture_output=True, text=True, check=True,
|
||||
).stdout.strip()
|
||||
assert is_ancestor_of_head(repo, head) is True
|
||||
|
||||
|
||||
def test_is_ancestor_false_for_unknown_rev(repo: Path) -> None:
|
||||
assert is_ancestor_of_head(repo, "deadbeef" * 5) is False
|
||||
|
||||
|
||||
def test_is_ancestor_false_for_nonexistent_repo(tmp_path: Path) -> None:
|
||||
assert is_ancestor_of_head(tmp_path / "missing", "HEAD") is False
|
||||
|
||||
@@ -78,7 +78,7 @@ class TestSetupLogging:
|
||||
root = logging.getLogger()
|
||||
|
||||
agent_handlers = [
|
||||
h for h in hermes_logging.rotating_file_handlers()
|
||||
h for h in hermes_logging._queued_file_handlers
|
||||
if isinstance(h, RotatingFileHandler)
|
||||
and "agent.log" in getattr(h, "baseFilename", "")
|
||||
]
|
||||
@@ -92,7 +92,7 @@ class TestSetupLogging:
|
||||
|
||||
root = logging.getLogger()
|
||||
agent_handlers = [
|
||||
h for h in hermes_logging.rotating_file_handlers()
|
||||
h for h in hermes_logging._queued_file_handlers
|
||||
if isinstance(h, RotatingFileHandler)
|
||||
and "agent.log" in getattr(h, "baseFilename", "")
|
||||
]
|
||||
@@ -155,7 +155,7 @@ class TestSetupLogging:
|
||||
|
||||
root = logging.getLogger()
|
||||
agent_handlers = [
|
||||
h for h in hermes_logging.rotating_file_handlers()
|
||||
h for h in hermes_logging._queued_file_handlers
|
||||
if isinstance(h, RotatingFileHandler)
|
||||
and "agent.log" in getattr(h, "baseFilename", "")
|
||||
]
|
||||
@@ -171,7 +171,7 @@ class TestGatewayMode:
|
||||
root = logging.getLogger()
|
||||
|
||||
gw_handlers = [
|
||||
h for h in hermes_logging.rotating_file_handlers()
|
||||
h for h in hermes_logging._queued_file_handlers
|
||||
if isinstance(h, RotatingFileHandler)
|
||||
and "gateway.log" in getattr(h, "baseFilename", "")
|
||||
]
|
||||
@@ -182,7 +182,7 @@ class TestGatewayMode:
|
||||
root = logging.getLogger()
|
||||
|
||||
gw_handlers = [
|
||||
h for h in hermes_logging.rotating_file_handlers()
|
||||
h for h in hermes_logging._queued_file_handlers
|
||||
if isinstance(h, RotatingFileHandler)
|
||||
and "gateway.log" in getattr(h, "baseFilename", "")
|
||||
]
|
||||
@@ -231,7 +231,7 @@ class TestGuiMode:
|
||||
root = logging.getLogger()
|
||||
|
||||
gui_handlers = [
|
||||
h for h in hermes_logging.rotating_file_handlers()
|
||||
h for h in hermes_logging._queued_file_handlers
|
||||
if isinstance(h, RotatingFileHandler)
|
||||
and "gui.log" in getattr(h, "baseFilename", "")
|
||||
]
|
||||
@@ -326,31 +326,25 @@ class TestAddRotatingHandler:
|
||||
|
||||
def test_no_duplicate_for_same_path(self, tmp_path):
|
||||
log_path = tmp_path / "test.log"
|
||||
logger = logging.getLogger("_test_rotating_dup")
|
||||
formatter = logging.Formatter("%(message)s")
|
||||
|
||||
hermes_logging._add_rotating_handler(
|
||||
logger, log_path,
|
||||
log_path,
|
||||
level=logging.INFO, max_bytes=1024, backup_count=1,
|
||||
formatter=formatter,
|
||||
)
|
||||
hermes_logging._add_rotating_handler(
|
||||
logger, log_path,
|
||||
log_path,
|
||||
level=logging.INFO, max_bytes=1024, backup_count=1,
|
||||
formatter=formatter,
|
||||
)
|
||||
|
||||
rotating_handlers = [
|
||||
h for h in hermes_logging.rotating_file_handlers()
|
||||
h for h in hermes_logging._queued_file_handlers
|
||||
if isinstance(h, RotatingFileHandler)
|
||||
]
|
||||
assert len(rotating_handlers) == 1
|
||||
# Clean up
|
||||
for h in list(logger.handlers):
|
||||
if isinstance(h, RotatingFileHandler):
|
||||
logger.removeHandler(h)
|
||||
h.close()
|
||||
|
||||
|
||||
def test_no_session_filter_on_handler(self, tmp_path):
|
||||
"""Handlers rely on record factory, not per-handler _SessionFilter."""
|
||||
@@ -359,12 +353,12 @@ class TestAddRotatingHandler:
|
||||
formatter = logging.Formatter("%(session_tag)s%(message)s")
|
||||
|
||||
hermes_logging._add_rotating_handler(
|
||||
logger, log_path,
|
||||
log_path,
|
||||
level=logging.INFO, max_bytes=1024, backup_count=1,
|
||||
formatter=formatter,
|
||||
)
|
||||
|
||||
handlers = [h for h in hermes_logging.rotating_file_handlers() if isinstance(h, RotatingFileHandler)]
|
||||
handlers = [h for h in hermes_logging._queued_file_handlers if isinstance(h, RotatingFileHandler)]
|
||||
assert len(handlers) == 1
|
||||
# No _SessionFilter on the handler — record factory handles it
|
||||
assert len(handlers[0].filters) == 0
|
||||
@@ -377,21 +371,15 @@ class TestAddRotatingHandler:
|
||||
assert "[factory_test]" in content
|
||||
|
||||
# Clean up
|
||||
for h in list(logger.handlers):
|
||||
if isinstance(h, RotatingFileHandler):
|
||||
logger.removeHandler(h)
|
||||
h.close()
|
||||
|
||||
def test_managed_mode_initial_open_sets_group_writable(self, tmp_path):
|
||||
log_path = tmp_path / "managed-open.log"
|
||||
logger = logging.getLogger("_test_rotating_managed_open")
|
||||
formatter = logging.Formatter("%(message)s")
|
||||
|
||||
old_umask = os.umask(0o022)
|
||||
try:
|
||||
with patch("hermes_cli.config.is_managed", return_value=True):
|
||||
hermes_logging._add_rotating_handler(
|
||||
logger, log_path,
|
||||
log_path,
|
||||
level=logging.INFO, max_bytes=1024, backup_count=1,
|
||||
formatter=formatter,
|
||||
)
|
||||
@@ -401,11 +389,6 @@ class TestAddRotatingHandler:
|
||||
assert log_path.exists()
|
||||
assert stat.S_IMODE(log_path.stat().st_mode) == 0o660
|
||||
|
||||
for h in list(logger.handlers):
|
||||
if isinstance(h, RotatingFileHandler):
|
||||
logger.removeHandler(h)
|
||||
h.close()
|
||||
|
||||
|
||||
|
||||
class TestWindowsConcurrentLogLockTimeout:
|
||||
@@ -700,6 +683,6 @@ class TestAsyncQueueLogging:
|
||||
# The real file handlers are discoverable via the accessor.
|
||||
assert any(
|
||||
"agent.log" in getattr(h, "baseFilename", "")
|
||||
for h in hermes_logging.rotating_file_handlers()
|
||||
for h in hermes_logging._queued_file_handlers
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user