From 58aad80d56d33379f45d9ba34479ffec476e145e Mon Sep 17 00:00:00 2001 From: Sameer Kadam Date: Mon, 24 Aug 2026 21:08:46 +0530 Subject: [PATCH] fix: guard Agno and OpenClaw integration requests against SSRF (#1212) * fix: guard integration HTTP requests against SSRF * fix(openclaw): complete fallback validation and base URL handling Address the remaining review findings in the OpenClaw integration. - Strengthen fallback base_url validation to require a non-empty string, valid HTTP(S) scheme, netloc, and hostname. - Strip leading and trailing whitespace from base_url before storing it. - Replace the flaky endpoint-construction test that made a real network connection with mocked session assertions. - Add coverage for _get and _post endpoint construction and timeout forwarding. - Add regression tests for whitespace-padded base URLs and the fallback validation path. These changes complete the Qodo review fixes and harden OpenClaw URL handling without changing the intended localhost/private deployment behavior. --- integrations/agno/knowledge_graph.py | 23 +- integrations/openclaw/mcp_tool.py | 36 ++- .../integrations/agno/test_load_urls_ssrf.py | 278 +++++++++++++++++ tests/integrations/openclaw/__init__.py | 1 + .../openclaw/test_mcp_tool_ssrf.py | 290 ++++++++++++++++++ 5 files changed, 614 insertions(+), 14 deletions(-) create mode 100644 tests/integrations/agno/test_load_urls_ssrf.py create mode 100644 tests/integrations/openclaw/__init__.py create mode 100644 tests/integrations/openclaw/test_mcp_tool_ssrf.py diff --git a/integrations/agno/knowledge_graph.py b/integrations/agno/knowledge_graph.py index 21b6b280..78004bc3 100644 --- a/integrations/agno/knowledge_graph.py +++ b/integrations/agno/knowledge_graph.py @@ -277,25 +277,22 @@ class AgnoKnowledgeGraph(_KnowledgeBase): # type: ignore[misc] def load_urls(self, urls: List[str]) -> None: """Fetch each URL and ingest the response body. - Only ``http`` and ``https`` schemes are permitted to prevent SSRF. + Uses the shared SSRF guard so that ``http`` and ``https`` are the only + permitted schemes, private/loopback/link-local/cloud-metadata addresses + are blocked by default, DNS resolution is validated, and every redirect + hop is re-checked before being followed. """ - import urllib.request - from urllib.parse import urlparse + from semantica.ingest.ssrf import request_with_ssrf_guard + from semantica.utils.exceptions import ValidationError for url in urls: - parsed = urlparse(url) - if parsed.scheme not in ("http", "https"): - logger.warning( - "Skipping URL with disallowed scheme '%s': %s", - parsed.scheme, - url, - ) - continue try: - with urllib.request.urlopen(url, timeout=10) as resp: # noqa: S310 - text = resp.read().decode("utf-8", errors="replace") + response = request_with_ssrf_guard("GET", url, timeout=10) + text = response.text self._ingest_text(text, source=url) logger.info("Loaded URL: %s", url) + except ValidationError as exc: + logger.warning("Skipping URL (SSRF check failed) %s: %s", url, exc) except Exception as exc: logger.warning("Failed to fetch %s: %s", url, exc) diff --git a/integrations/openclaw/mcp_tool.py b/integrations/openclaw/mcp_tool.py index dff6b6db..626f8b95 100644 --- a/integrations/openclaw/mcp_tool.py +++ b/integrations/openclaw/mcp_tool.py @@ -116,7 +116,41 @@ class OpenClawKGTool: ) def __init__(self, base_url: str = "http://localhost:8000", timeout: int = 30) -> None: - self.base_url = base_url.rstrip("/") + # Validate base_url at construction time so callers get an immediate, + # actionable error rather than a cryptic failure on the first request. + # allow_private_ips=True because the documented default (localhost:8000) + # is intentionally a local Semantica server; the scheme check and + # URL-structure check still apply unconditionally. + try: + from semantica.ingest.ssrf import validate_url_for_request + validate_url_for_request(base_url, allow_private_ips=True) + except ImportError: + # semantica.ingest not installed in minimal openclaw-only environments; + # mirror the structural checks that validate_url_for_request performs + # unconditionally (before allow_private_ips is consulted), so the + # guarantee in the comment above — "scheme check and URL-structure check + # still apply unconditionally" — holds in this path too. + from urllib.parse import urlparse as _urlparse + if not isinstance(base_url, str) or not base_url.strip(): + raise ValueError("OpenClawKGTool base_url must be a non-empty string.") + _parsed = _urlparse(base_url.strip()) + _scheme = (_parsed.scheme or "").lower() + if _scheme not in ("http", "https"): + raise ValueError( + f"OpenClawKGTool base_url scheme '{_parsed.scheme}' is not permitted. " + "Only http and https are allowed." + ) + if not _parsed.netloc: + raise ValueError( + f"Invalid OpenClawKGTool base_url '{base_url}': " + "URL must include a netloc (domain or host)." + ) + if not _parsed.hostname: + raise ValueError( + f"Invalid OpenClawKGTool base_url '{base_url}': " + "URL must include a hostname." + ) + self.base_url = base_url.strip().rstrip("/") self.timeout = timeout self._session: Any = None diff --git a/tests/integrations/agno/test_load_urls_ssrf.py b/tests/integrations/agno/test_load_urls_ssrf.py new file mode 100644 index 00000000..28ad1cbe --- /dev/null +++ b/tests/integrations/agno/test_load_urls_ssrf.py @@ -0,0 +1,278 @@ +"""SSRF regression tests for AgnoKnowledgeGraph.load_urls(). + +Prior to the fix, load_urls() used urllib.request.urlopen with only a +scheme check — private/loopback/link-local/metadata IPs were not blocked +and redirects were followed without re-validation. + +These tests exercise the real SSRF guard (no mock of request_with_ssrf_guard +itself) by patching at the socket.getaddrinfo level, confirming that +blocked addresses never reach the network layer. +""" + +from __future__ import annotations + +import socket +from unittest.mock import MagicMock, patch + +import pytest + +# conftest.py installs the full agno stub before this file is collected. +from integrations.agno.knowledge_graph import AgnoKnowledgeGraph + +from semantica.utils.exceptions import ValidationError + + +# --------------------------------------------------------------------------- +# Minimal fakes so AgnoKnowledgeGraph.__init__ succeeds without real imports. +# --------------------------------------------------------------------------- +class _FakeNER: + def extract_entities(self, text): + return [] + + +class _FakeRelExtractor: + def extract_relations(self, text, entities=None): + return [] + + +class _FakeGraphBuilder: + def build(self, sources): + pass + + +class _FakeContextGraph: + def find_nodes(self, label=None): + return [] + + def get_neighbors(self, node_id=None, hops=1): + return [] + + +def _make_kg() -> AgnoKnowledgeGraph: + return AgnoKnowledgeGraph( + graph_builder=_FakeGraphBuilder(), + ner_extractor=_FakeNER(), + relation_extractor=_FakeRelExtractor(), + context_graph=_FakeContextGraph(), + ) + + +def _public_getaddrinfo(host, *args, **kwargs): + """Stub that makes every hostname resolve to a public IP.""" + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))] + + +# --------------------------------------------------------------------------- +# Tests: blocked addresses must never be fetched +# --------------------------------------------------------------------------- + +class TestLoadUrlsBlockedAddresses: + """load_urls() must silently skip (warn) any URL that fails the SSRF guard.""" + + @pytest.mark.parametrize("url", [ + "http://127.0.0.1/secret", + "http://127.0.0.1:9200/", # common internal service port + "http://0.0.0.0/", + "http://169.254.169.254/latest/meta-data/", + "http://169.254.169.254/computeMetadata/v1/", + "http://10.0.0.1/internal", + "http://10.255.255.255/", + "http://172.16.0.1/", + "http://172.31.255.255/", + "http://192.168.0.1/admin", + "http://192.168.100.200/", + "http://[::1]/ipv6-loopback", + "http://[fc00::1]/ipv6-ula", + "http://[fe80::1]/ipv6-link-local", + ]) + def test_blocked_ip_never_reaches_network(self, url): + """Blocked addresses must raise ValidationError inside the guard, + which load_urls() catches and logs — _ingest_text must NOT be called.""" + kg = _make_kg() + with patch.object(kg, "_ingest_text") as mock_ingest: + kg.load_urls([url]) + mock_ingest.assert_not_called() + + def test_localhost_hostname_blocked(self): + kg = _make_kg() + with patch.object(kg, "_ingest_text") as mock_ingest: + kg.load_urls(["http://localhost/admin"]) + mock_ingest.assert_not_called() + + def test_localhost_subdomain_blocked(self): + kg = _make_kg() + with patch.object(kg, "_ingest_text") as mock_ingest: + kg.load_urls(["http://foo.localhost/"]) + mock_ingest.assert_not_called() + + def test_hostname_resolving_to_private_ip_blocked(self): + """A hostname that resolves to a private IP must be blocked even though + the URL string itself looks like a normal hostname.""" + def _internal_getaddrinfo(host, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", 0))] + + kg = _make_kg() + with patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_internal_getaddrinfo): + with patch.object(kg, "_ingest_text") as mock_ingest: + kg.load_urls(["http://internal.corp/secret"]) + mock_ingest.assert_not_called() + + def test_hostname_resolving_to_metadata_ip_blocked(self): + def _meta_getaddrinfo(host, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("169.254.169.254", 0))] + + kg = _make_kg() + with patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_meta_getaddrinfo): + with patch.object(kg, "_ingest_text") as mock_ingest: + kg.load_urls(["http://metadata.internal/v1/token"]) + mock_ingest.assert_not_called() + + +class TestLoadUrlsNonHttpSchemes: + """Non-HTTP(S) schemes must be rejected.""" + + @pytest.mark.parametrize("url", [ + "file:///etc/passwd", + "file://localhost/etc/shadow", + "ftp://example.com/file.txt", + "gopher://example.com/1", + "dict://example.com/", + "sftp://example.com/data", + ]) + def test_non_http_scheme_blocked(self, url): + kg = _make_kg() + with patch.object(kg, "_ingest_text") as mock_ingest: + kg.load_urls([url]) + mock_ingest.assert_not_called() + + +class TestLoadUrlsRedirects: + """Redirects to private/blocked addresses must be rejected.""" + + def test_redirect_to_loopback_blocked(self): + """A public first hop that redirects to loopback must be blocked.""" + redirect = MagicMock() + redirect.status_code = 302 + redirect.headers = {"Location": "http://127.0.0.1/secret"} + redirect.close = MagicMock() + + kg = _make_kg() + with patch( + "semantica.ingest.ssrf.socket.getaddrinfo", + return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))], + ): + # Patch requests.Session so the first hop returns our redirect mock. + # The guard sees the 302, then validates the Location — 127.0.0.1 is + # blocked without a second network call. + with patch("semantica.ingest.ssrf.requests.Session") as MockSession: + mock_session = MockSession.return_value + mock_session.adapters = {} + mock_session.headers = {} + mock_session.auth = None + mock_session.trust_env = True + mock_session.request.return_value = redirect + + with patch.object(kg, "_ingest_text") as mock_ingest: + kg.load_urls(["https://example.com/start"]) + mock_ingest.assert_not_called() + + def test_redirect_to_metadata_ip_blocked(self): + """Redirect to cloud metadata endpoint must be blocked.""" + redirect = MagicMock() + redirect.status_code = 301 + redirect.headers = {"Location": "http://169.254.169.254/latest/meta-data/"} + redirect.close = MagicMock() + + kg = _make_kg() + with patch( + "semantica.ingest.ssrf.socket.getaddrinfo", + return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))], + ): + with patch("semantica.ingest.ssrf.requests.Session") as MockSession: + mock_session = MockSession.return_value + mock_session.adapters = {} + mock_session.headers = {} + mock_session.auth = None + mock_session.trust_env = True + mock_session.request.return_value = redirect + + with patch.object(kg, "_ingest_text") as mock_ingest: + kg.load_urls(["https://example.com/redirect-me"]) + mock_ingest.assert_not_called() + + +class TestLoadUrlsValidUrls: + """Valid public URLs must succeed and call _ingest_text.""" + + def test_valid_public_url_ingested(self): + """A URL resolving to a public IP must be fetched and ingested.""" + ok_response = MagicMock() + ok_response.status_code = 200 + ok_response.headers = {} + ok_response.text = "This is the document content." + + kg = _make_kg() + with patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo): + with patch("semantica.ingest.ssrf.requests.Session") as MockSession: + mock_session = MockSession.return_value + mock_session.adapters = {} + mock_session.headers = {} + mock_session.auth = None + mock_session.trust_env = True + mock_session.request.return_value = ok_response + + with patch.object(kg, "_ingest_text") as mock_ingest: + kg.load_urls(["https://example.com/doc.txt"]) + + mock_ingest.assert_called_once_with( + "This is the document content.", source="https://example.com/doc.txt" + ) + + def test_multiple_urls_each_independently_validated(self): + """Each URL in the list is independently validated; one blocked URL + must not prevent valid subsequent URLs from being ingested.""" + ok_response = MagicMock() + ok_response.status_code = 200 + ok_response.headers = {} + ok_response.text = "Valid content." + + def _selective_getaddrinfo(host, *a, **kw): + if host == "internal.corp": + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", 0))] + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))] + + kg = _make_kg() + with patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_selective_getaddrinfo): + with patch("semantica.ingest.ssrf.requests.Session") as MockSession: + mock_session = MockSession.return_value + mock_session.adapters = {} + mock_session.headers = {} + mock_session.auth = None + mock_session.trust_env = True + mock_session.request.return_value = ok_response + + with patch.object(kg, "_ingest_text") as mock_ingest: + kg.load_urls([ + "http://internal.corp/secret", # blocked + "https://example.com/public.txt", # allowed + ]) + + # Only the valid URL triggers ingestion + mock_ingest.assert_called_once_with("Valid content.", source="https://example.com/public.txt") + + def test_failed_fetch_does_not_raise(self): + """A network failure on a valid URL must log a warning, not raise.""" + import requests as _requests + + kg = _make_kg() + with patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo): + with patch("semantica.ingest.ssrf.requests.Session") as MockSession: + mock_session = MockSession.return_value + mock_session.adapters = {} + mock_session.headers = {} + mock_session.auth = None + mock_session.trust_env = True + mock_session.request.side_effect = _requests.exceptions.ConnectionError("refused") + + # Must not raise; failure is logged and skipped + kg.load_urls(["https://example.com/unreachable"]) diff --git a/tests/integrations/openclaw/__init__.py b/tests/integrations/openclaw/__init__.py new file mode 100644 index 00000000..6def8d9a --- /dev/null +++ b/tests/integrations/openclaw/__init__.py @@ -0,0 +1 @@ +# tests/integrations/openclaw package diff --git a/tests/integrations/openclaw/test_mcp_tool_ssrf.py b/tests/integrations/openclaw/test_mcp_tool_ssrf.py new file mode 100644 index 00000000..4c75c772 --- /dev/null +++ b/tests/integrations/openclaw/test_mcp_tool_ssrf.py @@ -0,0 +1,290 @@ +"""SSRF hardening tests for OpenClawKGTool. + +OpenClawKGTool is designed to speak to a locally-running Semantica REST server +(default: http://localhost:8000). The fix validates base_url at construction +time so that obviously wrong schemes (file://, ftp://, gopher://, etc.) and +malformed URLs are rejected immediately, while localhost and other private +addresses remain valid because allow_private_ips=True is the correct posture +for this tool's intended use case. + +These are construction-time tests; per-request SSRF guarding is not the +contract of this tool (its threat model is operator-configured base_url, not +untrusted per-call URLs). +""" + +from __future__ import annotations + +import pytest + +from integrations.openclaw.mcp_tool import OpenClawKGTool +from semantica.utils.exceptions import ValidationError + + +class TestOpenClawKGToolBaseUrlValidation: + """base_url is validated at __init__ time.""" + + # ------------------------------------------------------------------ + # Valid base_urls — all must construct without raising + # ------------------------------------------------------------------ + + @pytest.mark.parametrize("url", [ + "http://localhost:8000", + "http://localhost", + "http://127.0.0.1:8000", + "http://127.0.0.1", + "https://localhost:8443", + "http://0.0.0.0:8000", + "http://192.168.1.10:8000", # LAN Semantica server + "http://10.0.0.5:8000", # corporate intranet deployment + "https://semantica.internal/api", + "https://semantica.example.com", + ]) + def test_valid_base_url_accepted(self, url): + """All reasonable operator-configured base_urls must be accepted.""" + tool = OpenClawKGTool(base_url=url) + assert tool.base_url == url.rstrip("/") + + # ------------------------------------------------------------------ + # Invalid schemes — must raise at construction + # ------------------------------------------------------------------ + + @pytest.mark.parametrize("url", [ + "file:///etc/passwd", + "file://localhost/etc/shadow", + "ftp://example.com/", + "gopher://example.com/1", + "dict://example.com/", + "sftp://example.com/", + "ldap://example.com/", + "javascript:alert(1)", + ]) + def test_invalid_scheme_rejected(self, url): + """Non-HTTP(S) schemes must be rejected at construction time.""" + with pytest.raises((ValidationError, ValueError)): + OpenClawKGTool(base_url=url) + + # ------------------------------------------------------------------ + # Malformed URLs + # ------------------------------------------------------------------ + + def test_empty_string_rejected(self): + with pytest.raises((ValidationError, ValueError)): + OpenClawKGTool(base_url="") + + def test_no_scheme_rejected(self): + """A bare hostname without a scheme must be rejected.""" + with pytest.raises((ValidationError, ValueError)): + OpenClawKGTool(base_url="localhost:8000") + + def test_whitespace_only_rejected(self): + with pytest.raises((ValidationError, ValueError)): + OpenClawKGTool(base_url=" ") + + # ------------------------------------------------------------------ + # Default is the documented localhost value + # ------------------------------------------------------------------ + + def test_default_base_url_is_localhost(self): + """The default must remain http://localhost:8000 for backward compat.""" + tool = OpenClawKGTool() + assert tool.base_url == "http://localhost:8000" + + def test_trailing_slash_stripped(self): + """base_url trailing slash must be stripped so paths concatenate cleanly.""" + tool = OpenClawKGTool(base_url="http://localhost:8000/") + assert tool.base_url == "http://localhost:8000" + + def test_multiple_trailing_slashes_stripped(self): + tool = OpenClawKGTool(base_url="http://localhost:8000///") + assert tool.base_url == "http://localhost:8000" + + def test_leading_and_trailing_whitespace_stripped(self): + """Whitespace around a valid URL must be stripped before storage so + _post/_get don't build requests with space-padded URLs like + ' http://localhost:8000 /extract'.""" + tool = OpenClawKGTool(base_url=" http://localhost:8000 ") + assert tool.base_url == "http://localhost:8000" + + def test_whitespace_plus_trailing_slash_both_stripped(self): + tool = OpenClawKGTool(base_url=" http://localhost:8000/ ") + assert tool.base_url == "http://localhost:8000" + + +class TestOpenClawKGToolFallbackValidation: + """When semantica.ingest.ssrf is unavailable (ImportError path), the fallback + must perform the same structural checks as validate_url_for_request: + non-empty string, http/https scheme, netloc present, hostname present. + + The fallback is exercised by temporarily hiding semantica.ingest.ssrf + from sys.modules so the import inside __init__ raises ImportError. + """ + + @staticmethod + def _hide_ssrf(monkeypatch): + """Return a context in which semantica.ingest.ssrf appears unimportable.""" + import sys + monkeypatch.setitem(sys.modules, "semantica.ingest.ssrf", None) + + # ------------------------------------------------------------------ + # Valid URLs must still be accepted in the fallback path + # ------------------------------------------------------------------ + + @pytest.mark.parametrize("url", [ + "http://localhost:8000", + "http://127.0.0.1:8000", + "https://semantica.example.com", + ]) + def test_fallback_valid_url_accepted(self, url, monkeypatch): + self._hide_ssrf(monkeypatch) + tool = OpenClawKGTool(base_url=url) + assert tool.base_url == url.rstrip("/") + + # ------------------------------------------------------------------ + # Malformed URLs that the fallback previously let through + # ------------------------------------------------------------------ + + @pytest.mark.parametrize("url", [ + "http://", # scheme only, no netloc or hostname + "https://", # same + "http:///path", # empty hostname (netloc is present but hostname is None) + ]) + def test_fallback_no_netloc_rejected(self, url, monkeypatch): + """URLs with a valid scheme but missing netloc/hostname must be rejected + in the fallback path, matching validate_url_for_request's behaviour.""" + self._hide_ssrf(monkeypatch) + with pytest.raises((ValidationError, ValueError)): + OpenClawKGTool(base_url=url) + + def test_fallback_empty_string_rejected(self, monkeypatch): + self._hide_ssrf(monkeypatch) + with pytest.raises((ValidationError, ValueError)): + OpenClawKGTool(base_url="") + + def test_fallback_whitespace_only_rejected(self, monkeypatch): + self._hide_ssrf(monkeypatch) + with pytest.raises((ValidationError, ValueError)): + OpenClawKGTool(base_url=" ") + + def test_fallback_invalid_scheme_rejected(self, monkeypatch): + self._hide_ssrf(monkeypatch) + with pytest.raises((ValidationError, ValueError)): + OpenClawKGTool(base_url="file:///etc/passwd") + + def test_fallback_no_scheme_rejected(self, monkeypatch): + self._hide_ssrf(monkeypatch) + with pytest.raises((ValidationError, ValueError)): + OpenClawKGTool(base_url="localhost:8000") + + def test_fallback_whitespace_padded_valid_url_stored_clean(self, monkeypatch): + """Whitespace around a valid URL must be stripped before storage in the + fallback path too — same guarantee as the normal path.""" + self._hide_ssrf(monkeypatch) + tool = OpenClawKGTool(base_url=" http://localhost:8000 ") + assert tool.base_url == "http://localhost:8000" + + +class TestOpenClawKGToolEndpointConstruction: + """Verify that per-method URLs are assembled from base_url + hardcoded paths. + + The endpoint strings are always literals defined in the class body — + they are not caller-supplied — so these tests confirm the URL assembly + logic is correct rather than testing SSRF guards on the endpoints. + + All HTTP calls are mocked so no real network connection is made. + """ + + def _mock_session(self, status: int = 200, body: bytes = b"{}") -> "MagicMock": + """Return a mock session whose post/get return a minimal JSON response.""" + from unittest.mock import MagicMock + mock_resp = MagicMock() + mock_resp.status_code = status + mock_resp.raise_for_status = MagicMock() + mock_resp.json.return_value = {} + session = MagicMock() + session.post.return_value = mock_resp + session.get.return_value = mock_resp + return session + + def test_post_url_constructed_from_base_url(self): + """_post must call session.post with the exact URL base_url+endpoint, + the supplied payload as json=, and the tool timeout. No real connection.""" + from unittest.mock import patch + + tool = OpenClawKGTool(base_url="http://localhost:8000") + mock_session = self._mock_session() + + with patch.object(tool, "_get_session", return_value=mock_session): + tool._post("/extract", {"text": "hello"}) + + mock_session.post.assert_called_once_with( + "http://localhost:8000/extract", + json={"text": "hello"}, + timeout=30, + ) + + def test_post_url_with_custom_base_url(self): + """base_url is reflected correctly in the outbound URL for _post.""" + from unittest.mock import patch + + tool = OpenClawKGTool(base_url="http://192.168.1.10:9000") + mock_session = self._mock_session() + + with patch.object(tool, "_get_session", return_value=mock_session): + tool._post("/decisions", {"decision": "deploy"}) + + mock_session.post.assert_called_once_with( + "http://192.168.1.10:9000/decisions", + json={"decision": "deploy"}, + timeout=30, + ) + + def test_get_url_constructed_from_base_url(self): + """_get must call session.get with the exact URL base_url+endpoint, + params={} when none are supplied, and the tool timeout.""" + from unittest.mock import patch + + tool = OpenClawKGTool(base_url="http://localhost:8000") + mock_session = self._mock_session() + + with patch.object(tool, "_get_session", return_value=mock_session): + tool._get("/analytics") + + mock_session.get.assert_called_once_with( + "http://localhost:8000/analytics", + params={}, + timeout=30, + ) + + def test_get_url_with_params(self): + """_get must forward supplied params to session.get.""" + from unittest.mock import patch + + tool = OpenClawKGTool(base_url="http://localhost:8000") + mock_session = self._mock_session() + + with patch.object(tool, "_get_session", return_value=mock_session): + tool._get("/decisions/search", {"q": "deploy", "limit": 5}) + + mock_session.get.assert_called_once_with( + "http://localhost:8000/decisions/search", + params={"q": "deploy", "limit": 5}, + timeout=30, + ) + + def test_custom_timeout_forwarded(self): + """A non-default timeout must reach session.post and session.get.""" + from unittest.mock import patch + + tool = OpenClawKGTool(base_url="http://localhost:8000", timeout=60) + mock_session = self._mock_session() + + with patch.object(tool, "_get_session", return_value=mock_session): + tool._post("/extract", {"text": "x"}) + tool._get("/analytics") + + assert mock_session.post.call_args.kwargs["timeout"] == 60 + assert mock_session.get.call_args.kwargs["timeout"] == 60 + + def test_repr_includes_base_url(self): + tool = OpenClawKGTool(base_url="http://localhost:9000") + assert "http://localhost:9000" in repr(tool)