85cd82f1dd
Change-detectors, tautologies, source-reading tests, redundant duplicates, mock-echo tests and dead/unrunnable tests. Per-test rationale in the lane ledger (category + reason for every removal).
940 lines
44 KiB
Python
940 lines
44 KiB
Python
"""Tests for plugins/memory/honcho/cli.py."""
|
|
|
|
from types import SimpleNamespace
|
|
import json
|
|
|
|
import pytest
|
|
|
|
|
|
class TestResolveApiKey:
|
|
"""Test _resolve_api_key with various config shapes."""
|
|
|
|
def test_returns_api_key_from_root(self, monkeypatch):
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
|
|
monkeypatch.delenv("HONCHO_API_KEY", raising=False)
|
|
assert honcho_cli._resolve_api_key({"apiKey": "root-key"}) == "root-key"
|
|
|
|
|
|
def test_rejects_garbage_base_url_without_scheme(self, monkeypatch):
|
|
"""Obvious non-URL literals in baseUrl (typos) must not pass the guard."""
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
|
|
monkeypatch.delenv("HONCHO_API_KEY", raising=False)
|
|
monkeypatch.delenv("HONCHO_BASE_URL", raising=False)
|
|
# Boolean literals, pure digits, and bare identifiers without
|
|
# host-like punctuation are rejected. Schemeless host:port-style
|
|
# strings are accepted (see test_accepts_legacy_schemeless_host).
|
|
for garbage in ("true", "false", "null", "1", "12345", "localhost"):
|
|
assert honcho_cli._resolve_api_key({"baseUrl": garbage}) == "", \
|
|
f"expected empty for garbage {garbage!r}"
|
|
|
|
# file:/// parses with scheme='file' but empty netloc, so the
|
|
# http/https guard rejects; the schemeless fallback also rejects
|
|
# because 'file:' starts with a known-non-http scheme prefix.
|
|
# ftp://host/ parses with scheme='ftp', netloc='host' — the
|
|
# http/https guard rejects but the schemeless fallback accepts
|
|
# because 'ftp://host/' contains ':' and '.'. Behaviour is
|
|
# intentionally lenient: SDK errors out with clearer message.
|
|
|
|
def test_accepts_https_base_url(self, monkeypatch):
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
|
|
monkeypatch.delenv("HONCHO_API_KEY", raising=False)
|
|
monkeypatch.delenv("HONCHO_BASE_URL", raising=False)
|
|
assert honcho_cli._resolve_api_key({"baseUrl": "https://honcho.example.com"}) == "local"
|
|
|
|
|
|
class TestCmdSetupLocalJwt:
|
|
"""Local-deployment setup must allow configuring a JWT for AUTH_JWT_SECRET-backed Honcho servers."""
|
|
|
|
def _run_setup(self, monkeypatch, tmp_path, initial_cfg, prompt_answers):
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
|
|
# Avoid touching real config / SDK / filesystem.
|
|
cfg_path = tmp_path / "honcho.json"
|
|
monkeypatch.setattr(honcho_cli, "_read_config", lambda: dict(initial_cfg))
|
|
monkeypatch.setattr(honcho_cli, "_local_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
|
|
monkeypatch.setattr(honcho_cli, "_ensure_sdk_installed", lambda: True)
|
|
# No gateway import, config.yaml write or real SDK connection attempt.
|
|
monkeypatch.setattr(honcho_cli, "_gateway_platforms", lambda: [])
|
|
monkeypatch.setattr("hermes_cli.config.load_config", lambda: {"memory": {}}, raising=False)
|
|
monkeypatch.setattr("hermes_cli.config.save_config", lambda c: None, raising=False)
|
|
|
|
def _offline(*a, **k):
|
|
raise ConnectionError("offline in tests")
|
|
|
|
monkeypatch.setattr(honcho_cli, "_connect", _offline)
|
|
|
|
written = {}
|
|
|
|
def _capture_write(cfg, path=None):
|
|
written["cfg"] = cfg
|
|
written["path"] = path
|
|
|
|
monkeypatch.setattr(honcho_cli, "_write_config", _capture_write)
|
|
|
|
# Feed scripted prompt answers in order.
|
|
answers = list(prompt_answers)
|
|
|
|
def _fake_prompt(label, default=None, secret=False):
|
|
if not answers:
|
|
# Default-through any remaining prompts to keep the wizard moving.
|
|
return default or ""
|
|
return answers.pop(0)
|
|
|
|
monkeypatch.setattr(honcho_cli, "_prompt", _fake_prompt)
|
|
|
|
honcho_cli.cmd_setup(SimpleNamespace())
|
|
return written.get("cfg")
|
|
|
|
def test_local_setup_stores_jwt_under_host_block(self, monkeypatch, tmp_path):
|
|
"""Self-hosted users supplying a JWT must have it written under hosts.<host>.apiKey,
|
|
not as the top-level cloud apiKey, so cloud/hybrid switching is preserved and
|
|
get_honcho_client treats it as an explicit local auth opt-in."""
|
|
cfg = self._run_setup(
|
|
monkeypatch,
|
|
tmp_path,
|
|
initial_cfg={},
|
|
prompt_answers=[
|
|
"local", # deployment
|
|
"http://localhost:8000", # base URL
|
|
"my-local-jwt-token", # local JWT
|
|
],
|
|
)
|
|
assert cfg is not None
|
|
assert cfg.get("baseUrl") == "http://localhost:8000"
|
|
# Top-level apiKey must remain unset (cloud field).
|
|
assert not cfg.get("apiKey")
|
|
# The new local JWT belongs under the host block.
|
|
host_block = (cfg.get("hosts") or {}).get("hermes") or {}
|
|
assert host_block.get("apiKey") == "my-local-jwt-token"
|
|
|
|
|
|
class TestCmdStatus:
|
|
def test_reports_connection_failure_when_session_setup_fails(self, monkeypatch, capsys, tmp_path):
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
|
|
cfg_path = tmp_path / "honcho.json"
|
|
cfg_path.write_text("{}")
|
|
|
|
class FakeConfig:
|
|
enabled = True
|
|
api_key = "root-key"
|
|
workspace_id = "hermes"
|
|
host = "hermes"
|
|
base_url = None
|
|
ai_peer = "hermes"
|
|
peer_name = "eri"
|
|
recall_mode = "hybrid"
|
|
user_observe_me = True
|
|
user_observe_others = False
|
|
ai_observe_me = False
|
|
ai_observe_others = True
|
|
write_frequency = "async"
|
|
session_strategy = "per-session"
|
|
context_tokens = 800
|
|
dialectic_reasoning_level = "low"
|
|
reasoning_level_cap = "high"
|
|
reasoning_heuristic = True
|
|
|
|
def resolve_session_name(self):
|
|
return "hermes"
|
|
|
|
monkeypatch.setattr(honcho_cli, "_read_config", lambda: {"apiKey": "***"})
|
|
monkeypatch.setattr(honcho_cli, "_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_local_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_active_profile_name", lambda: "default")
|
|
monkeypatch.setattr(
|
|
"plugins.memory.honcho.client.HonchoClientConfig.from_global_config",
|
|
lambda host=None: FakeConfig(),
|
|
)
|
|
monkeypatch.setattr(
|
|
"plugins.memory.honcho.client.get_honcho_client",
|
|
lambda cfg: object(),
|
|
)
|
|
|
|
def _boom(hcfg, client):
|
|
raise RuntimeError("Invalid API key")
|
|
|
|
monkeypatch.setattr(honcho_cli, "_show_peer_cards", _boom)
|
|
monkeypatch.setitem(__import__("sys").modules, "honcho", SimpleNamespace())
|
|
|
|
honcho_cli.cmd_status(SimpleNamespace(all=False))
|
|
|
|
out = capsys.readouterr().out
|
|
assert "FAILED (Invalid API key)" in out
|
|
assert "Connection... OK" not in out
|
|
|
|
|
|
|
|
class TestCloneHonchoForProfile:
|
|
"""Identity-key carryover during profile cloning.
|
|
|
|
The host-scoped identity-mapping keys (``userPeerAliases``,
|
|
``runtimePeerPrefix``, ``pinUserPeer``) must survive a clone; otherwise
|
|
the new profile silently fragments memory by resolving gateway users to
|
|
raw runtime IDs instead of operator-declared peers.
|
|
"""
|
|
|
|
def _setup_clone_env(self, monkeypatch, tmp_path, cfg):
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
cfg_path = tmp_path / "config.json"
|
|
cfg_path.write_text("{}")
|
|
monkeypatch.setattr(honcho_cli, "_read_config", lambda: cfg)
|
|
monkeypatch.setattr(honcho_cli, "_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_local_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_ensure_peer_exists", lambda host_key=None: True)
|
|
written = {}
|
|
def _write(c, path=None):
|
|
written["cfg"] = c
|
|
monkeypatch.setattr(honcho_cli, "_write_config", _write)
|
|
return honcho_cli, written
|
|
|
|
def test_user_peer_aliases_carry_into_cloned_profile(self, monkeypatch, tmp_path):
|
|
cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {
|
|
"hermes": {
|
|
"userPeerAliases": {"7654321": "eri", "discord-491827364": "eri"},
|
|
"peerName": "eri",
|
|
},
|
|
},
|
|
}
|
|
honcho_cli, written = self._setup_clone_env(monkeypatch, tmp_path, cfg)
|
|
ok = honcho_cli.clone_honcho_for_profile("coder")
|
|
assert ok is True
|
|
new_block = written["cfg"]["hosts"]["hermes_coder"]
|
|
assert new_block["userPeerAliases"] == {"7654321": "eri", "discord-491827364": "eri"}
|
|
|
|
def test_runtime_peer_prefix_carries_into_cloned_profile(self, monkeypatch, tmp_path):
|
|
cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {
|
|
"hermes": {
|
|
"runtimePeerPrefix": "telegram_",
|
|
"peerName": "eri",
|
|
},
|
|
},
|
|
}
|
|
honcho_cli, written = self._setup_clone_env(monkeypatch, tmp_path, cfg)
|
|
ok = honcho_cli.clone_honcho_for_profile("coder")
|
|
assert ok is True
|
|
new_block = written["cfg"]["hosts"]["hermes_coder"]
|
|
assert new_block["runtimePeerPrefix"] == "telegram_"
|
|
|
|
def test_session_ai_peer_prefix_carries_into_cloned_profile(self, monkeypatch, tmp_path):
|
|
cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {
|
|
"hermes": {
|
|
"sessionAiPeerPrefix": True,
|
|
"peerName": "eri",
|
|
},
|
|
},
|
|
}
|
|
honcho_cli, written = self._setup_clone_env(monkeypatch, tmp_path, cfg)
|
|
ok = honcho_cli.clone_honcho_for_profile("coder")
|
|
assert ok is True
|
|
new_block = written["cfg"]["hosts"]["hermes_coder"]
|
|
assert new_block["sessionAiPeerPrefix"] is True
|
|
|
|
def test_legacy_pin_peer_name_migrates_to_canonical_on_clone(self, monkeypatch, tmp_path):
|
|
cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {
|
|
"hermes": {
|
|
"pinPeerName": True,
|
|
"peerName": "eri",
|
|
},
|
|
},
|
|
}
|
|
honcho_cli, written = self._setup_clone_env(monkeypatch, tmp_path, cfg)
|
|
ok = honcho_cli.clone_honcho_for_profile("coder")
|
|
assert ok is True
|
|
new_block = written["cfg"]["hosts"]["hermes_coder"]
|
|
assert new_block["pinUserPeer"] is True
|
|
assert "pinPeerName" not in new_block
|
|
|
|
def test_unset_identity_keys_do_not_appear_in_cloned_profile(self, monkeypatch, tmp_path):
|
|
cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {"hermes": {"peerName": "eri"}},
|
|
}
|
|
honcho_cli, written = self._setup_clone_env(monkeypatch, tmp_path, cfg)
|
|
ok = honcho_cli.clone_honcho_for_profile("coder")
|
|
assert ok is True
|
|
new_block = written["cfg"]["hosts"]["hermes_coder"]
|
|
assert "userPeerAliases" not in new_block
|
|
assert "runtimePeerPrefix" not in new_block
|
|
assert "pinUserPeer" not in new_block
|
|
assert "pinPeerName" not in new_block
|
|
|
|
|
|
class TestSetupWizardDeploymentShape:
|
|
"""The gateway identity-mapping tree writes pinUserPeer / userPeerAliases /
|
|
runtimePeerPrefix based on the operator's intent.
|
|
|
|
Choice [1] (just me) collapses all platforms to peerName.
|
|
Choice [3] (only other people) leaves the resolver to route per-runtime.
|
|
Choice [2] (me + others, pooled) aliases the operator's own runtime IDs.
|
|
|
|
These tests mock gateway detection and script the interactive _prompt
|
|
calls, asserting the resulting hermes_host block so the tree's routing
|
|
semantics stay locked even as adjacent prompts are added.
|
|
"""
|
|
|
|
def _run_setup(self, monkeypatch, tmp_path, *, answers, initial_cfg=None,
|
|
gateway_platforms=("telegram",)):
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
|
|
cfg_path = tmp_path / "config.json"
|
|
cfg_path.write_text("{}")
|
|
cfg = initial_cfg if initial_cfg is not None else {"apiKey": "***"}
|
|
|
|
monkeypatch.setattr(honcho_cli, "_read_config", lambda: cfg)
|
|
monkeypatch.setattr(honcho_cli, "_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_local_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
|
|
monkeypatch.setattr(honcho_cli, "_ensure_sdk_installed", lambda: True)
|
|
monkeypatch.setattr(honcho_cli, "_write_config", lambda *a, **k: None)
|
|
# No network probe / environment sniffing in tests.
|
|
monkeypatch.setattr(honcho_cli, "_device_login_available", lambda: False)
|
|
monkeypatch.setattr(honcho_cli, "_headless", lambda: (False, True))
|
|
# Gate detection is mocked so tests control whether the tree runs.
|
|
# None → undetectable; list (possibly empty) → connected platforms.
|
|
gw = None if gateway_platforms is None else list(gateway_platforms)
|
|
monkeypatch.setattr(honcho_cli, "_gateway_platforms", lambda: gw)
|
|
|
|
# Bypass config.yaml + connection test side effects.
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config", lambda: {"memory": {}}, raising=False,
|
|
)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.save_config", lambda c: None, raising=False,
|
|
)
|
|
|
|
class _FakeClientCfg:
|
|
def resolve_session_name(self):
|
|
return "hermes-test"
|
|
workspace_id = "hermes"
|
|
peer_name = "eri"
|
|
ai_peer = "hermetika"
|
|
observation_mode = "directional"
|
|
write_frequency = "async"
|
|
recall_mode = "hybrid"
|
|
session_strategy = "per-session"
|
|
|
|
monkeypatch.setattr(
|
|
"plugins.memory.honcho.client.HonchoClientConfig.from_global_config",
|
|
lambda host=None: _FakeClientCfg(),
|
|
)
|
|
monkeypatch.setattr(
|
|
"plugins.memory.honcho.client.reset_honcho_client",
|
|
lambda: None,
|
|
)
|
|
monkeypatch.setattr(
|
|
"plugins.memory.honcho.client.get_honcho_client",
|
|
lambda hcfg: object(),
|
|
)
|
|
|
|
# Scripted _prompt: pop answers in order. Default-return for unconsumed prompts.
|
|
answer_iter = iter(answers)
|
|
def _scripted_prompt(label, default=None, secret=False):
|
|
# Auth-method prompt is orthogonal to shape; auto-answer apikey so the answer lists stay shape-only.
|
|
if "OAuth" in label:
|
|
return "apikey"
|
|
try:
|
|
return next(answer_iter)
|
|
except StopIteration:
|
|
return default if default is not None else ""
|
|
monkeypatch.setattr(honcho_cli, "_prompt", _scripted_prompt)
|
|
|
|
honcho_cli.cmd_setup(SimpleNamespace())
|
|
return cfg["hosts"]["hermes"]
|
|
|
|
def test_just_me_pins_and_clears_aliases(self, monkeypatch, tmp_path):
|
|
answers = [
|
|
"cloud", # deployment
|
|
"", # api key (keep)
|
|
"eri", # peer name
|
|
"hermetika", # ai peer
|
|
"hermes", # workspace
|
|
"1", # tree: just me ← key answer
|
|
# remaining prompts fall through to defaults
|
|
]
|
|
initial_cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {"hermes": {
|
|
"userPeerAliases": {"old": "stale"},
|
|
"runtimePeerPrefix": "old_",
|
|
}},
|
|
}
|
|
host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
|
|
assert host["pinUserPeer"] is True
|
|
assert "userPeerAliases" not in host
|
|
assert "runtimePeerPrefix" not in host
|
|
|
|
def test_only_others_leaves_pin_false_and_accepts_prefix(self, monkeypatch, tmp_path):
|
|
answers = [
|
|
"cloud", # deployment
|
|
"", # api key (keep)
|
|
"eri", # peer name
|
|
"hermetika", # ai peer
|
|
"hermes", # workspace
|
|
"3", # tree: only other people
|
|
"telegram_", # runtime peer prefix
|
|
]
|
|
host = self._run_setup(monkeypatch, tmp_path, answers=answers)
|
|
assert host["pinUserPeer"] is False
|
|
# Multi must NOT auto-write ``userPeerAliases: {}``: an empty host
|
|
# map would silently override a root-level baseline. Absence is
|
|
# the correct "no host opinion" signal.
|
|
assert "userPeerAliases" not in host
|
|
assert host["runtimePeerPrefix"] == "telegram_"
|
|
|
|
def test_pooled_aliases_operator_runtime_ids_to_peer_name(self, monkeypatch, tmp_path):
|
|
answers = [
|
|
"cloud", # deployment
|
|
"", # api key (keep)
|
|
"eri", # peer name
|
|
"hermetika", # ai peer
|
|
"hermes", # workspace
|
|
"2", # tree: me + other people
|
|
"y", # keep my memory pooled? → hybrid
|
|
"7654321", # telegram uid
|
|
"491827364", # discord snowflake
|
|
"", # slack (skip)
|
|
"", # matrix (skip)
|
|
"", # runtime peer prefix (skip)
|
|
]
|
|
host = self._run_setup(monkeypatch, tmp_path, answers=answers)
|
|
assert host["pinUserPeer"] is False
|
|
assert host["userPeerAliases"] == {
|
|
"7654321": "eri",
|
|
"491827364": "eri",
|
|
}
|
|
assert "runtimePeerPrefix" not in host
|
|
|
|
def test_skip_shape_preserves_existing_identity_config(self, monkeypatch, tmp_path):
|
|
# Seeds the legacy ``pinPeerName``: skip must leave the mapping intact
|
|
# except for the on-load migration onto the canonical key.
|
|
initial_cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {"hermes": {
|
|
"pinPeerName": True,
|
|
"userPeerAliases": {"keep": "me"},
|
|
"runtimePeerPrefix": "keep_",
|
|
}},
|
|
}
|
|
answers = [
|
|
"cloud", "", "eri", "hermetika", "hermes", "s",
|
|
]
|
|
host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
|
|
assert host["pinUserPeer"] is True
|
|
assert "pinPeerName" not in host
|
|
assert host["userPeerAliases"] == {"keep": "me"}
|
|
assert host["runtimePeerPrefix"] == "keep_"
|
|
|
|
def test_unpin_steers_to_pooled_by_default(self, monkeypatch, tmp_path):
|
|
"""Choosing 'only other people' on a currently-pinned profile triggers
|
|
the orphan warning, which auto-steers to pooled (hybrid) so the
|
|
operator's own runtime IDs keep landing on peerName.
|
|
"""
|
|
initial_cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {"hermes": {"pinPeerName": True, "peerName": "eri"}},
|
|
}
|
|
answers = [
|
|
"cloud", # deployment
|
|
"", # api key (keep)
|
|
"eri", # peer name
|
|
"hermetika", # ai peer
|
|
"hermes", # workspace
|
|
"3", # tree: only others — triggers the orphan guard
|
|
"y", # pool my own memory instead? → hybrid
|
|
"7654321", # telegram uid
|
|
"", # discord (skip)
|
|
"", # slack (skip)
|
|
"", # matrix (skip)
|
|
"", # runtime prefix (skip)
|
|
]
|
|
host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
|
|
assert host["pinUserPeer"] is False
|
|
assert host["userPeerAliases"] == {"7654321": "eri"}
|
|
|
|
|
|
|
|
def test_host_pin_user_peer_true_is_detected_as_single(self, monkeypatch, tmp_path):
|
|
"""Host-level ``pinUserPeer: true`` must classify as ``single``.
|
|
|
|
Pressing Enter at the choice prompt then preserves the pin instead
|
|
of falling through to per-user routing and orphaning the user's
|
|
memory pool — the bug the wizard regressed when ``pinUserPeer``
|
|
landed as a higher-precedence alias.
|
|
"""
|
|
initial_cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {"hermes": {"pinUserPeer": True, "peerName": "eri"}},
|
|
}
|
|
# Exhaust the iterator before the choice prompt so the scripted
|
|
# mock falls through to the prompt's default (the detected shape →
|
|
# choice "1"). Scripting an explicit "" would NOT exercise that
|
|
# fallthrough — the mock returns it literally.
|
|
answers = ["cloud", "", "eri", "hermetika", "hermes"]
|
|
host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
|
|
# Scrub-then-write normalises onto the canonical pinUserPeer.
|
|
assert host["pinUserPeer"] is True
|
|
assert "pinPeerName" not in host
|
|
|
|
|
|
def test_root_user_peer_aliases_detected_as_hybrid(self, monkeypatch, tmp_path):
|
|
"""Root-level ``userPeerAliases`` must classify as ``hybrid`` even
|
|
when the host block has no aliases of its own.
|
|
"""
|
|
initial_cfg = {
|
|
"apiKey": "***",
|
|
"userPeerAliases": {"7654321": "eri"},
|
|
"hosts": {"hermes": {"peerName": "eri"}},
|
|
}
|
|
answers = ["cloud", "", "eri", "hermetika", "hermes"]
|
|
host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
|
|
assert host["pinUserPeer"] is False
|
|
# Hybrid materialises the root aliases into the host so subsequent
|
|
# operator edits live on the host block they're inspecting.
|
|
assert host["userPeerAliases"] == {"7654321": "eri"}
|
|
|
|
@pytest.mark.parametrize("initial_cfg, expected_pin", [
|
|
(None, True),
|
|
({"apiKey": "***", "hosts": {"hermes": {}}}, True),
|
|
({"apiKey": "***", "hosts": {"hermes": {"pinUserPeer": False, "peerName": "eri"}}}, False),
|
|
({"apiKey": "***", "hosts": {"hermes": {"enabled": True, "workspace": "hermes", "peerName": "eri"}}}, False),
|
|
({"apiKey": "***", "enabled": True, "workspace": "hermes", "peerName": "eri"}, False),
|
|
], ids=["fresh-config-defaults-to-single", "empty-host-block-defaults-to-single",
|
|
"configured-multi-keeps-multi", "existing-install-without-mapping-keys-keeps-multi",
|
|
"legacy-root-level-install-keeps-multi"])
|
|
def test_choice_default_follows_config(self, monkeypatch, tmp_path, initial_cfg, expected_pin):
|
|
"""Enter on a fresh config picks the pinned personal shape. An existing install, with or
|
|
without mapping keys, keeps its detected shape so Enter never merges every account onto one peer."""
|
|
answers = ["cloud", "", "eri", "hermetika", "hermes"]
|
|
host = self._run_setup(monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg)
|
|
assert host["pinUserPeer"] is expected_pin
|
|
|
|
|
|
def test_no_gateway_connected_skips_mapping_when_declined(self, monkeypatch, tmp_path):
|
|
"""With no gateway platforms connected, the tree is gated off; declining
|
|
the 'configure anyway?' prompt leaves identity mapping untouched."""
|
|
initial_cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {"hermes": {"peerName": "eri"}},
|
|
}
|
|
answers = ["cloud", "", "eri", "hermetika", "hermes", "n"]
|
|
host = self._run_setup(
|
|
monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg,
|
|
gateway_platforms=[],
|
|
)
|
|
assert "pinUserPeer" not in host
|
|
assert "userPeerAliases" not in host
|
|
assert "runtimePeerPrefix" not in host
|
|
|
|
def test_undetectable_gateway_skips_mapping_when_declined(self, monkeypatch, tmp_path):
|
|
"""When the gateway package can't be inspected (None), the wizard asks
|
|
whether the gateway is running; 'no' skips the mapping step."""
|
|
initial_cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {"hermes": {"peerName": "eri"}},
|
|
}
|
|
answers = ["cloud", "", "eri", "hermetika", "hermes", "n"]
|
|
host = self._run_setup(
|
|
monkeypatch, tmp_path, answers=answers, initial_cfg=initial_cfg,
|
|
gateway_platforms=None,
|
|
)
|
|
assert "pinUserPeer" not in host
|
|
|
|
def test_raw_edit_sets_resolver_knobs_directly(self, monkeypatch, tmp_path):
|
|
"""The [e] escape hatch lets a power user set pinUserPeer + an alias +
|
|
prefix directly, bypassing the intent tree."""
|
|
answers = [
|
|
"cloud", "", "eri", "hermetika", "hermes",
|
|
"e", # tree: edit raw keys
|
|
"false", # pinUserPeer
|
|
"99887766=eri", # one alias pair
|
|
"", # finish aliases
|
|
"discord_", # runtimePeerPrefix
|
|
]
|
|
host = self._run_setup(monkeypatch, tmp_path, answers=answers)
|
|
assert host["pinUserPeer"] is False
|
|
assert host["userPeerAliases"] == {"99887766": "eri"}
|
|
assert host["runtimePeerPrefix"] == "discord_"
|
|
|
|
|
|
class TestCloneCarriesPinUserPeer:
|
|
"""``pinUserPeer`` (canonical name for ``pinPeerName``) must survive a
|
|
profile clone. Without this, a default profile that uses the newer
|
|
key would silently produce cloned profiles without the pin even
|
|
though the resolver prefers ``pinUserPeer`` over ``pinPeerName``.
|
|
"""
|
|
|
|
def test_clone_inherits_host_pin_user_peer(self, monkeypatch, tmp_path):
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
|
|
cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {"hermes": {"pinUserPeer": True, "peerName": "eri"}},
|
|
}
|
|
cfg_path = tmp_path / "config.json"
|
|
cfg_path.write_text("{}")
|
|
monkeypatch.setattr(honcho_cli, "_read_config", lambda: cfg)
|
|
monkeypatch.setattr(honcho_cli, "_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_local_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_ensure_peer_exists", lambda host_key=None: True)
|
|
written = {}
|
|
monkeypatch.setattr(
|
|
honcho_cli, "_write_config", lambda c, path=None: written.setdefault("cfg", c),
|
|
)
|
|
|
|
ok = honcho_cli.clone_honcho_for_profile("partner")
|
|
assert ok is True
|
|
new_block = written["cfg"]["hosts"]["hermes_partner"]
|
|
assert new_block["pinUserPeer"] is True
|
|
|
|
|
|
class TestMigratePinKey:
|
|
"""``_migrate_pin_key`` rewrites the legacy ``pinPeerName`` onto the
|
|
canonical ``pinUserPeer`` in place, without clobbering an existing
|
|
canonical value."""
|
|
|
|
|
|
def test_canonical_key_wins_when_both_present(self):
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
block = {"pinPeerName": True, "pinUserPeer": False}
|
|
assert honcho_cli._migrate_pin_key(block) is True
|
|
assert block == {"pinUserPeer": False}
|
|
|
|
def test_noop_when_no_legacy_key(self):
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
block = {"pinUserPeer": True}
|
|
assert honcho_cli._migrate_pin_key(block) is False
|
|
assert block == {"pinUserPeer": True}
|
|
|
|
|
|
class TestCmdSetupDeviceFlow:
|
|
"""The cloud auth-method menu's device-code branch (RFC 8628)."""
|
|
|
|
def _run_setup(self, monkeypatch, tmp_path, *, answers, device_available=True,
|
|
headless=(False, True), device_result=None, device_error=None):
|
|
"""Run cmd_setup with the device flow stubbed; returns (cfg, calls, prompts)."""
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
import plugins.memory.honcho.oauth_flow as oauth_flow
|
|
from plugins.memory.honcho.oauth import OAuthCredential
|
|
|
|
cfg_path = tmp_path / "config.json"
|
|
cfg_path.write_text("{}")
|
|
cfg = {"apiKey": "***"}
|
|
|
|
monkeypatch.setattr(honcho_cli, "_read_config", lambda: cfg)
|
|
monkeypatch.setattr(honcho_cli, "_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_local_config_path", lambda: cfg_path)
|
|
monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
|
|
monkeypatch.setattr(honcho_cli, "_ensure_sdk_installed", lambda: True)
|
|
monkeypatch.setattr(honcho_cli, "_write_config", lambda *a, **k: None)
|
|
monkeypatch.setattr(honcho_cli, "_gateway_platforms", lambda: [])
|
|
monkeypatch.setattr(honcho_cli, "_device_login_available", lambda: device_available)
|
|
monkeypatch.setattr(honcho_cli, "_headless", lambda: headless)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config", lambda: {"memory": {}}, raising=False,
|
|
)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.save_config", lambda c: None, raising=False,
|
|
)
|
|
|
|
class _FakeClientCfg:
|
|
def resolve_session_name(self):
|
|
return "hermes-test"
|
|
workspace_id = "hermes"
|
|
peer_name = "eri"
|
|
ai_peer = "hermetika"
|
|
observation_mode = "directional"
|
|
write_frequency = "async"
|
|
recall_mode = "hybrid"
|
|
session_strategy = "per-session"
|
|
|
|
monkeypatch.setattr(
|
|
"plugins.memory.honcho.client.HonchoClientConfig.from_global_config",
|
|
lambda host=None: _FakeClientCfg(),
|
|
)
|
|
monkeypatch.setattr("plugins.memory.honcho.client.reset_honcho_client", lambda: None)
|
|
monkeypatch.setattr("plugins.memory.honcho.client.get_honcho_client", lambda hcfg: object())
|
|
|
|
calls: list[dict] = []
|
|
cred = OAuthCredential(
|
|
access_token="hch-at-x", refresh_token="hch-rt-x", expires_at=9_999_999_999,
|
|
client_id="hermes-agent", token_endpoint="http://x/oauth/token",
|
|
consent_peer_name="lyra",
|
|
)
|
|
|
|
def fake_device_flow(**kwargs):
|
|
calls.append(kwargs)
|
|
if device_error is not None:
|
|
raise device_error
|
|
return device_result or cred
|
|
|
|
monkeypatch.setattr(oauth_flow, "authorize_via_device_code", fake_device_flow)
|
|
|
|
prompts: list[tuple[str, str | None]] = []
|
|
answer_iter = iter(answers)
|
|
def _scripted_prompt(label, default=None, secret=False):
|
|
prompts.append((label, default))
|
|
try:
|
|
# Mirror the real _prompt: blank input falls back to the default.
|
|
return next(answer_iter) or (default or "")
|
|
except StopIteration:
|
|
return default if default is not None else ""
|
|
monkeypatch.setattr(honcho_cli, "_prompt", _scripted_prompt)
|
|
|
|
honcho_cli.cmd_setup(SimpleNamespace())
|
|
return cfg, calls, prompts
|
|
|
|
def test_device_choice_runs_flow_and_stores_grant(self, monkeypatch, tmp_path):
|
|
cfg, calls, _ = self._run_setup(monkeypatch, tmp_path, answers=["cloud", "device"])
|
|
assert len(calls) == 1
|
|
assert calls[0]["apply_config"] is False
|
|
host = cfg["hosts"]["hermes"]
|
|
assert host["apiKey"] == "hch-at-x"
|
|
assert host["oauth"]["refreshToken"] == "hch-rt-x"
|
|
assert host["peerName"] == "lyra"
|
|
|
|
def test_headless_defaults_to_device(self, monkeypatch, tmp_path):
|
|
# Blank answer takes the prompt default, which flips to device on a
|
|
# remote/no-browser environment.
|
|
cfg, calls, prompts = self._run_setup(
|
|
monkeypatch, tmp_path, answers=["cloud", ""], headless=(True, False),
|
|
)
|
|
method_prompts = [p for p in prompts if "apikey" in p[0]]
|
|
assert method_prompts[0][1] == "device"
|
|
assert len(calls) == 1
|
|
assert calls[0]["open_url"] is None # never auto-open a browser headless
|
|
assert cfg["hosts"]["hermes"]["apiKey"] == "hch-at-x"
|
|
|
|
def test_denied_device_flow_aborts_without_grant(self, monkeypatch, tmp_path):
|
|
from plugins.memory.honcho.oauth_flow import AccessDenied
|
|
|
|
cfg, calls, _ = self._run_setup(
|
|
monkeypatch, tmp_path, answers=["cloud", "device"],
|
|
device_error=AccessDenied("access_denied", "user denied"),
|
|
)
|
|
assert len(calls) == 1
|
|
assert "apiKey" not in cfg.get("hosts", {}).get("hermes", {})
|
|
|
|
|
|
def _point_cli_at(monkeypatch, cfg_path, **attrs):
|
|
"""Route the honcho CLI's config reads and writes at ``cfg_path``; ``attrs`` replace other module names."""
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
for name, value in {"_config_path": lambda: cfg_path, "_local_config_path": lambda: cfg_path, **attrs}.items():
|
|
monkeypatch.setattr(honcho_cli, name, value)
|
|
return honcho_cli
|
|
|
|
|
|
class TestWriteRefusesUnparseableStore:
|
|
"""An unparseable honcho.json reads as {} on the tolerant path; writing that back would drop every other host."""
|
|
|
|
@pytest.mark.parametrize("run", [
|
|
lambda cli: cli.honcho_command(SimpleNamespace(honcho_command="mode", mode="tools", target_profile=None)),
|
|
lambda cli: cli.cmd_setup(SimpleNamespace()),
|
|
], ids=["command", "setup"])
|
|
def test_command_asks_nothing_and_writes_nothing(self, monkeypatch, tmp_path, run):
|
|
cfg_path = tmp_path / "honcho.json"
|
|
cfg_path.write_text("{not json", encoding="utf-8")
|
|
honcho_cli = _point_cli_at(monkeypatch, cfg_path, _host_key=lambda: "hermes_coder",
|
|
_prompt=lambda *a, **k: pytest.fail("asked a question"))
|
|
run(honcho_cli)
|
|
assert cfg_path.read_text(encoding="utf-8") == "{not json"
|
|
|
|
|
|
class TestSetupApiKeyReplacesStaleGrant:
|
|
"""Choosing apikey after a revoked grant left hosts.<name>.oauth in place, shadowing the fresh key."""
|
|
|
|
@pytest.mark.parametrize("grant, root_key, answer, ok, host_key, root_after, shown", [
|
|
(True, None, "hch-v3-fresh", True, "hch-v3-fresh", "hch-v3-fresh", ""),
|
|
(True, "hch-v3-rootkey", "", True, "hch-v3-rootkey", "hch-v3-rootkey", "...-rootkey"),
|
|
(True, None, "", False, "hch-at-dead", None, "Current API key: not set"),
|
|
(False, None, "", True, "hch-v3-hostkey", None, ""),
|
|
], ids=["new key clears oauth", "kept root key clears oauth", "dead access token not offered", "static host key kept"])
|
|
def test_apikey_answer(self, monkeypatch, tmp_path, capsys, grant, root_key, answer, ok, host_key, root_after, shown):
|
|
host = {"apiKey": "hch-v3-hostkey"}
|
|
if grant:
|
|
host = {"apiKey": "hch-at-dead", "oauth": {"refreshToken": "hch-rt-dead", "expiresAt": 1,
|
|
"clientId": "hermes-agent", "tokenEndpoint": "https://api.honcho.dev/oauth/token"}}
|
|
cfg = {"hosts": {"hermes": host}, **({"apiKey": root_key} if root_key else {})}
|
|
honcho_cli = _point_cli_at(monkeypatch, tmp_path / "honcho.json", _device_login_available=lambda: False,
|
|
_headless=lambda: (False, True),
|
|
_prompt=lambda label, default=None, secret=False: "apikey" if "OAuth" in label else answer)
|
|
assert honcho_cli._setup_cloud_auth(cfg, host, tmp_path / "honcho.json") is ok
|
|
assert host["apiKey"] == host_key and cfg.get("apiKey") == root_after
|
|
assert ("oauth" in host) is (grant and not ok)
|
|
assert shown in capsys.readouterr().out
|
|
|
|
|
|
_OAUTH_DEFAULT = {"peerName": "eri", "hosts": {"hermes": {
|
|
"enabled": True, "apiKey": "hch-at-live", "workspace": "hermes", "peerName": "eri", "oauth": {"refreshToken": "hch-rt-live"},
|
|
}}}
|
|
_KEYLESS_DEFAULT = {"hosts": {"hermes": {"workspace": "hermes"}}}
|
|
|
|
|
|
class TestEnabledRequiresACredential:
|
|
"""A host block is written with enabled: true only when it can authenticate. Named profiles do not
|
|
inherit the default host's apiKey, so a clone of an OAuth default block has nothing to sign with."""
|
|
|
|
def _env(self, monkeypatch, tmp_path, cfg, *, env_key=None, host="hermes_dreamer", profile="dreamer"):
|
|
import copy
|
|
cfg, written = copy.deepcopy(cfg), {}
|
|
cfg_path = tmp_path / "honcho.json"
|
|
cfg_path.write_text("{}")
|
|
monkeypatch.delenv("HONCHO_API_KEY", raising=False)
|
|
monkeypatch.delenv("HONCHO_BASE_URL", raising=False)
|
|
if env_key:
|
|
monkeypatch.setenv("HONCHO_API_KEY", env_key)
|
|
honcho_cli = _point_cli_at(
|
|
monkeypatch, cfg_path, _read_config=lambda: cfg, _host_key=lambda: host, _active_profile_name=lambda: profile,
|
|
_ensure_peer_exists=lambda host_key=None: True, _write_config=lambda c, path=None: written.setdefault("cfg", c))
|
|
return honcho_cli, written
|
|
|
|
@pytest.mark.parametrize("cfg, env_key, enabled", [
|
|
(_OAUTH_DEFAULT, None, False),
|
|
({"hosts": {"hermes": {"enabled": True, "apiKey": "hch-v3-hostonly", "workspace": "hermes"}}}, None, False),
|
|
({"apiKey": "hch-v3-root", **_KEYLESS_DEFAULT}, None, True),
|
|
(_KEYLESS_DEFAULT, "hch-v3-from-env", False),
|
|
({"baseUrl": "http://localhost:8000", **_KEYLESS_DEFAULT}, None, True),
|
|
], ids=["oauth default", "host-only static key", "root key", "env key only", "self-hosted url"])
|
|
def test_clone_is_enabled_only_by_an_on_disk_credential(self, monkeypatch, tmp_path, cfg, env_key, enabled):
|
|
honcho_cli, written = self._env(monkeypatch, tmp_path, cfg, env_key=env_key)
|
|
assert honcho_cli.clone_honcho_for_profile("dreamer") is True
|
|
block = written["cfg"]["hosts"]["hermes_dreamer"]
|
|
assert block.get("enabled") is (True if enabled else None)
|
|
assert "apiKey" not in block and "oauth" not in block
|
|
assert block["aiPeer"] == "dreamer" and block["workspace"] == "hermes"
|
|
|
|
@pytest.mark.parametrize("cfg, env_key, profile, enabled", [
|
|
({"hosts": {"hermes_dreamer": {"workspace": "hermes"}}}, "hch-v3-from-env", "dreamer", False),
|
|
(_OAUTH_DEFAULT, None, "dreamer", False),
|
|
({"hosts": {"hermes_dreamer": {"enabled": True, "aiPeer": "dreamer", "workspace": "hermes"}}}, None, "dreamer",
|
|
False),
|
|
({"hosts": {}}, None, "default", False),
|
|
({"apiKey": "hch-v3-root", **_KEYLESS_DEFAULT}, None, "dreamer", True),
|
|
], ids=["env key only", "empty block", "legacy enabled keyless block", "default profile", "root key"])
|
|
def test_enable_writes_enabled_only_for_an_on_disk_credential(self, monkeypatch, tmp_path,
|
|
cfg, env_key, profile, enabled):
|
|
host = "hermes" if profile == "default" else "hermes_dreamer"
|
|
honcho_cli, written = self._env(monkeypatch, tmp_path, cfg, env_key=env_key, host=host, profile=profile)
|
|
honcho_cli.cmd_enable(SimpleNamespace())
|
|
assert written["cfg"]["hosts"][host]["enabled"] is True if enabled else written == {}
|
|
|
|
|
|
class TestWriteConfigMergesOntoDisk:
|
|
"""A refresh in another process may rotate the token while a command runs; the write must keep it."""
|
|
|
|
def _paths(self, monkeypatch, tmp_path, disk):
|
|
cfg_path = tmp_path / "honcho.json"
|
|
cfg_path.write_text(json.dumps(disk))
|
|
return _point_cli_at(monkeypatch, cfg_path), cfg_path
|
|
|
|
def _rotate_on_disk(self, cfg_path):
|
|
disk = json.loads(cfg_path.read_text())
|
|
disk["hosts"]["hermes"].update(apiKey="hch-at-new", oauth={"refreshToken": "hch-rt-new"})
|
|
cfg_path.write_text(json.dumps(disk))
|
|
|
|
def test_untouched_keys_take_disk_and_the_commands_edits_apply(self, monkeypatch, tmp_path):
|
|
disk = {"apiKey": "root", "hosts": {"hermes": {"apiKey": "hch-at-old", "oauth": {"refreshToken": "hch-rt-old"},
|
|
"recallMode": "hybrid", "runtimePeerPrefix": "tg_"}}}
|
|
honcho_cli, cfg_path = self._paths(monkeypatch, tmp_path, disk)
|
|
cfg = honcho_cli._read_config()
|
|
self._rotate_on_disk(cfg_path)
|
|
cfg["hosts"]["hermes"]["recallMode"] = "tools"
|
|
cfg["hosts"]["hermes"].pop("runtimePeerPrefix")
|
|
cfg["dialecticCadence"] = 3
|
|
honcho_cli._write_config(cfg)
|
|
out = json.loads(cfg_path.read_text())
|
|
assert out["hosts"]["hermes"] == {"apiKey": "hch-at-new", "oauth": {"refreshToken": "hch-rt-new"}, "recallMode": "tools"}
|
|
assert out["apiKey"] == "root" and out["dialecticCadence"] == 3
|
|
|
|
def test_a_credential_the_command_set_wins(self, monkeypatch, tmp_path):
|
|
disk = {"hosts": {"hermes": {"apiKey": "hch-at-old", "oauth": {"refreshToken": "hch-rt-old"}}}}
|
|
honcho_cli, cfg_path = self._paths(monkeypatch, tmp_path, disk)
|
|
cfg = honcho_cli._read_config()
|
|
self._rotate_on_disk(cfg_path)
|
|
cfg["hosts"]["hermes"]["apiKey"] = "hch-v3-pasted"
|
|
cfg["hosts"]["hermes"].pop("oauth")
|
|
honcho_cli._write_config(cfg)
|
|
assert json.loads(cfg_path.read_text())["hosts"]["hermes"] == {"apiKey": "hch-v3-pasted"}
|
|
|
|
def test_a_second_write_on_the_same_read_applies_only_the_edits_made_since_the_first(self, monkeypatch, tmp_path):
|
|
disk = {"hosts": {"hermes": {"apiKey": "hch-at-old", "oauth": {"refreshToken": "hch-rt-old"}, "workspace": "A"}}}
|
|
honcho_cli, cfg_path = self._paths(monkeypatch, tmp_path, disk)
|
|
cfg = honcho_cli._read_config()
|
|
cfg["hosts"]["hermes"]["workspace"] = "B"
|
|
honcho_cli._write_config(cfg)
|
|
self._rotate_on_disk(cfg_path)
|
|
cfg["hosts"]["hermes"]["workspace"] = "A"
|
|
honcho_cli._write_config(cfg)
|
|
out = json.loads(cfg_path.read_text())["hosts"]["hermes"]
|
|
assert out == {"apiKey": "hch-at-new", "oauth": {"refreshToken": "hch-rt-new"}, "workspace": "A"}
|
|
|
|
def test_a_grant_the_login_installed_yields_to_a_later_rotation(self, monkeypatch, tmp_path):
|
|
import plugins.memory.honcho.oauth as oauth
|
|
honcho_cli, cfg_path = self._paths(monkeypatch, tmp_path, {"hosts": {"hermes": {"peerName": "alice"}}})
|
|
monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
|
|
cfg = honcho_cli._read_config()
|
|
grant = {"access_token": "hch-at-login", "refresh_token": "hch-rt-login", "expires_in": 3600}
|
|
cred = oauth.install_grant(cfg_path, "hermes", grant, client_id="c", token_endpoint="e", apply_config=False)
|
|
honcho_cli._apply_grant_to_host(cfg, cfg["hosts"]["hermes"], cred)
|
|
self._rotate_on_disk(cfg_path)
|
|
cfg["hosts"]["hermes"]["recallMode"] = "tools"
|
|
honcho_cli._write_config(cfg)
|
|
out = json.loads(cfg_path.read_text())["hosts"]["hermes"]
|
|
assert out["apiKey"] == "hch-at-new" and out["oauth"] == {"refreshToken": "hch-rt-new"}
|
|
assert out["peerName"] == "alice" and out["recallMode"] == "tools"
|
|
|
|
_SEED = {"dialecticCadence": 3, "hosts": {"hermes": {"peerName": "alice"}}}
|
|
|
|
def _seeded(self, monkeypatch, tmp_path):
|
|
seed, local = tmp_path / "seed.json", tmp_path / "honcho.json"
|
|
seed.write_text(json.dumps(self._SEED))
|
|
return _point_cli_at(monkeypatch, local, _config_path=lambda: seed, _host_key=lambda: "hermes"), local
|
|
|
|
def test_a_read_seeded_from_another_file_applies_only_its_edits_onto_the_local_file(self, monkeypatch, tmp_path):
|
|
import plugins.memory.honcho.oauth as oauth
|
|
honcho_cli, local = self._seeded(monkeypatch, tmp_path)
|
|
cfg = honcho_cli._read_config()
|
|
grant = {"access_token": "hch-at-login", "refresh_token": "hch-rt-login", "expires_in": 3600}
|
|
cred = oauth.install_grant(local, "hermes", grant, client_id="c", token_endpoint="e", apply_config=False)
|
|
honcho_cli._apply_grant_to_host(cfg, cfg["hosts"]["hermes"], cred)
|
|
rotated = oauth.OAuthCredential.from_token_response(
|
|
{"access_token": "hch-at-new", "refresh_token": "hch-rt-new", "expires_in": 3600},
|
|
now=0.0, client_id="c", token_endpoint="e")
|
|
oauth._persist_credential(local, "hermes", rotated)
|
|
cfg["hosts"]["hermes"]["recallMode"] = "tools"
|
|
honcho_cli._write_config(cfg)
|
|
out = json.loads(local.read_text())
|
|
assert out["hosts"]["hermes"]["apiKey"] == "hch-at-new"
|
|
assert out["hosts"]["hermes"]["oauth"]["refreshToken"] == "hch-rt-new"
|
|
assert out["hosts"]["hermes"]["recallMode"] == "tools" and out["hosts"]["hermes"]["peerName"] == "alice"
|
|
assert out["dialecticCadence"] == 3
|
|
|
|
def test_a_read_seeded_from_another_file_is_written_whole_while_no_local_file_exists(self, monkeypatch, tmp_path):
|
|
honcho_cli, local = self._seeded(monkeypatch, tmp_path)
|
|
cfg = honcho_cli._read_config()
|
|
cfg["hosts"]["hermes"]["recallMode"] = "tools"
|
|
honcho_cli._write_config(cfg)
|
|
assert json.loads(local.read_text()) == {"dialecticCadence": 3, "hosts": {"hermes": {"peerName": "alice", "recallMode": "tools"}}}
|
|
|
|
@pytest.mark.parametrize("build", [lambda cli: {"hosts": {"other": {"apiKey": "o"}}}, lambda cli: dict(cli._read_config())],
|
|
ids=["never read", "rebuilt from the read"])
|
|
def test_a_plain_dict_is_written_whole(self, monkeypatch, tmp_path, build):
|
|
honcho_cli, cfg_path = self._paths(monkeypatch, tmp_path, {"hosts": {"hermes": {"apiKey": "hch-at-old"}}})
|
|
cfg = build(honcho_cli)
|
|
self._rotate_on_disk(cfg_path)
|
|
honcho_cli._write_config(cfg)
|
|
assert json.loads(cfg_path.read_text()) == cfg
|