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).
279 lines
14 KiB
Python
279 lines
14 KiB
Python
"""Tests for ``hermes honcho peers map``: account discovery, the resolution preview, and alias writes."""
|
|
|
|
import json
|
|
import sqlite3
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
import plugins.memory.honcho.cli as honcho_cli
|
|
from plugins.memory.honcho.cli import _preview_peer_resolution, _seen_gateway_accounts
|
|
from plugins.memory.honcho.session_peers import sanitize_peer_id
|
|
|
|
|
|
def _make_state_db(path, rows):
|
|
"""Create a minimal sessions table with only the columns the query reads."""
|
|
conn = sqlite3.connect(path)
|
|
conn.execute(
|
|
"""CREATE TABLE sessions (
|
|
id TEXT PRIMARY KEY, source TEXT, user_id TEXT, session_key TEXT,
|
|
display_name TEXT, origin_json TEXT, started_at REAL
|
|
)"""
|
|
)
|
|
conn.executemany("INSERT INTO sessions VALUES (?, ?, ?, ?, ?, ?, ?)", rows)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def _origin(user_name=None, user_id_alt=None):
|
|
return json.dumps({"user_name": user_name, "user_id_alt": user_id_alt})
|
|
|
|
|
|
def _row(user_id, platform="telegram", name="eri", started=100.0):
|
|
return (f"s{user_id}", platform, user_id, "k", None, _origin(name), started)
|
|
|
|
|
|
def _cfg(root=None, **host):
|
|
return {"apiKey": "***", **(root or {}), "hosts": {"hermes": host}}
|
|
|
|
|
|
class TestSeenGatewayAccounts:
|
|
def test_groups_orders_and_filters_rows(self, tmp_path):
|
|
db = tmp_path / "state.db"
|
|
_make_state_db(db, [
|
|
("s1", "telegram", "111", None, "Eri", _origin("eri"), 100.0), # predates session_key
|
|
("s2", "telegram", "111", "", "Eri", _origin("eri"), 200.0),
|
|
("s3", "discord", "222", "k2", "Tek DM", None, 900.0),
|
|
("s5", "cli", None, None, None, None, 999.0),
|
|
])
|
|
assert _seen_gateway_accounts(db) == [
|
|
{"platform": "discord", "user_id": "222", "user_id_alt": "", "label": "Tek DM", "sessions": 1, "profiles": []},
|
|
{"platform": "telegram", "user_id": "111", "user_id_alt": "", "label": "eri", "sessions": 2, "profiles": []},
|
|
]
|
|
|
|
def test_shared_session_lists_only_its_last_author(self, tmp_path):
|
|
"""record_gateway_session_peer overwrites the row's user_id, so earlier authors are gone."""
|
|
from hermes_state import SessionDB
|
|
|
|
db = SessionDB(tmp_path / "state.db")
|
|
for uid in ("alice", "bob"):
|
|
db.record_gateway_session_peer("shared", source="telegram", user_id=uid, session_key="telegram:group:1")
|
|
assert [a["user_id"] for a in _seen_gateway_accounts(tmp_path / "state.db")] == ["bob"]
|
|
|
|
def test_missing_db_or_table_returns_empty(self, tmp_path):
|
|
assert _seen_gateway_accounts(tmp_path / "absent.db") == []
|
|
sqlite3.connect(tmp_path / "empty.db").close()
|
|
assert _seen_gateway_accounts(tmp_path / "empty.db") == []
|
|
|
|
|
|
@pytest.mark.parametrize("user_id, kwargs, expected", [
|
|
("111", dict(pin=True, aliases={"111": "alice"}, prefix="tg_", peer_name="eri"), "eri (pinned)"),
|
|
("111", dict(pin=True, aliases={}, prefix="", peer_name=""), "111"),
|
|
("111", dict(pin=False, aliases={"111": "alice"}, prefix="tg_", peer_name="eri"), "alice"),
|
|
("111", dict(pin=False, aliases={}, prefix="tg_", peer_name="eri"), "tg_111 (prefixed)"),
|
|
("@you:matrix.org", dict(pin=False, aliases={}, prefix="", peer_name=""), "-you-matrix-org"),
|
|
("device-777", dict(pin=False, aliases={"uuid-abc": "eri"}, prefix="", peer_name="", user_id_alt="uuid-abc"), "eri"),
|
|
("111", dict(pin=False, aliases={"111": "alice", "uuid-abc": "bob"}, prefix="", peer_name="", user_id_alt="uuid-abc"),
|
|
"alice"),
|
|
], ids=["pin-wins", "pin-without-peer-name", "alias", "prefix", "raw-sanitized", "alt-id-alias", "primary-alias-over-alt"])
|
|
def test_preview_peer_resolution(user_id, kwargs, expected):
|
|
assert _preview_peer_resolution(user_id, **kwargs) == expected
|
|
|
|
|
|
@pytest.mark.parametrize("user_id, peer_name", [("a:b", "eri"), ("111", "tg_111")],
|
|
ids=["sanitizing-changed-the-id", "collides-with-peer-name"])
|
|
def test_prefixed_preview_matches_runtime_hash_suffix(user_id, peer_name):
|
|
"""The runtime appends a hash when sanitizing changed the id or it collides with an explicit peer."""
|
|
from plugins.memory.honcho.client import HonchoClientConfig
|
|
from plugins.memory.honcho.session import HonchoSessionManager
|
|
|
|
manager = HonchoSessionManager(
|
|
config=HonchoClientConfig(peer_name=peer_name, runtime_peer_prefix="tg_"), runtime_user_peer_name=user_id,
|
|
)
|
|
runtime = manager._resolve_user_peer_id("telegram:dm:1")
|
|
assert runtime.startswith(sanitize_peer_id(f"tg_{user_id}") + "-")
|
|
assert _preview_peer_resolution(user_id, pin=False, aliases={}, prefix="tg_", peer_name=peer_name) == f"{runtime} (prefixed)"
|
|
|
|
|
|
def _run_map(monkeypatch, tmp_path, *, answers, cfg, db_rows=(), ws_peers=None, workspaces=None):
|
|
"""Drive cmd_peers_map with scripted answers; returns what _write_config received."""
|
|
db = tmp_path / "state.db"
|
|
if db_rows:
|
|
_make_state_db(db, list(db_rows))
|
|
written = {}
|
|
answer_iter = iter(answers)
|
|
# One profile per host block: "hermes" is default, "hermes.<name>" is profile <name>.
|
|
profiles = [("default" if k == "hermes" else k.removeprefix("hermes."), k, v) for k, v in cfg["hosts"].items()]
|
|
# API seams are offline unless ws_peers is given, then a sentinel client stands in.
|
|
client = object() if ws_peers is not None else None
|
|
for name, value in {
|
|
"_read_config": lambda: cfg,
|
|
"_host_key": lambda: "hermes",
|
|
"_active_profile_name": lambda: "default",
|
|
"_state_db_path": lambda: db,
|
|
"_local_config_path": lambda: tmp_path / "honcho.json",
|
|
"_write_config": lambda c, path=None: written.update({"cfg": c}),
|
|
"_all_profile_host_configs": lambda: profiles,
|
|
"_peers_map_client": lambda workspace=None: (client, SimpleNamespace(workspace_id="hermes") if client else None),
|
|
"_api_workspace_peers": lambda c: list(ws_peers) if c is not None and ws_peers is not None else None,
|
|
"_api_workspaces": lambda c: list(workspaces) if workspaces is not None else None,
|
|
"_api_peer_detail": lambda c, pid: f"(card of {pid})",
|
|
"_prompt": lambda label, default=None, secret=False: next(answer_iter, default or ""),
|
|
}.items():
|
|
monkeypatch.setattr(honcho_cli, name, value)
|
|
honcho_cli.cmd_peers_map(SimpleNamespace())
|
|
return written
|
|
|
|
|
|
class TestCmdPeersMap:
|
|
@pytest.mark.parametrize("cfg, answers, kwargs", [
|
|
(_cfg(peerName="eri"), ["1", "eri", ""], dict(db_rows=[_row("111")])),
|
|
(_cfg(), ["111", "eri", ""], {}),
|
|
(_cfg(peerName="eri"), ["1", "p1", ""], dict(db_rows=[_row("111")], ws_peers=["eri", "hermes"])),
|
|
(_cfg(pinUserPeer=True, peerName="eri"), ["y", "111", "eri", ""], {}),
|
|
], ids=["account-number", "typed-runtime-id", "picked-from-peers-table", "pinned-but-accepted"])
|
|
def test_writes_alias_to_host_block(self, monkeypatch, tmp_path, cfg, answers, kwargs):
|
|
written = _run_map(monkeypatch, tmp_path, answers=answers, cfg=cfg, **kwargs)
|
|
assert written["cfg"]["hosts"]["hermes"]["userPeerAliases"] == {"111": "eri"}
|
|
|
|
@pytest.mark.parametrize("aliases, expected_host", [
|
|
({"111": "eri", "222": "tek"}, {"userPeerAliases": {"222": "tek"}}),
|
|
({"111": "eri"}, {"userPeerAliases": {}}),
|
|
], ids=["one-of-two", "last-alias-keeps-empty-map"])
|
|
def test_dash_clears_alias(self, monkeypatch, tmp_path, aliases, expected_host):
|
|
written = _run_map(monkeypatch, tmp_path, answers=["111", "-", ""], cfg=_cfg(userPeerAliases=aliases))
|
|
assert written["cfg"]["hosts"]["hermes"] == expected_host
|
|
|
|
def test_cleared_host_map_does_not_resurrect_root_aliases(self, monkeypatch, tmp_path, capsys):
|
|
"""A host block without the key inherits root, so clearing must leave an empty map behind."""
|
|
from plugins.memory.honcho.client import HonchoClientConfig
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
cfg = _cfg(root={"userPeerAliases": {"111": "root-person"}}, userPeerAliases={"111": "host-person"})
|
|
written = _run_map(monkeypatch, tmp_path, answers=["111", "-", ""], cfg=cfg)
|
|
assert written["cfg"]["hosts"]["hermes"]["userPeerAliases"] == {}
|
|
assert "root aliases no longer apply to [hermes]" in capsys.readouterr().out
|
|
|
|
path = tmp_path / "honcho.json"
|
|
path.write_text(json.dumps(written["cfg"]))
|
|
assert HonchoClientConfig.from_global_config(host="hermes", config_path=path).user_peer_aliases == {}
|
|
|
|
@pytest.mark.parametrize("cfg, answers, kwargs", [
|
|
(_cfg(), [""], {}),
|
|
(_cfg(userPeerAliases={"111": "eri"}), ["111", "eri", ""], {}),
|
|
(_cfg(pinUserPeer=True, peerName="eri"), ["n"], {}),
|
|
(_cfg(peerName="eri"), ["w", "2", "n", ""], dict(ws_peers=["eri"], workspaces=["hermes", "cosmania-dex"])),
|
|
], ids=["nothing-entered", "kept-current-value", "pinned-declined", "workspace-switch-declined"])
|
|
def test_nothing_changed_writes_nothing(self, monkeypatch, tmp_path, cfg, answers, kwargs):
|
|
assert _run_map(monkeypatch, tmp_path, answers=answers, cfg=cfg, **kwargs) == {}
|
|
|
|
|
|
|
|
|
|
class TestWorkspaceSwitch:
|
|
def test_browse_and_confirmed_switch_writes_workspace(self, monkeypatch, tmp_path):
|
|
written = _run_map(
|
|
monkeypatch, tmp_path, answers=["w", "2", "y", ""],
|
|
cfg=_cfg(peerName="eri"), ws_peers=["eri"], workspaces=["hermes", "cosmania-dex"],
|
|
)
|
|
assert written["cfg"]["hosts"]["hermes"]["workspace"] == "cosmania-dex"
|
|
|
|
def test_browse_client_is_built_for_the_browsed_workspace(self, monkeypatch):
|
|
"""get_honcho_client keys its cache on workspace_id, so a browse must not reuse the profile's client."""
|
|
import plugins.memory.honcho.client as client_mod
|
|
|
|
seen = []
|
|
base = client_mod.HonchoClientConfig(host="hermes", workspace_id="hermes", api_key="k")
|
|
monkeypatch.setattr(client_mod.HonchoClientConfig, "from_global_config",
|
|
classmethod(lambda cls, host=None, config_path=None: base))
|
|
monkeypatch.setattr(client_mod, "get_honcho_client", lambda cfg: seen.append(cfg) or object())
|
|
monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
|
|
|
|
_, own = honcho_cli._peers_map_client()
|
|
_, browsed = honcho_cli._peers_map_client(workspace="cosmania-dex")
|
|
|
|
assert own.workspace_id == "hermes"
|
|
assert browsed.workspace_id == "cosmania-dex"
|
|
assert [c.workspace_id for c in seen] == ["hermes", "cosmania-dex"]
|
|
|
|
|
|
class _Page:
|
|
"""The honcho SDK's SyncPage: iterating it walks every page, ``items`` is this page only."""
|
|
|
|
def __init__(self, pages, index=0):
|
|
self._pages, self._index = pages, index
|
|
self.items = [SimpleNamespace(id=p) for p in pages[index]]
|
|
|
|
def __iter__(self):
|
|
for page in self._pages[self._index:]:
|
|
yield from (SimpleNamespace(id=p) for p in page)
|
|
|
|
def has_next_page(self):
|
|
return self._index + 1 < len(self._pages)
|
|
|
|
def get_next_page(self):
|
|
return _Page(self._pages, self._index + 1)
|
|
|
|
|
|
def test_api_workspace_peers_reads_only_the_pages_the_cap_needs():
|
|
"""Iterating a SyncPage walks the whole workspace; a 300-peer workspace with a 200 cap must touch
|
|
four pages, list each peer once, and never fetch page five."""
|
|
pages = [[f"p{i}" for i in range(n, n + 50)] for n in range(0, 300, 50)]
|
|
fetched = []
|
|
|
|
class _CountingPage(_Page):
|
|
def __init__(self, pages, index=0):
|
|
super().__init__(pages, index)
|
|
fetched.append(index)
|
|
|
|
def get_next_page(self):
|
|
return _CountingPage(self._pages, self._index + 1)
|
|
|
|
client = SimpleNamespace(peers=lambda page, size: _CountingPage(pages, page - 1))
|
|
peers = honcho_cli._api_workspace_peers(client)
|
|
assert len(peers) == len(set(peers)) == honcho_cli._PEERS_MAP_FETCH_CAP
|
|
assert fetched == [0, 1, 2, 3]
|
|
|
|
|
|
class TestSaveScope:
|
|
@pytest.mark.parametrize("hosts, answers", [
|
|
({"hermes": {}}, ["222", "tek", ""]),
|
|
({"hermes": {}, "hermes.dreamer": {}}, ["222", "tek", "", "all"]),
|
|
], ids=["single-profile-no-prompt", "multi-profile-all"])
|
|
def test_root_sourced_aliases_write_back_to_root(self, monkeypatch, tmp_path, hosts, answers):
|
|
cfg = {"apiKey": "***", "userPeerAliases": {"111": "eri"}, "hosts": hosts}
|
|
written = _run_map(monkeypatch, tmp_path, answers=answers, cfg=cfg)
|
|
assert written["cfg"]["userPeerAliases"] == {"111": "eri", "222": "tek"}
|
|
assert "userPeerAliases" not in written["cfg"]["hosts"]["hermes"]
|
|
|
|
def test_scope_this_forks_host_block(self, monkeypatch, tmp_path):
|
|
cfg = {"apiKey": "***", "userPeerAliases": {"111": "eri"}, "hosts": {"hermes": {}, "hermes.dreamer": {}}}
|
|
written = _run_map(monkeypatch, tmp_path, answers=["222", "tek", "", "this"], cfg=cfg)
|
|
assert written["cfg"]["hosts"]["hermes"]["userPeerAliases"] == {"111": "eri", "222": "tek"}
|
|
assert written["cfg"]["userPeerAliases"] == {"111": "eri"} # root stays the other profiles' baseline
|
|
|
|
|
|
def test_classify_workspace_peers_labels_from_local_config(monkeypatch):
|
|
monkeypatch.setattr(honcho_cli, "_host_key", lambda: "hermes")
|
|
rows = [
|
|
("default", "hermes", {"peerName": "eri", "aiPeer": "hermetika"}),
|
|
("dreamer", "hermes.dreamer", {"aiPeer": "dreamer-ai"}),
|
|
]
|
|
cfg = {"peerName": "eri", "hosts": {"claude_code": {"aiPeer": "clawd"}}}
|
|
accounts = [{"platform": "telegram", "user_id": "7654321", "user_id_alt": ""}]
|
|
labels = honcho_cli._classify_workspace_peers(
|
|
["eri", "hermetika", "dreamer-ai", "clawd", "7654321", "tg_7654321", "friend", "user-default-root", "meow"],
|
|
cfg, accounts, {"999": "friend"}, "tg_", rows,
|
|
)
|
|
assert labels == {
|
|
"eri": "your peer (peerName)",
|
|
"hermetika": "AI peer · this profile",
|
|
"dreamer-ai": "AI peer · profile dreamer",
|
|
"clawd": "AI peer of app 'claude_code'",
|
|
"7654321": "runtime peer · telegram 7654321",
|
|
"tg_7654321": "runtime peer · telegram 7654321",
|
|
"friend": "alias target",
|
|
"user-default-root": "fallback peer (pre-identity traffic)",
|
|
"meow": "unrecognized",
|
|
}
|