Merge branch 'main' into bb/gui
This commit is contained in:
emozilla
@@ -170,6 +170,50 @@ def test_auth_add_nous_oauth_persists_pool_entry(tmp_path, monkeypatch):
|
||||
assert singleton["inference_base_url"] == "https://inference.example.com/v1"
|
||||
|
||||
|
||||
def test_auth_add_minimax_oauth_starts_login_and_persists_pool_entry(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
|
||||
_write_auth_store(tmp_path, {"version": 1, "providers": {}})
|
||||
token = _jwt_with_email("minimax@example.com")
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.auth._minimax_oauth_login",
|
||||
lambda **kwargs: {
|
||||
"provider": "minimax-oauth",
|
||||
"region": "global",
|
||||
"portal_base_url": "https://api.minimax.io",
|
||||
"inference_base_url": "https://api.minimax.io/anthropic",
|
||||
"client_id": "client-id",
|
||||
"scope": "group_id profile model.completion",
|
||||
"token_type": "Bearer",
|
||||
"access_token": token,
|
||||
"refresh_token": "refresh-token",
|
||||
"resource_url": None,
|
||||
"obtained_at": "2026-05-11T10:00:00+00:00",
|
||||
"expires_at": "2026-05-14T10:00:00+00:00",
|
||||
"expires_in": 259200,
|
||||
},
|
||||
)
|
||||
|
||||
from hermes_cli.auth_commands import auth_add_command
|
||||
|
||||
class _Args:
|
||||
provider = "minimax-oauth"
|
||||
auth_type = "oauth"
|
||||
api_key = None
|
||||
label = None
|
||||
no_browser = True
|
||||
timeout = None
|
||||
|
||||
auth_add_command(_Args())
|
||||
|
||||
payload = json.loads((tmp_path / "hermes" / "auth.json").read_text())
|
||||
entries = payload["credential_pool"]["minimax-oauth"]
|
||||
entry = next(item for item in entries if item["source"] == "manual:minimax_oauth")
|
||||
assert entry["label"] == "minimax@example.com"
|
||||
assert entry["access_token"] == token
|
||||
assert entry["refresh_token"] == "refresh-token"
|
||||
assert entry["base_url"] == "https://api.minimax.io/anthropic"
|
||||
|
||||
|
||||
def test_auth_add_nous_oauth_honors_custom_label(tmp_path, monkeypatch):
|
||||
"""`hermes auth add nous --type oauth --label <name>` must preserve the
|
||||
custom label end-to-end — it was silently dropped in the first cut of the
|
||||
|
||||
@@ -242,12 +242,14 @@ class TestTelegramBotCommands:
|
||||
tg_name = cmd.name.replace("-", "_")
|
||||
assert tg_name not in names
|
||||
|
||||
def test_excludes_commands_with_required_args(self):
|
||||
def test_includes_builtin_commands_with_required_args(self):
|
||||
"""Built-in arg-taking commands (e.g. /queue, /steer, /background)
|
||||
are now included because their handlers return usage text when
|
||||
invoked without arguments — issue #24312."""
|
||||
names = {name for name, _ in telegram_bot_commands()}
|
||||
assert "background" not in names
|
||||
assert "queue" not in names
|
||||
assert "steer" not in names
|
||||
assert "background" in GATEWAY_KNOWN_COMMANDS
|
||||
assert "background" in names
|
||||
assert "queue" in names
|
||||
assert "steer" in names
|
||||
|
||||
|
||||
class TestSlackSubcommandMap:
|
||||
|
||||
@@ -2,10 +2,11 @@
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def test_profiles_nav_label_uses_short_multi_agents_copy():
|
||||
def test_profiles_nav_label_uses_short_copy():
|
||||
en_i18n = Path(__file__).resolve().parents[2] / "web" / "src" / "i18n" / "en.ts"
|
||||
|
||||
content = en_i18n.read_text(encoding="utf-8")
|
||||
|
||||
assert 'profiles: "profiles : multi agents"' in content
|
||||
assert "Profiles: Running Multiple Agents" not in content
|
||||
# Nav label should be the clean short form, not the old verbose string
|
||||
assert 'profiles: "Profiles"' in content
|
||||
assert "profiles : multi agents" not in content
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
"""Host-specific gating in ``hermes_cli.gateway._all_platforms()``.
|
||||
|
||||
Some messaging platforms can't function on every host. The gate lives
|
||||
in one place — ``_all_platforms()`` — so the setup wizard, the curses
|
||||
gateway-config menu, and any future picker all see the same filtered
|
||||
list.
|
||||
|
||||
Currently:
|
||||
- Matrix is hidden on Windows. The ``[matrix]`` extra pulls
|
||||
``mautrix[encryption]`` -> ``python-olm``, which has no Windows wheel
|
||||
and needs ``make`` + libolm to build from sdist. There's no native
|
||||
Windows path that works.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
|
||||
class TestMatrixHiddenOnWindows:
|
||||
def test_matrix_present_on_linux(self, monkeypatch):
|
||||
"""Sanity: matrix is still in the picker on Linux/macOS."""
|
||||
import hermes_cli.gateway as gateway_mod
|
||||
|
||||
monkeypatch.setattr(gateway_mod.sys, "platform", "linux")
|
||||
platforms = gateway_mod._all_platforms()
|
||||
keys = {p["key"] for p in platforms}
|
||||
assert "matrix" in keys, "matrix must be available on Linux"
|
||||
|
||||
def test_matrix_present_on_macos(self, monkeypatch):
|
||||
import hermes_cli.gateway as gateway_mod
|
||||
|
||||
monkeypatch.setattr(gateway_mod.sys, "platform", "darwin")
|
||||
platforms = gateway_mod._all_platforms()
|
||||
keys = {p["key"] for p in platforms}
|
||||
assert "matrix" in keys, "matrix must be available on macOS"
|
||||
|
||||
def test_matrix_hidden_on_windows(self, monkeypatch):
|
||||
"""The actual gate: matrix must NOT appear on Windows."""
|
||||
import hermes_cli.gateway as gateway_mod
|
||||
|
||||
monkeypatch.setattr(gateway_mod.sys, "platform", "win32")
|
||||
platforms = gateway_mod._all_platforms()
|
||||
keys = {p["key"] for p in platforms}
|
||||
assert "matrix" not in keys, (
|
||||
"matrix must be hidden on Windows — python-olm has no "
|
||||
"Windows wheel and no native build path"
|
||||
)
|
||||
|
||||
def test_other_platforms_unaffected_on_windows(self, monkeypatch):
|
||||
"""Gating must only drop matrix, not collateral damage."""
|
||||
import hermes_cli.gateway as gateway_mod
|
||||
|
||||
monkeypatch.setattr(gateway_mod.sys, "platform", "win32")
|
||||
platforms = gateway_mod._all_platforms()
|
||||
keys = {p["key"] for p in platforms}
|
||||
# A representative sample of platforms that have no Windows
|
||||
# blockers — picker should still surface them.
|
||||
for must_have in ("telegram", "discord", "slack", "mattermost"):
|
||||
assert must_have in keys, (
|
||||
f"{must_have} disappeared from Windows picker — gate is "
|
||||
"over-filtering"
|
||||
)
|
||||
@@ -7,6 +7,7 @@ from hermes_cli.models import (
|
||||
is_nous_free_tier, partition_nous_models_by_tier,
|
||||
check_nous_free_tier, _FREE_TIER_CACHE_TTL,
|
||||
union_with_portal_free_recommendations,
|
||||
union_with_portal_paid_recommendations,
|
||||
)
|
||||
import hermes_cli.models as _models_mod
|
||||
|
||||
@@ -506,6 +507,147 @@ class TestUnionWithPortalFreeRecommendations:
|
||||
assert p["qwen/qwen3.6-plus"] == self._FREE
|
||||
|
||||
|
||||
class TestUnionWithPortalPaidRecommendations:
|
||||
"""Tests for union_with_portal_paid_recommendations.
|
||||
|
||||
Mirror of TestUnionWithPortalFreeRecommendations: the Portal's
|
||||
paidRecommendedModels endpoint is the source of truth for what's a
|
||||
blessed paid model *right now*. The in-repo curated list and
|
||||
docs-hosted manifest can lag — this helper guarantees newly-launched
|
||||
paid models surface in the picker for paid-tier users without a CLI
|
||||
release.
|
||||
"""
|
||||
|
||||
_PAID = {"prompt": "0.000003", "completion": "0.000015"}
|
||||
_FREE = {"prompt": "0", "completion": "0"}
|
||||
|
||||
def _payload(self, paid_models: list[str]) -> dict:
|
||||
return {
|
||||
"paidRecommendedModels": [
|
||||
{"modelName": mid, "displayName": mid} for mid in paid_models
|
||||
],
|
||||
}
|
||||
|
||||
def test_adds_portal_paid_model_missing_from_curated(self):
|
||||
"""A Portal-advertised paid model not in curated is prepended."""
|
||||
curated = ["anthropic/claude-opus-4.6"]
|
||||
pricing = {"anthropic/claude-opus-4.6": self._PAID}
|
||||
with patch(
|
||||
"hermes_cli.models.fetch_nous_recommended_models",
|
||||
return_value=self._payload(["openai/gpt-5.4"]),
|
||||
):
|
||||
ids, p = union_with_portal_paid_recommendations(curated, pricing, "")
|
||||
|
||||
assert ids[0] == "openai/gpt-5.4" # prepended
|
||||
assert "anthropic/claude-opus-4.6" in ids
|
||||
# Existing pricing untouched
|
||||
assert p["anthropic/claude-opus-4.6"] == self._PAID
|
||||
|
||||
def test_does_not_synthesize_pricing_for_paid_models(self):
|
||||
"""Paid recommendations missing from live pricing get no synthetic entry.
|
||||
|
||||
Synthesizing zero pricing (like the free helper does) would mislead
|
||||
:func:`partition_nous_models_by_tier` into treating them as free;
|
||||
synthesizing a non-zero placeholder would lie to the user. The
|
||||
right thing is to leave pricing absent so the picker shows a blank
|
||||
column until the live pricing endpoint catches up.
|
||||
"""
|
||||
curated = ["anthropic/claude-opus-4.6"]
|
||||
pricing = {"anthropic/claude-opus-4.6": self._PAID}
|
||||
with patch(
|
||||
"hermes_cli.models.fetch_nous_recommended_models",
|
||||
return_value=self._payload(["openai/gpt-5.4"]),
|
||||
):
|
||||
_, p = union_with_portal_paid_recommendations(curated, pricing, "")
|
||||
|
||||
assert "openai/gpt-5.4" not in p
|
||||
assert p["anthropic/claude-opus-4.6"] == self._PAID
|
||||
|
||||
def test_does_not_duplicate_curated_entries(self):
|
||||
"""A Portal paid model already in curated is not duplicated."""
|
||||
curated = ["openai/gpt-5.4", "anthropic/claude-opus-4.6"]
|
||||
pricing = {
|
||||
"openai/gpt-5.4": self._PAID,
|
||||
"anthropic/claude-opus-4.6": self._PAID,
|
||||
}
|
||||
with patch(
|
||||
"hermes_cli.models.fetch_nous_recommended_models",
|
||||
return_value=self._payload(["openai/gpt-5.4"]),
|
||||
):
|
||||
ids, p = union_with_portal_paid_recommendations(curated, pricing, "")
|
||||
|
||||
assert ids == curated
|
||||
assert p == pricing
|
||||
|
||||
def test_empty_payload_returns_inputs_unchanged(self):
|
||||
"""Empty Portal response leaves curated + pricing untouched."""
|
||||
curated = ["a", "b"]
|
||||
pricing = {"a": self._PAID}
|
||||
with patch("hermes_cli.models.fetch_nous_recommended_models", return_value={}):
|
||||
ids, p = union_with_portal_paid_recommendations(curated, pricing, "")
|
||||
assert ids == curated
|
||||
assert p == pricing
|
||||
|
||||
def test_missing_paidRecommendedModels_key(self):
|
||||
"""Portal payload without paidRecommendedModels degrades gracefully."""
|
||||
curated = ["a"]
|
||||
pricing = {"a": self._PAID}
|
||||
with patch(
|
||||
"hermes_cli.models.fetch_nous_recommended_models",
|
||||
return_value={"freeRecommendedModels": [{"modelName": "x"}]},
|
||||
):
|
||||
ids, p = union_with_portal_paid_recommendations(curated, pricing, "")
|
||||
assert ids == curated
|
||||
assert p == pricing
|
||||
|
||||
def test_fetch_failure_returns_inputs(self):
|
||||
"""Network failures don't blow up the picker."""
|
||||
curated = ["a"]
|
||||
pricing = {"a": self._PAID}
|
||||
with patch(
|
||||
"hermes_cli.models.fetch_nous_recommended_models",
|
||||
side_effect=RuntimeError("network down"),
|
||||
):
|
||||
ids, p = union_with_portal_paid_recommendations(curated, pricing, "")
|
||||
assert ids == curated
|
||||
assert p == pricing
|
||||
|
||||
def test_invalid_entries_skipped(self):
|
||||
"""Non-dict / missing-modelName entries are filtered out."""
|
||||
curated = ["a"]
|
||||
pricing = {"a": self._PAID}
|
||||
with patch(
|
||||
"hermes_cli.models.fetch_nous_recommended_models",
|
||||
return_value={
|
||||
"paidRecommendedModels": [
|
||||
"not-a-dict",
|
||||
{"displayName": "no-modelName"},
|
||||
{"modelName": ""},
|
||||
{"modelName": "openai/gpt-5.4"},
|
||||
]
|
||||
},
|
||||
):
|
||||
ids, p = union_with_portal_paid_recommendations(curated, pricing, "")
|
||||
assert ids == ["openai/gpt-5.4", "a"]
|
||||
# No synthetic entry — pricing is untouched.
|
||||
assert "openai/gpt-5.4" not in p
|
||||
|
||||
def test_preserves_relative_order_of_new_paid_models(self):
|
||||
"""Multiple new paid models are prepended in payload order."""
|
||||
curated = ["anthropic/claude-opus-4.6"]
|
||||
pricing = {"anthropic/claude-opus-4.6": self._PAID}
|
||||
with patch(
|
||||
"hermes_cli.models.fetch_nous_recommended_models",
|
||||
return_value=self._payload(["openai/gpt-5.4", "openai/gpt-5.5"]),
|
||||
):
|
||||
ids, _ = union_with_portal_paid_recommendations(curated, pricing, "")
|
||||
assert ids == [
|
||||
"openai/gpt-5.4",
|
||||
"openai/gpt-5.5",
|
||||
"anthropic/claude-opus-4.6",
|
||||
]
|
||||
|
||||
|
||||
class TestCheckNousFreeTierCache:
|
||||
"""Tests for the TTL cache on check_nous_free_tier()."""
|
||||
|
||||
|
||||
@@ -2285,3 +2285,39 @@ def test_minimax_oauth_runtime_uses_inference_base_url(monkeypatch):
|
||||
resolved = rp.resolve_runtime_provider(requested="minimax-oauth")
|
||||
|
||||
assert MINIMAX_OAUTH_CN_INFERENCE.rstrip("/") in resolved["base_url"]
|
||||
|
||||
|
||||
def test_minimax_oauth_pool_forces_anthropic_messages_despite_stale_config(monkeypatch):
|
||||
"""A pooled MiniMax OAuth token must not inherit stale chat_completions config."""
|
||||
|
||||
class _Entry:
|
||||
access_token = "oauth-token"
|
||||
source = "manual:minimax_oauth"
|
||||
base_url = "https://api.minimax.io/anthropic"
|
||||
|
||||
class _Pool:
|
||||
def has_credentials(self):
|
||||
return True
|
||||
|
||||
def select(self):
|
||||
return _Entry()
|
||||
|
||||
monkeypatch.setattr(rp, "resolve_provider", lambda *a, **k: "minimax-oauth")
|
||||
monkeypatch.setattr(
|
||||
rp,
|
||||
"_get_model_config",
|
||||
lambda: {
|
||||
"provider": "minimax-oauth",
|
||||
"default": "MiniMax-M2.7",
|
||||
"api_mode": "chat_completions",
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(rp, "load_pool", lambda provider: _Pool())
|
||||
monkeypatch.setattr(rp, "_resolve_named_custom_runtime", lambda **k: None)
|
||||
monkeypatch.setattr(rp, "_resolve_explicit_runtime", lambda **k: None)
|
||||
|
||||
resolved = rp.resolve_runtime_provider(requested="minimax-oauth")
|
||||
|
||||
assert resolved["provider"] == "minimax-oauth"
|
||||
assert resolved["api_mode"] == "anthropic_messages"
|
||||
assert resolved["base_url"] == "https://api.minimax.io/anthropic"
|
||||
|
||||
@@ -0,0 +1,330 @@
|
||||
"""Tests for hermes_cli.security_advisories.
|
||||
|
||||
The advisory module is the user-facing detection / remediation surface
|
||||
for supply-chain attacks (e.g. the Mini Shai-Hulud worm of May 2026 that
|
||||
poisoned mistralai 2.4.6 on PyPI). These tests exercise the public API in
|
||||
isolation — no real package metadata, no real config, no real cache.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Iterator
|
||||
|
||||
import pytest
|
||||
|
||||
import hermes_cli.security_advisories as adv
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_advisory() -> adv.Advisory:
|
||||
"""A self-contained Advisory used across tests."""
|
||||
return adv.Advisory(
|
||||
id="test-advisory-2026-99",
|
||||
title="Test advisory",
|
||||
summary="Pretend this package has been compromised.",
|
||||
url="https://example.com/advisory",
|
||||
compromised=(
|
||||
("fake-malicious-pkg", frozenset({"6.6.6"})),
|
||||
),
|
||||
remediation=(
|
||||
"pip uninstall -y fake-malicious-pkg",
|
||||
"Rotate any credentials that may have been exposed.",
|
||||
),
|
||||
published="2026-01-01",
|
||||
severity="critical",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def isolated_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
"""Redirect HERMES_HOME so banner cache and config writes are sandboxed."""
|
||||
home = tmp_path / ".hermes"
|
||||
home.mkdir()
|
||||
(home / "cache").mkdir()
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
return home
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def patched_version(monkeypatch: pytest.MonkeyPatch) -> Iterator[dict[str, str]]:
|
||||
"""Override _installed_version with a controllable lookup table."""
|
||||
table: dict[str, str] = {}
|
||||
monkeypatch.setattr(adv, "_installed_version", lambda pkg: table.get(pkg))
|
||||
yield table
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# detect_compromised
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDetectCompromised:
|
||||
def test_no_match_returns_empty_list(self, fake_advisory, patched_version):
|
||||
# No matching package installed.
|
||||
hits = adv.detect_compromised(advisories=[fake_advisory])
|
||||
assert hits == []
|
||||
|
||||
def test_exact_version_match(self, fake_advisory, patched_version):
|
||||
patched_version["fake-malicious-pkg"] = "6.6.6"
|
||||
hits = adv.detect_compromised(advisories=[fake_advisory])
|
||||
assert len(hits) == 1
|
||||
assert hits[0].advisory.id == fake_advisory.id
|
||||
assert hits[0].package == "fake-malicious-pkg"
|
||||
assert hits[0].installed_version == "6.6.6"
|
||||
|
||||
def test_safe_version_does_not_match(self, fake_advisory, patched_version):
|
||||
# Package is installed but the version is not in the compromised set.
|
||||
patched_version["fake-malicious-pkg"] = "6.6.5"
|
||||
hits = adv.detect_compromised(advisories=[fake_advisory])
|
||||
assert hits == []
|
||||
|
||||
def test_empty_compromised_set_matches_any_version(
|
||||
self, patched_version
|
||||
):
|
||||
# An advisory with an empty version set is a "any version is suspect"
|
||||
# wildcard — used when an entire maintainer namespace is owned.
|
||||
wildcard = adv.Advisory(
|
||||
id="wildcard",
|
||||
title="Whole namespace owned",
|
||||
summary="x",
|
||||
url="x",
|
||||
compromised=(("evil-namespace", frozenset()),),
|
||||
remediation=("uninstall it",),
|
||||
)
|
||||
patched_version["evil-namespace"] = "0.0.1"
|
||||
hits = adv.detect_compromised(advisories=[wildcard])
|
||||
assert len(hits) == 1
|
||||
assert hits[0].installed_version == "0.0.1"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Acknowledgement persistence
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAck:
|
||||
def test_get_acked_ids_empty_when_no_config(self, monkeypatch):
|
||||
# load_config raises → returns empty set, doesn't crash.
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config",
|
||||
lambda: (_ for _ in ()).throw(RuntimeError("boom")),
|
||||
)
|
||||
assert adv.get_acked_ids() == set()
|
||||
|
||||
def test_filter_unacked_strips_dismissed(self, fake_advisory, monkeypatch):
|
||||
hit = adv.AdvisoryHit(
|
||||
advisory=fake_advisory,
|
||||
package="fake-malicious-pkg",
|
||||
installed_version="6.6.6",
|
||||
)
|
||||
monkeypatch.setattr(adv, "get_acked_ids", lambda: {fake_advisory.id})
|
||||
assert adv.filter_unacked([hit]) == []
|
||||
|
||||
def test_filter_unacked_passes_through_unknown(
|
||||
self, fake_advisory, monkeypatch
|
||||
):
|
||||
hit = adv.AdvisoryHit(
|
||||
advisory=fake_advisory,
|
||||
package="fake-malicious-pkg",
|
||||
installed_version="6.6.6",
|
||||
)
|
||||
monkeypatch.setattr(adv, "get_acked_ids", lambda: set())
|
||||
assert adv.filter_unacked([hit]) == [hit]
|
||||
|
||||
def test_ack_advisory_persists_id(self, isolated_home, monkeypatch):
|
||||
# Stub the config layer end-to-end with a tiny in-memory store so we
|
||||
# don't depend on the full hermes_cli.config bootstrap.
|
||||
store: dict = {"security": {}}
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.load_config", lambda: store
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.config.save_config",
|
||||
lambda cfg: store.update(cfg) or None,
|
||||
)
|
||||
assert adv.ack_advisory("test-advisory-2026-99") is True
|
||||
assert "test-advisory-2026-99" in store["security"]["acked_advisories"]
|
||||
# Idempotent.
|
||||
adv.ack_advisory("test-advisory-2026-99")
|
||||
assert (
|
||||
store["security"]["acked_advisories"].count("test-advisory-2026-99")
|
||||
== 1
|
||||
)
|
||||
|
||||
def test_ack_advisory_rejects_blank(self, isolated_home):
|
||||
assert adv.ack_advisory("") is False
|
||||
assert adv.ack_advisory(" ") is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Banner cache rate limiting
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBannerCache:
|
||||
def test_first_call_returns_due_hits(
|
||||
self, fake_advisory, isolated_home, monkeypatch
|
||||
):
|
||||
monkeypatch.setattr(adv, "get_acked_ids", lambda: set())
|
||||
hit = adv.AdvisoryHit(
|
||||
advisory=fake_advisory,
|
||||
package="fake-malicious-pkg",
|
||||
installed_version="6.6.6",
|
||||
)
|
||||
due = adv.hits_due_for_banner([hit])
|
||||
assert due == [hit]
|
||||
|
||||
def test_second_call_within_window_suppresses(
|
||||
self, fake_advisory, isolated_home, monkeypatch
|
||||
):
|
||||
monkeypatch.setattr(adv, "get_acked_ids", lambda: set())
|
||||
hit = adv.AdvisoryHit(
|
||||
advisory=fake_advisory,
|
||||
package="fake-malicious-pkg",
|
||||
installed_version="6.6.6",
|
||||
)
|
||||
adv.hits_due_for_banner([hit])
|
||||
# Same banner inside repeat window → suppressed.
|
||||
again = adv.hits_due_for_banner([hit])
|
||||
assert again == []
|
||||
|
||||
def test_call_after_window_re_banners(
|
||||
self, fake_advisory, isolated_home, monkeypatch
|
||||
):
|
||||
monkeypatch.setattr(adv, "get_acked_ids", lambda: set())
|
||||
hit = adv.AdvisoryHit(
|
||||
advisory=fake_advisory,
|
||||
package="fake-malicious-pkg",
|
||||
installed_version="6.6.6",
|
||||
)
|
||||
adv.hits_due_for_banner([hit])
|
||||
# Backdate the cache so it looks like the banner was shown more
|
||||
# than 24h ago — should re-banner.
|
||||
cache_path = adv._banner_cache_path()
|
||||
assert cache_path is not None
|
||||
old_lines = cache_path.read_text(encoding="utf-8").splitlines()
|
||||
backdated = []
|
||||
for line in old_lines:
|
||||
parts = line.split(None, 1)
|
||||
if len(parts) == 2:
|
||||
backdated.append(f"{parts[0]} {time.time() - 48 * 3600}")
|
||||
cache_path.write_text("\n".join(backdated) + "\n", encoding="utf-8")
|
||||
again = adv.hits_due_for_banner([hit])
|
||||
assert again == [hit]
|
||||
|
||||
def test_acked_hits_never_banner(
|
||||
self, fake_advisory, isolated_home, monkeypatch
|
||||
):
|
||||
monkeypatch.setattr(adv, "get_acked_ids", lambda: {fake_advisory.id})
|
||||
hit = adv.AdvisoryHit(
|
||||
advisory=fake_advisory,
|
||||
package="fake-malicious-pkg",
|
||||
installed_version="6.6.6",
|
||||
)
|
||||
assert adv.hits_due_for_banner([hit]) == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Rendering
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRendering:
|
||||
def test_short_banner_lines_includes_id_and_version(self, fake_advisory):
|
||||
hit = adv.AdvisoryHit(
|
||||
advisory=fake_advisory,
|
||||
package="fake-malicious-pkg",
|
||||
installed_version="6.6.6",
|
||||
)
|
||||
lines = adv.short_banner_lines([hit])
|
||||
joined = "\n".join(lines)
|
||||
assert fake_advisory.id in joined
|
||||
assert fake_advisory.title in joined
|
||||
assert "fake-malicious-pkg==6.6.6" in joined
|
||||
assert "hermes doctor" in joined
|
||||
|
||||
def test_full_remediation_text_contains_all_steps(self, fake_advisory):
|
||||
hit = adv.AdvisoryHit(
|
||||
advisory=fake_advisory,
|
||||
package="fake-malicious-pkg",
|
||||
installed_version="6.6.6",
|
||||
)
|
||||
body = "\n".join(adv.full_remediation_text(hit))
|
||||
# All remediation steps must be present.
|
||||
for step in fake_advisory.remediation:
|
||||
assert step in body
|
||||
assert fake_advisory.url in body
|
||||
assert fake_advisory.summary in body
|
||||
|
||||
def test_render_doctor_section_clean_state(self):
|
||||
# No hits → success message, has_problems=False.
|
||||
has_problems, lines = adv.render_doctor_section([])
|
||||
assert has_problems is False
|
||||
assert any("No active security advisories" in line for line in lines)
|
||||
|
||||
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
|
||||
|
||||
def test_gateway_log_message_singular(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",
|
||||
)
|
||||
msg = adv.gateway_log_message([hit])
|
||||
assert msg is not None
|
||||
assert fake_advisory.id in msg
|
||||
assert "fake-malicious-pkg==6.6.6" in msg
|
||||
|
||||
def test_gateway_log_message_returns_none_for_no_hits(self):
|
||||
assert adv.gateway_log_message([]) is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Real catalog smoke test
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRealCatalog:
|
||||
def test_advisories_well_formed(self):
|
||||
"""Every shipped advisory must be self-consistent.
|
||||
|
||||
Catches data-entry mistakes (empty IDs, missing remediation, bad
|
||||
compromised tuples) before they ship.
|
||||
"""
|
||||
seen_ids: set[str] = set()
|
||||
for advisory in adv.ADVISORIES:
|
||||
assert advisory.id, "advisory has empty id"
|
||||
assert advisory.id not in seen_ids, f"duplicate id {advisory.id}"
|
||||
seen_ids.add(advisory.id)
|
||||
assert advisory.title, f"{advisory.id}: empty title"
|
||||
assert advisory.summary, f"{advisory.id}: empty summary"
|
||||
assert advisory.remediation, f"{advisory.id}: empty remediation"
|
||||
assert advisory.url.startswith("http"), \
|
||||
f"{advisory.id}: bad url {advisory.url!r}"
|
||||
assert advisory.compromised, \
|
||||
f"{advisory.id}: empty compromised tuple"
|
||||
for pkg, versions in advisory.compromised:
|
||||
assert pkg, f"{advisory.id}: empty package name"
|
||||
assert isinstance(versions, frozenset), \
|
||||
f"{advisory.id}: versions must be frozenset"
|
||||
@@ -6,6 +6,7 @@ rather than leaving zombie processes or telling users to manually restart
|
||||
when launchd will auto-respawn.
|
||||
"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch, MagicMock
|
||||
@@ -1068,13 +1069,18 @@ class TestFindGatewayPidsExclude:
|
||||
|
||||
def test_excludes_specified_pids(self, monkeypatch):
|
||||
monkeypatch.setattr(gateway_cli, "is_windows", lambda: False)
|
||||
# Bypass /proc scan so the subprocess (ps) fallback is used
|
||||
_real_isdir = os.path.isdir
|
||||
monkeypatch.setattr("os.path.isdir", lambda p: False if p == "/proc" else _real_isdir(p))
|
||||
monkeypatch.setattr(gateway_cli, "_get_service_pids", lambda: set())
|
||||
monkeypatch.setattr(gateway_cli, "_get_ancestor_pids", lambda: {999})
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
return subprocess.CompletedProcess(
|
||||
cmd, 0,
|
||||
stdout=(
|
||||
"user 100 0.0 0.0 0 0 ? S 00:00 0:00 python gateway/run.py\n"
|
||||
"user 200 0.0 0.0 0 0 ? S 00:00 0:00 python gateway/run.py\n"
|
||||
"100 python gateway/run.py\n"
|
||||
"200 python gateway/run.py\n"
|
||||
),
|
||||
stderr="",
|
||||
)
|
||||
@@ -1082,19 +1088,24 @@ class TestFindGatewayPidsExclude:
|
||||
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
||||
monkeypatch.setattr("os.getpid", lambda: 999)
|
||||
|
||||
pids = gateway_cli.find_gateway_pids(exclude_pids={100})
|
||||
pids = gateway_cli.find_gateway_pids(exclude_pids={100}, all_profiles=True)
|
||||
assert 100 not in pids
|
||||
assert 200 in pids
|
||||
|
||||
def test_no_exclude_returns_all(self, monkeypatch):
|
||||
monkeypatch.setattr(gateway_cli, "is_windows", lambda: False)
|
||||
# Bypass /proc scan so the subprocess (ps) fallback is used
|
||||
_real_isdir = os.path.isdir
|
||||
monkeypatch.setattr("os.path.isdir", lambda p: False if p == "/proc" else _real_isdir(p))
|
||||
monkeypatch.setattr(gateway_cli, "_get_service_pids", lambda: set())
|
||||
monkeypatch.setattr(gateway_cli, "_get_ancestor_pids", lambda: {999})
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
return subprocess.CompletedProcess(
|
||||
cmd, 0,
|
||||
stdout=(
|
||||
"user 100 0.0 0.0 0 0 ? S 00:00 0:00 python gateway/run.py\n"
|
||||
"user 200 0.0 0.0 0 0 ? S 00:00 0:00 python gateway/run.py\n"
|
||||
"100 python gateway/run.py\n"
|
||||
"200 python gateway/run.py\n"
|
||||
),
|
||||
stderr="",
|
||||
)
|
||||
@@ -1102,7 +1113,7 @@ class TestFindGatewayPidsExclude:
|
||||
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
||||
monkeypatch.setattr("os.getpid", lambda: 999)
|
||||
|
||||
pids = gateway_cli.find_gateway_pids()
|
||||
pids = gateway_cli.find_gateway_pids(all_profiles=True)
|
||||
assert 100 in pids
|
||||
assert 200 in pids
|
||||
|
||||
@@ -1111,6 +1122,10 @@ class TestFindGatewayPidsExclude:
|
||||
profile_dir.mkdir(parents=True)
|
||||
monkeypatch.setattr(gateway_cli, "is_windows", lambda: False)
|
||||
monkeypatch.setattr(gateway_cli, "get_hermes_home", lambda: profile_dir)
|
||||
# Bypass /proc scan so the subprocess (ps) fallback is used
|
||||
_real_isdir = os.path.isdir
|
||||
monkeypatch.setattr("os.path.isdir", lambda p: False if p == "/proc" else _real_isdir(p))
|
||||
monkeypatch.setattr(gateway_cli, "_get_ancestor_pids", lambda: {999})
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
return subprocess.CompletedProcess(
|
||||
|
||||
@@ -19,6 +19,8 @@ The fix:
|
||||
|
||||
These tests pin the corrected behavior.
|
||||
"""
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
@@ -67,6 +69,53 @@ def test_minimax_login_does_not_launch_anthropic_flow():
|
||||
assert body["expires_in"] == 600
|
||||
|
||||
|
||||
def test_minimax_dashboard_poller_accepts_absolute_ms_expired_in():
|
||||
"""Dashboard MiniMax completion must accept unix-ms token expiry values."""
|
||||
from hermes_cli import web_server as ws
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
abs_ms = int((now.timestamp() + 1800) * 1000)
|
||||
session_id = "minimax-absolute-ms-test"
|
||||
ws._oauth_sessions[session_id] = {
|
||||
"session_id": session_id,
|
||||
"provider": "minimax-oauth",
|
||||
"flow": "device_code",
|
||||
"created_at": time.time(),
|
||||
"status": "pending",
|
||||
"error_message": None,
|
||||
"portal_base_url": "https://api.minimax.io",
|
||||
"client_id": "client-id",
|
||||
"user_code": "ABCD-1234",
|
||||
"code_verifier": "verifier",
|
||||
"interval_ms": 2000,
|
||||
"expired_in_raw": abs_ms,
|
||||
"region": "global",
|
||||
}
|
||||
captured_state = {}
|
||||
|
||||
try:
|
||||
with patch(
|
||||
"hermes_cli.auth._minimax_poll_token",
|
||||
return_value={
|
||||
"status": "success",
|
||||
"access_token": "access",
|
||||
"refresh_token": "refresh",
|
||||
"expired_in": abs_ms,
|
||||
"token_type": "Bearer",
|
||||
},
|
||||
), patch(
|
||||
"hermes_cli.auth._minimax_save_auth_state",
|
||||
side_effect=lambda state: captured_state.update(state),
|
||||
):
|
||||
ws._minimax_poller(session_id)
|
||||
finally:
|
||||
ws._oauth_sessions.pop(session_id, None)
|
||||
|
||||
assert captured_state["access_token"] == "access"
|
||||
assert 1790 <= captured_state["expires_in"] <= 1810
|
||||
assert datetime.fromisoformat(captured_state["expires_at"]).year < 9999
|
||||
|
||||
|
||||
def test_anthropic_pkce_branch_still_works():
|
||||
"""Sanity: the dispatcher tightening doesn't break the legitimate Anthropic PKCE path."""
|
||||
fake_anthropic_response = {
|
||||
|
||||
Reference in New Issue
Block a user