"""Real config, dispatch and local HTTP must preserve web billing boundaries.""" import json from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from threading import Thread import pytest import httpx @pytest.fixture def local_gateway(monkeypatch): from tools import managed_tool_gateway requests = [] class Handler(BaseHTTPRequestHandler): def do_POST(self): body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) requests.append((self.path, dict(self.headers), body)) if self.path == "/v2/search": status = 200 payload = {"success": True, "data": {"web": [{ "title": "Local fixture", "url": "https://example.test", "description": "fixture", }]}} else: status, payload = 503, {"error": "local test outage"} content = json.dumps(payload).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(content))) self.end_headers() self.wfile.write(content) def log_message(self, format, *args): pass server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = Thread(target=server.serve_forever, daemon=True) thread.start() base = f"http://127.0.0.1:{server.server_port}" monkeypatch.setenv("TOOL_GATEWAY_USER_TOKEN", "test-nous-token") monkeypatch.setenv("PERPLEXITY_GATEWAY_URL", base + "/perplexity") monkeypatch.setenv("FIRECRAWL_GATEWAY_URL", base) monkeypatch.setenv("PERPLEXITY_BASE_URL", base + "/direct") # Only entitlement is stubbed: route and credential resolution remain real. Fast search carries # its own (registered-identity) gate, stubbed here to match the entitled premise; the real tier # resolution — including the anonymous exclusion — lives in test_web_free_fast_search.py. monkeypatch.setattr(managed_tool_gateway, "managed_nous_tools_enabled", lambda **kw: True) monkeypatch.setattr(managed_tool_gateway, "fast_search_entitled", lambda **kw: True) # CI omits the firecrawl extra; stand in for the SDK, still over real HTTP. class FirecrawlSDK: def __init__(self, api_key, api_url): self.api_key, self.api_url = api_key, api_url def search(self, query, limit): response = httpx.post( f"{self.api_url}/v2/search", json={"query": query, "limit": limit}, headers={"Authorization": f"Bearer {self.api_key}"}, ) return response.json() monkeypatch.setattr("plugins.web.firecrawl.provider.Firecrawl", FirecrawlSDK) monkeypatch.setattr("tools.web_tools._firecrawl_client", None, raising=False) try: yield requests finally: server.shutdown() server.server_close() thread.join(timeout=2) @pytest.mark.parametrize( "selection,direct_key,expected_paths", [ ({"backend": "nous"}, False, ["/perplexity/search", "/v2/search"]), ({"backend": "nous"}, True, ["/direct/search"]), ({"backend": "nous", "search_backend": "perplexity"}, True, ["/direct/search"]), ({"search_backend": "perplexity"}, False, []), ({"backend": "nous", "search_backend": "perplexity"}, False, []), ({"backend": "perplexity"}, False, []), ({"backend": "firecrawl", "search_backend": "perplexity"}, False, []), ({}, False, ["/perplexity/search", "/v2/search"]), ({"extract_backend": "exa"}, False, ["/perplexity/search", "/v2/search"]), ], ) def test_only_managed_search_may_use_billed_fallback( monkeypatch, tmp_path, local_gateway, selection, direct_key, expected_paths, ): from hermes_cli.config import atomic_config_write from tools import web_tools from tests.tools.conftest import register_all_web_providers monkeypatch.setenv("HERMES_HOME", str(tmp_path)) atomic_config_write(tmp_path / "config.yaml", { "web": {**selection, "keyless_rescue": False, "cache_enabled": True}, }) if direct_key: monkeypatch.setenv("PERPLEXITY_API_KEY", "test-direct-key") else: monkeypatch.delenv("PERPLEXITY_API_KEY", raising=False) register_all_web_providers() expected = [ (path, "Bearer test-direct-key" if path == "/direct/search" else "Bearer test-nous-token") for path in expected_paths ] for attempt in (1, 2): result = json.loads(web_tools.web_search_tool("local fixture", limit=3)) # A fallback must not be cached: the identical query retries Perplexity. assert [(path, headers["Authorization"]) for path, headers, _ in local_gateway] == expected * attempt assert result["success"] == bool(expected_paths and expected_paths[-1] == "/v2/search") if result["success"]: assert result["data"]["fallback_from"] == "managed_primary" assert "local test outage" in result["data"]["backend_error"] assert web_tools._get_extract_backend() == selection.get("extract_backend", "firecrawl") for path, headers, body in local_gateway: if path != "/v2/search": managed = path == "/perplexity/search" assert body.get("search_type") == ("fast" if managed else None) assert body["search_context_size"] == "low" assert headers.get("X-Pplx-Integration") == (None if managed else "hermes-agent") def test_managed_search_without_identity_names_the_gateway(monkeypatch, tmp_path, local_gateway): from hermes_cli.config import atomic_config_write from tools import managed_tool_gateway, web_tools from tests.tools.conftest import register_all_web_providers monkeypatch.setenv("HERMES_HOME", str(tmp_path)) atomic_config_write(tmp_path / "config.yaml", {"web": {"backend": "nous", "keyless_rescue": False}}) monkeypatch.delenv("PERPLEXITY_API_KEY", raising=False) monkeypatch.delenv("TOOL_GATEWAY_USER_TOKEN", raising=False) monkeypatch.setattr(managed_tool_gateway, "managed_nous_tools_enabled", lambda **kw: False) register_all_web_providers() error = json.loads(web_tools.web_search_tool("local fixture", limit=3))["error"] assert "no usable Nous identity" in error and "hermes tools" in error assert "PERPLEXITY_API_KEY" not in error assert local_gateway == [] def test_unentitled_managed_search_gets_fast_search_but_no_billed_fallback(monkeypatch, tmp_path, local_gateway): from hermes_cli.config import atomic_config_write from tools import managed_tool_gateway, web_tools from tests.tools.conftest import register_all_web_providers monkeypatch.setenv("HERMES_HOME", str(tmp_path)) atomic_config_write(tmp_path / "config.yaml", {"web": {"backend": "nous", "keyless_rescue": False}}) monkeypatch.delenv("PERPLEXITY_API_KEY", raising=False) monkeypatch.setattr(managed_tool_gateway, "managed_nous_tools_enabled", lambda **kw: False) register_all_web_providers() # The fixture 503s the Perplexity route; an entitled caller would then be served by /v2/search. assert json.loads(web_tools.web_search_tool("local fixture", limit=3))["success"] is False assert [(path, body.get("search_type")) for path, _, body in local_gateway] == [("/perplexity/search", "fast")]