Erosika
dc7c8673d1
Extends the read-only 'hermes honcho peers' view with a 'map' action that joins two sources: workspace peers fetched from the Honcho API, labeled from local config (your peer, each profile's AI peer, alias targets, runtime peers of seen accounts, user-* fallback peers, 'unrecognized' otherwise), and the gateway accounts recorded in state.db with what each currently resolves to. Targets are picked from the workspace list so a typo cannot silently create a peer; every assignment states its consequence (aliases move future messages only; a runtime peer left behind keeps its history). 'w' lists every workspace the key can reach — the wrong-workspace fallback — and can repoint the profile's workspace on explicit confirmation. With multiple profiles, saving a root-cascading map asks whether to write root or fork the host block, root writes warn when sibling profiles sit on other workspaces, and the accounts table marks siblings that resolve an account differently. Offline the command degrades to typed targets over the local account list. The setup wizard's gateway step closes by pointing at the command.
998 lines
47 KiB
Python
998 lines
47 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)
|
|
|
|
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
|
|
|
|
def test_auth_line_detects_oauth_grant(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 = "hch-at-deadbeef"
|
|
workspace_id = "claude-code"
|
|
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 = None
|
|
dialectic_reasoning_level = "low"
|
|
reasoning_level_cap = "high"
|
|
reasoning_heuristic = True
|
|
raw = {
|
|
"hosts": {
|
|
"hermes": {
|
|
"apiKey": "hch-at-deadbeef",
|
|
"oauth": {
|
|
"refreshToken": "hch-rt-x",
|
|
"clientId": "hermes-agent",
|
|
"tokenEndpoint": "https://api.honcho.dev/oauth/token",
|
|
"expiresAt": 9999999999,
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
def resolve_session_name(self):
|
|
return "hermes"
|
|
|
|
monkeypatch.setattr(honcho_cli, "_read_config", lambda: {})
|
|
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())
|
|
monkeypatch.setattr(honcho_cli, "_show_peer_cards", lambda hcfg, client: None)
|
|
monkeypatch.setitem(__import__("sys").modules, "honcho", SimpleNamespace())
|
|
|
|
honcho_cli.cmd_status(SimpleNamespace(all=False))
|
|
|
|
out = capsys.readouterr().out
|
|
assert "Auth: OAuth (hermes-agent" in out
|
|
assert "API key:" 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_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_mapping_step_points_at_peers_map(self, monkeypatch, tmp_path, capsys):
|
|
answers = [
|
|
"cloud", # deployment
|
|
"", # api key (keep)
|
|
"eri", # peer name
|
|
"hermetika", # ai peer
|
|
"hermes", # workspace
|
|
"s", # tree: skip
|
|
]
|
|
self._run_setup(monkeypatch, tmp_path, answers=answers)
|
|
out = capsys.readouterr().out
|
|
assert "hermes honcho peers map" in out
|
|
|
|
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"}
|
|
|
|
def test_fresh_config_defaults_to_single(self, monkeypatch, tmp_path):
|
|
"""No identity key anywhere → the choice prompt defaults to [1].
|
|
|
|
A solo operator pressing Enter gets the pinned personal shape
|
|
instead of silently fragmenting their gateway account away from
|
|
peerName's history.
|
|
"""
|
|
answers = ["cloud", "", "eri", "hermetika", "hermes"]
|
|
host = self._run_setup(monkeypatch, tmp_path, answers=answers)
|
|
assert host["pinUserPeer"] is True
|
|
|
|
def test_configured_multi_still_defaults_to_multi(self, monkeypatch, tmp_path):
|
|
"""An explicit pinUserPeer: false config keeps [3] as its default —
|
|
the fresh-config [1] default must not override a chosen shape."""
|
|
initial_cfg = {
|
|
"apiKey": "***",
|
|
"hosts": {"hermes": {"pinUserPeer": False, "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
|
|
|
|
|
|
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
|
|
assert calls[0]["source"] == "hermes-cli"
|
|
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_prints_one_sentence_asks_nothing_and_writes_nothing(self, monkeypatch, tmp_path, capsys, 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)
|
|
out = capsys.readouterr().out
|
|
assert "could not be read as JSON" in out and "Nothing was written" in out
|
|
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, expect, enabled", [
|
|
({"hosts": {"hermes_dreamer": {"workspace": "hermes"}}}, "hch-v3-from-env", "dreamer", ["setup"], False),
|
|
(_OAUTH_DEFAULT, None, "dreamer",
|
|
["stays disabled", "hermes honcho setup --target-profile dreamer", "hosts.hermes_dreamer"], False),
|
|
({"hosts": {"hermes_dreamer": {"enabled": True, "aiPeer": "dreamer", "workspace": "hermes"}}}, None, "dreamer",
|
|
["stays disabled"], False),
|
|
({"hosts": {}}, None, "default", ["Run 'hermes honcho setup' to sign in"], False),
|
|
({"apiKey": "hch-v3-root", **_KEYLESS_DEFAULT}, None, "dreamer", ["Honcho enabled"], True),
|
|
], ids=["env key only", "empty block", "legacy enabled keyless block", "default profile hint", "root key"])
|
|
def test_enable_writes_enabled_only_for_an_on_disk_credential(self, monkeypatch, tmp_path, capsys,
|
|
cfg, env_key, profile, expect, 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())
|
|
out = capsys.readouterr().out
|
|
assert all(s in out for s in expect) and "already enabled" not in out
|
|
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
|