104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
"""Tests for deterministic Telegram participant-name provenance in Hermes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PATCHER = ROOT / "hermes-agent" / "patches" / "apply_telegram_sender_name_policy.py"
|
|
|
|
|
|
def load_patcher():
|
|
spec = importlib.util.spec_from_file_location("hermes_telegram_sender_name_policy", PATCHER)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def upstream_fixture() -> str:
|
|
return '''class Adapter:
|
|
def build_source(self, **values):
|
|
return values
|
|
|
|
def build_event_source(self, user):
|
|
return self.build_source(
|
|
user_id=str(user.id) if user else None,
|
|
user_name=user.full_name if user else None,
|
|
)
|
|
'''
|
|
|
|
|
|
def patched_adapter_class():
|
|
module = load_patcher()
|
|
patched, changed = module.patch_source(upstream_fixture())
|
|
assert changed is True
|
|
namespace: dict[str, object] = {}
|
|
exec(patched, namespace)
|
|
return namespace["Adapter"], module, patched
|
|
|
|
|
|
def test_current_visible_handle_overrides_stale_profile_name() -> None:
|
|
adapter_class, _module, _patched = patched_adapter_class()
|
|
user = SimpleNamespace(id=9070919, username="m3taversal", full_name="STALE-PERSONAL-NAME")
|
|
|
|
source = adapter_class().build_event_source(user)
|
|
|
|
assert "m3taversal" not in PATCHER.read_text(encoding="utf-8")
|
|
assert source == {"user_id": "9070919", "user_name": "m3taversal"}
|
|
assert "STALE-PERSONAL-NAME" not in source.values()
|
|
|
|
|
|
def test_each_participant_keeps_own_handle_and_stable_user_id() -> None:
|
|
adapter_class, _module, _patched = patched_adapter_class()
|
|
target = SimpleNamespace(id=9070919, username="m3taversal", full_name="STALE-PERSONAL-NAME")
|
|
other = SimpleNamespace(id=8181818, username="other_operator", full_name="Other Operator")
|
|
|
|
first_target = adapter_class().build_event_source(target)
|
|
other_source = adapter_class().build_event_source(other)
|
|
restarted_target = adapter_class().build_event_source(target)
|
|
|
|
assert first_target == restarted_target == {"user_id": "9070919", "user_name": "m3taversal"}
|
|
assert other_source == {"user_id": "8181818", "user_name": "other_operator"}
|
|
assert first_target["user_id"] != other_source["user_id"]
|
|
assert first_target["user_name"] != other_source["user_name"]
|
|
|
|
|
|
def test_full_name_is_only_the_no_username_fallback() -> None:
|
|
adapter_class, _module, _patched = patched_adapter_class()
|
|
user = SimpleNamespace(id=7171717, username=None, full_name="Visible Display Name")
|
|
|
|
assert adapter_class().build_event_source(user) == {
|
|
"user_id": "7171717",
|
|
"user_name": "Visible Display Name",
|
|
}
|
|
|
|
|
|
def test_patch_is_idempotent_and_rejects_unknown_upstream_shape() -> None:
|
|
_adapter_class, module, patched = patched_adapter_class()
|
|
|
|
assert module.patch_source(patched) == (patched, False)
|
|
with pytest.raises(ValueError, match="sender construction changed"):
|
|
module.patch_source("class Adapter:\n pass\n")
|
|
|
|
|
|
def test_install_check_reports_patch_required_then_installed(tmp_path: Path) -> None:
|
|
module = load_patcher()
|
|
target = tmp_path / "telegram.py"
|
|
target.write_text(upstream_fixture(), encoding="utf-8")
|
|
|
|
before, before_code = module.install(target, check=True)
|
|
assert before_code == 1
|
|
assert before["status"] == "patch_required"
|
|
|
|
applied, applied_code = module.install(target, check=False)
|
|
assert applied_code == 0
|
|
assert applied["status"] == "installed_now"
|
|
|
|
after, after_code = module.install(target, check=True)
|
|
assert after_code == 0
|
|
assert after["status"] == "installed"
|