test_langfuse_plugin.py
1 """Tests for the bundled observability/langfuse plugin.""" 2 from __future__ import annotations 3 4 import importlib 5 import sys 6 from pathlib import Path 7 8 import pytest 9 10 import yaml 11 12 13 REPO_ROOT = Path(__file__).resolve().parents[2] 14 PLUGIN_DIR = REPO_ROOT / "plugins" / "observability" / "langfuse" 15 16 17 # --------------------------------------------------------------------------- 18 # Manifest + layout 19 # --------------------------------------------------------------------------- 20 21 class TestManifest: 22 def test_plugin_directory_exists(self): 23 assert PLUGIN_DIR.is_dir() 24 assert (PLUGIN_DIR / "plugin.yaml").exists() 25 assert (PLUGIN_DIR / "__init__.py").exists() 26 27 def test_manifest_fields(self): 28 data = yaml.safe_load((PLUGIN_DIR / "plugin.yaml").read_text()) 29 assert data["name"] == "langfuse" 30 assert data["version"] 31 # All six hooks the plugin implements. 32 assert set(data["hooks"]) == { 33 "pre_api_request", "post_api_request", 34 "pre_llm_call", "post_llm_call", 35 "pre_tool_call", "post_tool_call", 36 } 37 # Required env vars are the user-facing HERMES_ prefixed keys. 38 assert "HERMES_LANGFUSE_PUBLIC_KEY" in data["requires_env"] 39 assert "HERMES_LANGFUSE_SECRET_KEY" in data["requires_env"] 40 41 42 # --------------------------------------------------------------------------- 43 # Plugin discovery: langfuse is opt-in (not loaded unless explicitly enabled). 44 # This guards against someone accidentally re-introducing a per-hook 45 # load_config() gate or making the plugin auto-load. 46 # --------------------------------------------------------------------------- 47 48 class TestDiscovery: 49 def test_plugin_is_discovered_as_standalone_opt_in(self, tmp_path, monkeypatch): 50 """Scanner should find the plugin but NOT load it by default.""" 51 from hermes_cli import plugins as plugins_mod 52 53 # Isolated HERMES_HOME so we don't read the developer's config.yaml. 54 home = tmp_path / ".hermes" 55 home.mkdir() 56 monkeypatch.setenv("HERMES_HOME", str(home)) 57 monkeypatch.setattr(Path, "home", lambda: tmp_path) 58 59 manager = plugins_mod.PluginManager() 60 manager.discover_and_load() 61 62 # observability/langfuse appears in the plugin registry … 63 loaded = manager._plugins.get("observability/langfuse") 64 assert loaded is not None, "plugin not discovered" 65 # … but is not loaded (opt-in default → no config.yaml means nothing enabled) 66 assert loaded.enabled is False 67 assert "not enabled" in (loaded.error or "").lower() 68 69 70 # --------------------------------------------------------------------------- 71 # Runtime gate: _get_langfuse() returns None and caches _INIT_FAILED when 72 # credentials are missing. Guards against regressing toward the rejected 73 # per-hook load_config() design. 74 # --------------------------------------------------------------------------- 75 76 class TestRuntimeGate: 77 def _fresh_plugin(self): 78 """Import the plugin module fresh (clears any cached client).""" 79 mod_name = "plugins.observability.langfuse" 80 sys.modules.pop(mod_name, None) 81 return importlib.import_module(mod_name) 82 83 def test_get_langfuse_returns_none_without_credentials(self, monkeypatch): 84 for k in ( 85 "HERMES_LANGFUSE_PUBLIC_KEY", "HERMES_LANGFUSE_SECRET_KEY", 86 "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", 87 ): 88 monkeypatch.delenv(k, raising=False) 89 90 langfuse_plugin = self._fresh_plugin() 91 assert langfuse_plugin._get_langfuse() is None 92 93 def test_get_langfuse_caches_failure_no_config_load(self, monkeypatch): 94 """A miss must be cached — no per-hook config.yaml reads, no env re-reads.""" 95 for k in ( 96 "HERMES_LANGFUSE_PUBLIC_KEY", "HERMES_LANGFUSE_SECRET_KEY", 97 "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", 98 ): 99 monkeypatch.delenv(k, raising=False) 100 101 langfuse_plugin = self._fresh_plugin() 102 103 # Prime the cache with one call. 104 assert langfuse_plugin._get_langfuse() is None 105 106 # Now block os.environ.get — a correctly-cached plugin must not 107 # touch env again. 108 import os 109 called = {"n": 0} 110 real_get = os.environ.get 111 112 def tracking_get(key, default=None): 113 if key.startswith(("HERMES_LANGFUSE_", "LANGFUSE_")): 114 called["n"] += 1 115 return real_get(key, default) 116 117 monkeypatch.setattr(os.environ, "get", tracking_get) 118 119 for _ in range(20): 120 assert langfuse_plugin._get_langfuse() is None 121 122 assert called["n"] == 0, ( 123 f"_get_langfuse() re-read env {called['n']} times after cache miss — " 124 "it should short-circuit via _INIT_FAILED" 125 ) 126 127 def test_get_langfuse_does_not_import_hermes_config(self, monkeypatch): 128 """The plugin must not re-read config.yaml per hook.""" 129 for k in ( 130 "HERMES_LANGFUSE_PUBLIC_KEY", "HERMES_LANGFUSE_SECRET_KEY", 131 "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", 132 ): 133 monkeypatch.delenv(k, raising=False) 134 135 # Drop any cached import of hermes_cli.config. 136 sys.modules.pop("hermes_cli.config", None) 137 138 langfuse_plugin = self._fresh_plugin() 139 for _ in range(20): 140 langfuse_plugin._get_langfuse() 141 142 assert "hermes_cli.config" not in sys.modules, ( 143 "langfuse plugin imported hermes_cli.config — regression toward " 144 "the rejected per-hook load_config() design" 145 ) 146 147 148 # --------------------------------------------------------------------------- 149 # Hooks are inert when the client is unavailable. 150 # --------------------------------------------------------------------------- 151 152 class TestHooksInert: 153 def test_hooks_noop_without_client(self, monkeypatch): 154 """All 6 hooks must return without raising when _get_langfuse() is None.""" 155 for k in ( 156 "HERMES_LANGFUSE_PUBLIC_KEY", "HERMES_LANGFUSE_SECRET_KEY", 157 "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", 158 ): 159 monkeypatch.delenv(k, raising=False) 160 161 sys.modules.pop("plugins.observability.langfuse", None) 162 import importlib 163 mod = importlib.import_module("plugins.observability.langfuse") 164 165 # Each hook should just return; no exceptions. 166 mod.on_pre_llm_call(task_id="t", session_id="s", messages=[{"role": "user", "content": "hi"}]) 167 mod.on_pre_llm_request(task_id="t", session_id="s", api_call_count=1, messages=[]) 168 mod.on_post_llm_call(task_id="t", session_id="s", api_call_count=1) 169 mod.on_pre_tool_call(tool_name="read_file", args={}, task_id="t", session_id="s") 170 mod.on_post_tool_call(tool_name="read_file", args={}, result="ok", task_id="t", session_id="s")