From fe185e5b8dd988df8f952c2a2106a8fd148e436b Mon Sep 17 00:00:00 2001 From: Alex Date: Sat, 28 Mar 2026 21:51:47 +0000 Subject: [PATCH] chore: api and tool tests --- tests/agents/test_api_body_serializer.py | 202 +++++++ tests/agents/test_api_tool.py | 282 ++++++++++ tests/agents/test_brave_tool.py | 132 +++++ tests/agents/test_cryptoprice_tool.py | 85 +++ tests/agents/test_duckduckgo_tool.py | 145 +++++ tests/agents/test_mcp_tool.py | 519 ++++++++++++++++++ tests/agents/test_ntfy_tool.py | 146 +++++ tests/agents/test_postgres_tool.py | 146 +++++ tests/agents/test_read_webpage_tool.py | 86 +++ tests/agents/test_spec_parser.py | 335 +++++++++++ tests/agents/test_telegram_tool.py | 79 +++ tests/api/answer/routes/test_answer.py | 235 ++++++++ tests/api/answer/routes/test_stream.py | 195 +++++++ .../answer/services/compression/__init__.py | 0 .../compression/test_message_builder.py | 303 ++++++++++ .../services/compression/test_orchestrator.py | 447 +++++++++++++++ .../services/compression/test_service.py | 423 ++++++++++++++ .../answer/services/compression/test_types.py | 131 +++++ 18 files changed, 3891 insertions(+) create mode 100644 tests/agents/test_api_body_serializer.py create mode 100644 tests/agents/test_api_tool.py create mode 100644 tests/agents/test_brave_tool.py create mode 100644 tests/agents/test_cryptoprice_tool.py create mode 100644 tests/agents/test_duckduckgo_tool.py create mode 100644 tests/agents/test_mcp_tool.py create mode 100644 tests/agents/test_ntfy_tool.py create mode 100644 tests/agents/test_postgres_tool.py create mode 100644 tests/agents/test_read_webpage_tool.py create mode 100644 tests/agents/test_spec_parser.py create mode 100644 tests/agents/test_telegram_tool.py create mode 100644 tests/api/answer/routes/test_answer.py create mode 100644 tests/api/answer/routes/test_stream.py create mode 100644 tests/api/answer/services/compression/__init__.py create mode 100644 tests/api/answer/services/compression/test_message_builder.py create mode 100644 tests/api/answer/services/compression/test_orchestrator.py create mode 100644 tests/api/answer/services/compression/test_service.py create mode 100644 tests/api/answer/services/compression/test_types.py diff --git a/tests/agents/test_api_body_serializer.py b/tests/agents/test_api_body_serializer.py new file mode 100644 index 00000000..b10d0429 --- /dev/null +++ b/tests/agents/test_api_body_serializer.py @@ -0,0 +1,202 @@ +"""Tests for application/agents/tools/api_body_serializer.py""" + +import json + +import pytest + +from application.agents.tools.api_body_serializer import ( + ContentType, + RequestBodySerializer, +) + + +@pytest.mark.unit +class TestContentTypeEnum: + def test_json_value(self): + assert ContentType.JSON == "application/json" + + def test_form_urlencoded_value(self): + assert ContentType.FORM_URLENCODED == "application/x-www-form-urlencoded" + + def test_multipart_value(self): + assert ContentType.MULTIPART_FORM_DATA == "multipart/form-data" + + def test_text_plain_value(self): + assert ContentType.TEXT_PLAIN == "text/plain" + + def test_xml_value(self): + assert ContentType.XML == "application/xml" + + def test_octet_stream_value(self): + assert ContentType.OCTET_STREAM == "application/octet-stream" + + +@pytest.mark.unit +class TestSerializeJson: + def test_basic_json(self): + body, headers = RequestBodySerializer.serialize( + {"key": "value"}, ContentType.JSON + ) + assert json.loads(body) == {"key": "value"} + assert headers["Content-Type"] == "application/json" + + def test_nested_json(self): + data = {"user": {"name": "Alice", "age": 30}} + body, headers = RequestBodySerializer.serialize(data, ContentType.JSON) + assert json.loads(body) == data + + def test_empty_body_returns_none(self): + body, headers = RequestBodySerializer.serialize({}, ContentType.JSON) + assert body is None + assert headers == {} + + def test_none_body(self): + body, headers = RequestBodySerializer.serialize(None, ContentType.JSON) + assert body is None + + def test_unknown_content_type_falls_back_to_json(self): + body, headers = RequestBodySerializer.serialize( + {"k": "v"}, "application/vnd.custom+json" + ) + assert json.loads(body) == {"k": "v"} + + def test_content_type_with_charset_suffix(self): + body, headers = RequestBodySerializer.serialize( + {"k": "v"}, "application/json; charset=utf-8" + ) + assert json.loads(body) == {"k": "v"} + + +@pytest.mark.unit +class TestSerializeFormUrlencoded: + def test_basic_form(self): + body, headers = RequestBodySerializer.serialize( + {"name": "Alice", "age": "30"}, ContentType.FORM_URLENCODED + ) + assert "name=Alice" in body + assert "age=30" in body + assert headers["Content-Type"] == "application/x-www-form-urlencoded" + + def test_none_values_skipped(self): + body, headers = RequestBodySerializer.serialize( + {"name": "Alice", "skip": None}, ContentType.FORM_URLENCODED + ) + assert "name=Alice" in body + assert "skip" not in body + + def test_list_explode_true(self): + body, headers = RequestBodySerializer.serialize( + {"tags": ["a", "b"]}, + ContentType.FORM_URLENCODED, + encoding_rules={"tags": {"style": "form", "explode": True}}, + ) + assert "tags=a" in body + assert "tags=b" in body + + def test_list_explode_false(self): + body, headers = RequestBodySerializer.serialize( + {"tags": ["a", "b"]}, + ContentType.FORM_URLENCODED, + encoding_rules={"tags": {"style": "form", "explode": False}}, + ) + # Value is percent-encoded by _serialize_form_value then urlencoded again + assert "tags=" in body + assert "a" in body and "b" in body + + def test_dict_value_json_content_type(self): + body, headers = RequestBodySerializer.serialize( + {"metadata": {"key": "val"}}, + ContentType.FORM_URLENCODED, + encoding_rules={"metadata": {"contentType": "application/json"}}, + ) + assert "metadata" in body + + +@pytest.mark.unit +class TestSerializeTextPlain: + def test_single_value(self): + body, headers = RequestBodySerializer.serialize( + {"message": "hello"}, ContentType.TEXT_PLAIN + ) + assert body == "hello" + assert headers["Content-Type"] == "text/plain" + + def test_multiple_values(self): + body, headers = RequestBodySerializer.serialize( + {"name": "Alice", "age": 30}, ContentType.TEXT_PLAIN + ) + assert "name: Alice" in body + assert "age: 30" in body + + +@pytest.mark.unit +class TestSerializeXml: + def test_basic_xml(self): + body, headers = RequestBodySerializer.serialize( + {"name": "Alice"}, ContentType.XML + ) + assert 'Alice" in body + assert headers["Content-Type"] == "application/xml" + + def test_nested_xml(self): + body, headers = RequestBodySerializer.serialize( + {"user": {"name": "Alice"}}, ContentType.XML + ) + assert "" in body + assert "Alice" in body + + def test_xml_escapes_special_chars(self): + body, headers = RequestBodySerializer.serialize( + {"data": ""}, ContentType.XML + ) + assert "<script>" in body + + +@pytest.mark.unit +class TestSerializeOctetStream: + def test_dict_body(self): + body, headers = RequestBodySerializer.serialize( + {"key": "val"}, ContentType.OCTET_STREAM + ) + assert isinstance(body, bytes) + assert headers["Content-Type"] == "application/octet-stream" + + +@pytest.mark.unit +class TestSerializeMultipartFormData: + def test_basic_multipart(self): + body, headers = RequestBodySerializer.serialize( + {"field": "value"}, ContentType.MULTIPART_FORM_DATA + ) + assert isinstance(body, bytes) + assert "multipart/form-data" in headers["Content-Type"] + assert "boundary=" in headers["Content-Type"] + + def test_none_values_skipped(self): + body, headers = RequestBodySerializer.serialize( + {"field": "value", "empty": None}, ContentType.MULTIPART_FORM_DATA + ) + body_str = body.decode("utf-8", errors="replace") + assert "field" in body_str + assert "empty" not in body_str + + +@pytest.mark.unit +class TestHelpers: + def test_percent_encode(self): + assert RequestBodySerializer._percent_encode("hello world") == "hello%20world" + assert RequestBodySerializer._percent_encode("a/b") == "a%2Fb" + assert RequestBodySerializer._percent_encode("safe", safe_chars="/") == "safe" + + def test_escape_xml(self): + assert "&" in RequestBodySerializer._escape_xml("&") + assert "<" in RequestBodySerializer._escape_xml("<") + assert ">" in RequestBodySerializer._escape_xml(">") + assert """ in RequestBodySerializer._escape_xml('"') + assert "'" in RequestBodySerializer._escape_xml("'") + + def test_dict_to_xml_list(self): + xml = RequestBodySerializer._dict_to_xml({"items": [1, 2, 3]}) + assert "1" in xml + assert "2" in xml diff --git a/tests/agents/test_api_tool.py b/tests/agents/test_api_tool.py new file mode 100644 index 00000000..8f59c225 --- /dev/null +++ b/tests/agents/test_api_tool.py @@ -0,0 +1,282 @@ +"""Tests for application/agents/tools/api_tool.py""" + +import json +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from application.agents.tools.api_tool import APITool + + +@pytest.fixture +def tool(): + return APITool( + config={ + "url": "https://api.example.com/data", + "method": "GET", + "headers": {"Accept": "application/json"}, + "query_params": {}, + } + ) + + +@pytest.fixture +def post_tool(): + return APITool( + config={ + "url": "https://api.example.com/items", + "method": "POST", + "headers": {}, + "query_params": {}, + } + ) + + +@pytest.mark.unit +class TestAPIToolInit: + def test_default_values(self): + tool = APITool(config={}) + assert tool.url == "" + assert tool.method == "GET" + assert tool.headers == {} + assert tool.query_params == {} + + +@pytest.mark.unit +class TestMakeApiCall: + @patch("application.agents.tools.api_tool.validate_url") + @patch("application.agents.tools.api_tool.requests.get") + def test_successful_get(self, mock_get, mock_validate, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.headers = {"Content-Type": "application/json"} + mock_resp.json.return_value = {"result": "ok"} + mock_resp.content = b'{"result":"ok"}' + mock_get.return_value = mock_resp + + result = tool.execute_action("any_action") + + assert result["status_code"] == 200 + assert result["data"] == {"result": "ok"} + assert result["message"] == "API call successful." + + @patch("application.agents.tools.api_tool.validate_url") + @patch("application.agents.tools.api_tool.requests.post") + def test_successful_post(self, mock_post, mock_validate, post_tool): + mock_resp = MagicMock() + mock_resp.status_code = 201 + mock_resp.headers = {"Content-Type": "application/json"} + mock_resp.json.return_value = {"id": 1} + mock_resp.content = b'{"id":1}' + mock_post.return_value = mock_resp + + result = post_tool.execute_action("create", name="test") + + assert result["status_code"] == 201 + + @patch("application.agents.tools.api_tool.validate_url") + def test_ssrf_blocked(self, mock_validate, tool): + from application.core.url_validation import SSRFError + + mock_validate.side_effect = SSRFError("blocked") + + result = tool.execute_action("any") + + assert result["status_code"] is None + assert "URL validation error" in result["message"] + + @patch("application.agents.tools.api_tool.validate_url") + @patch("application.agents.tools.api_tool.requests.get") + def test_timeout_error(self, mock_get, mock_validate, tool): + mock_get.side_effect = requests.exceptions.Timeout() + + result = tool.execute_action("any") + + assert result["status_code"] is None + assert "timeout" in result["message"].lower() + + @patch("application.agents.tools.api_tool.validate_url") + @patch("application.agents.tools.api_tool.requests.get") + def test_connection_error(self, mock_get, mock_validate, tool): + mock_get.side_effect = requests.exceptions.ConnectionError("refused") + + result = tool.execute_action("any") + + assert result["status_code"] is None + assert "Connection error" in result["message"] + + @patch("application.agents.tools.api_tool.validate_url") + @patch("application.agents.tools.api_tool.requests.get") + def test_http_error(self, mock_get, mock_validate, tool): + mock_resp = MagicMock() + mock_resp.status_code = 404 + mock_resp.text = "Not Found" + mock_resp.json.side_effect = json.JSONDecodeError("", "", 0) + mock_resp.raise_for_status.side_effect = requests.exceptions.HTTPError( + response=mock_resp + ) + mock_get.return_value = mock_resp + + result = tool.execute_action("any") + + assert result["status_code"] == 404 + assert "HTTP Error" in result["message"] + + @patch("application.agents.tools.api_tool.validate_url") + def test_unsupported_method(self, mock_validate): + tool = APITool( + config={"url": "https://example.com", "method": "CUSTOM"} + ) + result = tool.execute_action("any") + assert result["status_code"] is None + assert "Unsupported" in result["message"] + + @patch("application.agents.tools.api_tool.validate_url") + @patch("application.agents.tools.api_tool.requests.put") + def test_put_method(self, mock_put, mock_validate): + tool = APITool(config={"url": "https://example.com/item/1", "method": "PUT"}) + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.headers = {"Content-Type": "application/json"} + mock_resp.json.return_value = {} + mock_resp.content = b'{}' + mock_put.return_value = mock_resp + + result = tool.execute_action("update", name="new") + assert result["status_code"] == 200 + + @patch("application.agents.tools.api_tool.validate_url") + @patch("application.agents.tools.api_tool.requests.delete") + def test_delete_method(self, mock_delete, mock_validate): + tool = APITool(config={"url": "https://example.com/item/1", "method": "DELETE"}) + mock_resp = MagicMock() + mock_resp.status_code = 204 + mock_resp.headers = {"Content-Type": "text/plain"} + mock_resp.content = b'' + mock_delete.return_value = mock_resp + + result = tool.execute_action("delete") + assert result["status_code"] == 204 + + @patch("application.agents.tools.api_tool.validate_url") + @patch("application.agents.tools.api_tool.requests.patch") + def test_patch_method(self, mock_patch, mock_validate): + tool = APITool(config={"url": "https://example.com/item/1", "method": "PATCH"}) + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.headers = {"Content-Type": "application/json"} + mock_resp.json.return_value = {"patched": True} + mock_resp.content = b'{"patched":true}' + mock_patch.return_value = mock_resp + + result = tool.execute_action("patch", field="val") + assert result["status_code"] == 200 + + @patch("application.agents.tools.api_tool.validate_url") + @patch("application.agents.tools.api_tool.requests.head") + def test_head_method(self, mock_head, mock_validate): + tool = APITool(config={"url": "https://example.com", "method": "HEAD"}) + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.headers = {"Content-Type": "text/html"} + mock_resp.content = b'' + mock_head.return_value = mock_resp + + result = tool.execute_action("check") + assert result["status_code"] == 200 + + @patch("application.agents.tools.api_tool.validate_url") + @patch("application.agents.tools.api_tool.requests.options") + def test_options_method(self, mock_options, mock_validate): + tool = APITool(config={"url": "https://example.com", "method": "OPTIONS"}) + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.headers = {"Content-Type": "text/plain"} + mock_resp.content = b'' + mock_options.return_value = mock_resp + + result = tool.execute_action("options") + assert result["status_code"] == 200 + + +@pytest.mark.unit +class TestPathParamSubstitution: + @patch("application.agents.tools.api_tool.validate_url") + @patch("application.agents.tools.api_tool.requests.get") + def test_path_params_substituted(self, mock_get, mock_validate): + tool = APITool( + config={ + "url": "https://api.example.com/users/{user_id}/posts/{post_id}", + "method": "GET", + "query_params": {"user_id": "42", "post_id": "7"}, + } + ) + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.headers = {"Content-Type": "application/json"} + mock_resp.json.return_value = [] + mock_resp.content = b'[]' + mock_get.return_value = mock_resp + + tool.execute_action("get") + + called_url = mock_get.call_args[0][0] + assert "/users/42/posts/7" in called_url + assert "{user_id}" not in called_url + + +@pytest.mark.unit +class TestParseResponse: + def test_json_response(self, tool): + mock_resp = MagicMock() + mock_resp.headers = {"Content-Type": "application/json"} + mock_resp.json.return_value = {"key": "val"} + mock_resp.content = b'{"key":"val"}' + + result = tool._parse_response(mock_resp) + assert result == {"key": "val"} + + def test_text_response(self, tool): + mock_resp = MagicMock() + mock_resp.headers = {"Content-Type": "text/plain"} + mock_resp.text = "plain text" + mock_resp.content = b"plain text" + + result = tool._parse_response(mock_resp) + assert result == "plain text" + + def test_xml_response(self, tool): + mock_resp = MagicMock() + mock_resp.headers = {"Content-Type": "application/xml"} + mock_resp.text = "1" + mock_resp.content = b"1" + + result = tool._parse_response(mock_resp) + assert "" in result + + def test_empty_content(self, tool): + mock_resp = MagicMock() + mock_resp.headers = {"Content-Type": "application/json"} + mock_resp.content = b"" + + result = tool._parse_response(mock_resp) + assert result is None + + def test_html_response(self, tool): + mock_resp = MagicMock() + mock_resp.headers = {"Content-Type": "text/html"} + mock_resp.text = "Hi" + mock_resp.content = b"Hi" + + result = tool._parse_response(mock_resp) + assert "" in result + + +@pytest.mark.unit +class TestAPIToolMetadata: + def test_actions_metadata_empty(self, tool): + assert tool.get_actions_metadata() == [] + + def test_config_requirements_empty(self, tool): + assert tool.get_config_requirements() == {} diff --git a/tests/agents/test_brave_tool.py b/tests/agents/test_brave_tool.py new file mode 100644 index 00000000..0ff3c706 --- /dev/null +++ b/tests/agents/test_brave_tool.py @@ -0,0 +1,132 @@ +"""Tests for application/agents/tools/brave.py""" + +from unittest.mock import MagicMock, patch + +import pytest + +from application.agents.tools.brave import BraveSearchTool + + +@pytest.fixture +def tool(): + return BraveSearchTool(config={"token": "test_api_key"}) + + +@pytest.mark.unit +class TestBraveExecuteAction: + def test_unknown_action_raises(self, tool): + with pytest.raises(ValueError, match="Unknown action"): + tool.execute_action("invalid") + + @patch("application.agents.tools.brave.requests.get") + def test_web_search_success(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"web": {"results": [{"title": "Result"}]}} + mock_get.return_value = mock_resp + + result = tool.execute_action("brave_web_search", query="python") + + assert result["status_code"] == 200 + assert "results" in result + assert "successfully" in result["message"] + + call_kwargs = mock_get.call_args + assert call_kwargs[1]["headers"]["X-Subscription-Token"] == "test_api_key" + + @patch("application.agents.tools.brave.requests.get") + def test_web_search_failure(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 429 + mock_get.return_value = mock_resp + + result = tool.execute_action("brave_web_search", query="test") + + assert result["status_code"] == 429 + assert "failed" in result["message"].lower() + + @patch("application.agents.tools.brave.requests.get") + def test_image_search_success(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"results": [{"url": "https://img.com/1.jpg"}]} + mock_get.return_value = mock_resp + + result = tool.execute_action("brave_image_search", query="cats") + + assert result["status_code"] == 200 + assert "results" in result + + @patch("application.agents.tools.brave.requests.get") + def test_image_search_failure(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 500 + mock_get.return_value = mock_resp + + result = tool.execute_action("brave_image_search", query="cats") + + assert result["status_code"] == 500 + + @patch("application.agents.tools.brave.requests.get") + def test_count_capped_at_20(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {} + mock_get.return_value = mock_resp + + tool.execute_action("brave_web_search", query="test", count=100) + + params = mock_get.call_args[1]["params"] + assert params["count"] == 20 + + @patch("application.agents.tools.brave.requests.get") + def test_image_count_capped_at_100(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {} + mock_get.return_value = mock_resp + + tool.execute_action("brave_image_search", query="test", count=500) + + params = mock_get.call_args[1]["params"] + assert params["count"] == 100 + + @patch("application.agents.tools.brave.requests.get") + def test_freshness_param(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {} + mock_get.return_value = mock_resp + + tool.execute_action("brave_web_search", query="news", freshness="pd") + + params = mock_get.call_args[1]["params"] + assert params["freshness"] == "pd" + + @patch("application.agents.tools.brave.requests.get") + def test_offset_capped(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {} + mock_get.return_value = mock_resp + + tool.execute_action("brave_web_search", query="test", offset=100) + + params = mock_get.call_args[1]["params"] + assert params["offset"] == 9 + + +@pytest.mark.unit +class TestBraveMetadata: + def test_actions_metadata(self, tool): + meta = tool.get_actions_metadata() + assert len(meta) == 2 + names = {a["name"] for a in meta} + assert "brave_web_search" in names + assert "brave_image_search" in names + + def test_config_requirements(self, tool): + reqs = tool.get_config_requirements() + assert "token" in reqs + assert reqs["token"]["secret"] is True + assert reqs["token"]["required"] is True diff --git a/tests/agents/test_cryptoprice_tool.py b/tests/agents/test_cryptoprice_tool.py new file mode 100644 index 00000000..00f25b1c --- /dev/null +++ b/tests/agents/test_cryptoprice_tool.py @@ -0,0 +1,85 @@ +"""Tests for application/agents/tools/cryptoprice.py""" + +from unittest.mock import MagicMock, patch + +import pytest + +from application.agents.tools.cryptoprice import CryptoPriceTool + + +@pytest.fixture +def tool(): + return CryptoPriceTool(config={}) + + +@pytest.mark.unit +class TestCryptoPriceExecuteAction: + def test_unknown_action_raises(self, tool): + with pytest.raises(ValueError, match="Unknown action"): + tool.execute_action("invalid_action") + + @patch("application.agents.tools.cryptoprice.requests.get") + def test_successful_price_fetch(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"USD": 65000} + mock_get.return_value = mock_resp + + result = tool.execute_action("cryptoprice_get", symbol="BTC", currency="USD") + + assert result["status_code"] == 200 + assert result["price"] == 65000 + assert "successfully" in result["message"] + + @patch("application.agents.tools.cryptoprice.requests.get") + def test_currency_not_found(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"EUR": 60000} + mock_get.return_value = mock_resp + + result = tool.execute_action("cryptoprice_get", symbol="BTC", currency="USD") + + assert result["status_code"] == 200 + assert "Couldn't find" in result["message"] + assert "price" not in result + + @patch("application.agents.tools.cryptoprice.requests.get") + def test_api_failure(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 500 + mock_get.return_value = mock_resp + + result = tool.execute_action("cryptoprice_get", symbol="BTC", currency="USD") + + assert result["status_code"] == 500 + assert "Failed" in result["message"] + + @patch("application.agents.tools.cryptoprice.requests.get") + def test_symbol_case_insensitive(self, mock_get, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"USD": 100} + mock_get.return_value = mock_resp + + tool.execute_action("cryptoprice_get", symbol="btc", currency="usd") + + called_url = mock_get.call_args[0][0] + assert "fsym=BTC" in called_url + assert "tsyms=USD" in called_url + + +@pytest.mark.unit +class TestCryptoPriceMetadata: + def test_actions_metadata(self, tool): + meta = tool.get_actions_metadata() + assert len(meta) == 1 + assert meta[0]["name"] == "cryptoprice_get" + params = meta[0]["parameters"] + assert "symbol" in params["properties"] + assert "currency" in params["properties"] + assert "symbol" in params["required"] + assert "currency" in params["required"] + + def test_config_requirements(self, tool): + assert tool.get_config_requirements() == {} diff --git a/tests/agents/test_duckduckgo_tool.py b/tests/agents/test_duckduckgo_tool.py new file mode 100644 index 00000000..3dc8727d --- /dev/null +++ b/tests/agents/test_duckduckgo_tool.py @@ -0,0 +1,145 @@ +"""Tests for application/agents/tools/duckduckgo.py""" + +from unittest.mock import MagicMock, patch + +import pytest + +from application.agents.tools.duckduckgo import DuckDuckGoSearchTool + + +@pytest.fixture +def tool(): + return DuckDuckGoSearchTool(config={}) + + +@pytest.mark.unit +class TestDuckDuckGoExecuteAction: + def test_unknown_action_raises(self, tool): + with pytest.raises(ValueError, match="Unknown action"): + tool.execute_action("invalid") + + @patch.object(DuckDuckGoSearchTool, "_get_ddgs_client") + def test_web_search_success(self, mock_client_factory, tool): + mock_client = MagicMock() + mock_client.text.return_value = [ + {"title": "Result 1", "href": "https://example.com", "body": "snippet"} + ] + mock_client_factory.return_value = mock_client + + result = tool.execute_action("ddg_web_search", query="python") + + assert result["status_code"] == 200 + assert len(result["results"]) == 1 + assert "successfully" in result["message"] + + @patch.object(DuckDuckGoSearchTool, "_get_ddgs_client") + def test_image_search_success(self, mock_client_factory, tool): + mock_client = MagicMock() + mock_client.images.return_value = [{"image": "https://img.com/1.jpg"}] + mock_client_factory.return_value = mock_client + + result = tool.execute_action("ddg_image_search", query="cats") + + assert result["status_code"] == 200 + assert len(result["results"]) == 1 + + @patch.object(DuckDuckGoSearchTool, "_get_ddgs_client") + def test_news_search_success(self, mock_client_factory, tool): + mock_client = MagicMock() + mock_client.news.return_value = [{"title": "News"}] + mock_client_factory.return_value = mock_client + + result = tool.execute_action("ddg_news_search", query="tech") + + assert result["status_code"] == 200 + assert len(result["results"]) == 1 + + @patch.object(DuckDuckGoSearchTool, "_get_ddgs_client") + def test_search_error_returns_500(self, mock_client_factory, tool): + mock_client = MagicMock() + mock_client.text.side_effect = Exception("Network error") + mock_client_factory.return_value = mock_client + + result = tool.execute_action("ddg_web_search", query="test") + + assert result["status_code"] == 500 + assert "failed" in result["message"].lower() + assert result["results"] == [] + + @patch.object(DuckDuckGoSearchTool, "_get_ddgs_client") + def test_max_results_capped_at_20(self, mock_client_factory, tool): + mock_client = MagicMock() + mock_client.text.return_value = [] + mock_client_factory.return_value = mock_client + + tool.execute_action("ddg_web_search", query="test", max_results=100) + + call_kwargs = mock_client.text.call_args[1] + assert call_kwargs["max_results"] == 20 + + @patch.object(DuckDuckGoSearchTool, "_get_ddgs_client") + def test_image_max_results_capped_at_50(self, mock_client_factory, tool): + mock_client = MagicMock() + mock_client.images.return_value = [] + mock_client_factory.return_value = mock_client + + tool.execute_action("ddg_image_search", query="test", max_results=200) + + call_kwargs = mock_client.images.call_args[1] + assert call_kwargs["max_results"] == 50 + + @patch("application.agents.tools.duckduckgo.time.sleep") + @patch.object(DuckDuckGoSearchTool, "_get_ddgs_client") + def test_rate_limit_retries(self, mock_client_factory, mock_sleep, tool): + mock_client = MagicMock() + mock_client.text.side_effect = [ + Exception("RateLimit exceeded"), + [{"title": "Result"}], + ] + mock_client_factory.return_value = mock_client + + result = tool.execute_action("ddg_web_search", query="test") + + assert result["status_code"] == 200 + assert len(result["results"]) == 1 + mock_sleep.assert_called_once() + + @patch.object(DuckDuckGoSearchTool, "_get_ddgs_client") + def test_empty_results(self, mock_client_factory, tool): + mock_client = MagicMock() + mock_client.text.return_value = [] + mock_client_factory.return_value = mock_client + + result = tool.execute_action("ddg_web_search", query="obscure query") + + assert result["status_code"] == 200 + assert result["results"] == [] + + @patch.object(DuckDuckGoSearchTool, "_get_ddgs_client") + def test_none_results(self, mock_client_factory, tool): + mock_client = MagicMock() + mock_client.text.return_value = None + mock_client_factory.return_value = mock_client + + result = tool.execute_action("ddg_web_search", query="test") + + assert result["status_code"] == 200 + assert result["results"] == [] + + +@pytest.mark.unit +class TestDuckDuckGoMetadata: + def test_actions_metadata(self, tool): + meta = tool.get_actions_metadata() + assert len(meta) == 3 + names = {a["name"] for a in meta} + assert "ddg_web_search" in names + assert "ddg_image_search" in names + assert "ddg_news_search" in names + + def test_config_requirements(self, tool): + assert tool.get_config_requirements() == {} + + def test_custom_timeout(self): + tool = DuckDuckGoSearchTool(config={"timeout": 30}) + assert tool.timeout == 30 diff --git a/tests/agents/test_mcp_tool.py b/tests/agents/test_mcp_tool.py new file mode 100644 index 00000000..eb17c757 --- /dev/null +++ b/tests/agents/test_mcp_tool.py @@ -0,0 +1,519 @@ +"""Tests for application/agents/tools/mcp_tool.py""" + +from unittest.mock import MagicMock, patch + +import pytest + + +# mcp_tool has a circular import at module level (mcp_tool -> tasks -> user -> mcp.py -> mcp_tool). +# We must patch the dependencies BEFORE the module is first imported. +@pytest.fixture(autouse=True) +def _patch_mcp_globals(monkeypatch): + """Patch module-level MongoDB and cache to avoid real connections.""" + import sys + + # If the module is already loaded, just patch attributes directly + if "application.agents.tools.mcp_tool" in sys.modules: + mcp_mod = sys.modules["application.agents.tools.mcp_tool"] + else: + # Break the circular import by pre-populating the tasks import + # with a mock before mcp_tool tries to import it + mock_tasks = MagicMock() + monkeypatch.setitem(sys.modules, "application.api.user.tasks", mock_tasks) + import application.agents.tools.mcp_tool as mcp_mod + + mock_mongo = MagicMock() + mock_db = MagicMock() + mock_db.__getitem__ = MagicMock(return_value=MagicMock()) + monkeypatch.setattr(mcp_mod, "mongo", mock_mongo) + monkeypatch.setattr(mcp_mod, "db", mock_db) + monkeypatch.setattr(mcp_mod, "_mcp_clients_cache", {}) + + +@pytest.fixture +def mcp_config(): + return { + "server_url": "https://mcp.example.com/api", + "transport_type": "http", + "auth_type": "none", + "timeout": 10, + } + + +@pytest.fixture +def bearer_config(): + return { + "server_url": "https://mcp.example.com/api", + "transport_type": "http", + "auth_type": "bearer", + "auth_credentials": {"bearer_token": "tok_123"}, + "timeout": 10, + } + + +@pytest.mark.unit +class TestMCPToolInit: + def test_basic_init(self, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + with patch.object(MCPTool, "_setup_client"): + tool = MCPTool(mcp_config) + + assert tool.server_url == "https://mcp.example.com/api" + assert tool.transport_type == "http" + assert tool.auth_type == "none" + assert tool.timeout == 10 + + def test_bearer_auth_credentials(self, bearer_config): + from application.agents.tools.mcp_tool import MCPTool + + with patch.object(MCPTool, "_setup_client"): + tool = MCPTool(bearer_config) + assert tool.auth_credentials["bearer_token"] == "tok_123" + + def test_no_server_url_skips_setup(self): + from application.agents.tools.mcp_tool import MCPTool + + with patch.object(MCPTool, "_setup_client") as mock_setup: + MCPTool({"server_url": "", "auth_type": "none"}) + mock_setup.assert_not_called() + + def test_oauth_skips_setup(self): + from application.agents.tools.mcp_tool import MCPTool + + with patch.object(MCPTool, "_setup_client") as mock_setup: + MCPTool( + { + "server_url": "https://mcp.example.com", + "auth_type": "oauth", + } + ) + mock_setup.assert_not_called() + + +@pytest.mark.unit +class TestGenerateCacheKey: + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_none_auth(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + assert "none" in tool._cache_key + assert "mcp.example.com" in tool._cache_key + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_bearer_auth(self, mock_setup, bearer_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(bearer_config) + assert "bearer:" in tool._cache_key + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_api_key_auth(self, mock_setup): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool( + { + "server_url": "https://mcp.example.com", + "auth_type": "api_key", + "auth_credentials": {"api_key": "sk-test12345678"}, + } + ) + assert "apikey:" in tool._cache_key + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_basic_auth(self, mock_setup): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool( + { + "server_url": "https://mcp.example.com", + "auth_type": "basic", + "auth_credentials": {"username": "user1", "password": "pass"}, + } + ) + assert "basic:user1" in tool._cache_key + + +@pytest.mark.unit +class TestCreateTransport: + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_http_transport(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + transport = tool._create_transport() + assert transport is not None + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_sse_transport(self, mock_setup): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool( + { + "server_url": "https://mcp.example.com/sse", + "transport_type": "sse", + "auth_type": "none", + } + ) + transport = tool._create_transport() + assert transport is not None + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_auto_detects_sse(self, mock_setup): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool( + { + "server_url": "https://mcp.example.com/sse", + "transport_type": "auto", + "auth_type": "none", + } + ) + transport = tool._create_transport() + assert transport is not None + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_stdio_transport_disabled(self, mock_setup): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool( + { + "server_url": "https://mcp.example.com", + "transport_type": "stdio", + "auth_type": "none", + } + ) + with pytest.raises(ValueError, match="STDIO transport is disabled"): + tool._create_transport() + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_api_key_header_injected(self, mock_setup): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool( + { + "server_url": "https://mcp.example.com", + "transport_type": "http", + "auth_type": "api_key", + "auth_credentials": { + "api_key": "sk-test", + "api_key_header": "X-Custom-Key", + }, + } + ) + # _create_transport will be called; verify it doesn't raise + transport = tool._create_transport() + assert transport is not None + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_basic_auth_header_injected(self, mock_setup): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool( + { + "server_url": "https://mcp.example.com", + "transport_type": "http", + "auth_type": "basic", + "auth_credentials": {"username": "user", "password": "pass"}, + } + ) + transport = tool._create_transport() + assert transport is not None + + +@pytest.mark.unit +class TestFormatTools: + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_format_list_of_dicts(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + result = tool._format_tools([{"name": "tool1", "description": "desc"}]) + assert len(result) == 1 + assert result[0]["name"] == "tool1" + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_format_tools_with_name_attribute(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + mock_tool = MagicMock() + mock_tool.name = "my_tool" + mock_tool.description = "A tool" + mock_tool.inputSchema = {"type": "object", "properties": {}} + + result = tool._format_tools([mock_tool]) + assert len(result) == 1 + assert result[0]["name"] == "my_tool" + assert result[0]["inputSchema"] == {"type": "object", "properties": {}} + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_format_tools_response_object(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + resp = MagicMock() + resp.tools = [{"name": "t1", "description": "d1"}] + + result = tool._format_tools(resp) + assert len(result) == 1 + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_format_empty(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + assert tool._format_tools([]) == [] + assert tool._format_tools("unexpected") == [] + + +@pytest.mark.unit +class TestFormatResult: + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_format_result_with_content(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + mock_result = MagicMock() + text_item = MagicMock() + text_item.text = "Hello" + del text_item.data + mock_result.content = [text_item] + mock_result.isError = False + + result = tool._format_result(mock_result) + assert result["content"][0]["type"] == "text" + assert result["content"][0]["text"] == "Hello" + assert result["isError"] is False + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_format_raw_result(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + raw = {"key": "value"} + assert tool._format_result(raw) == raw + + +@pytest.mark.unit +class TestExecuteAction: + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_no_server_raises(self, mock_setup): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool({"server_url": "", "auth_type": "none"}) + with pytest.raises(Exception, match="No MCP server configured"): + tool.execute_action("test_action") + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + @patch("application.agents.tools.mcp_tool.MCPTool._run_async_operation") + def test_successful_execute(self, mock_run, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + tool._client = MagicMock() + mock_run.return_value = {"key": "value"} + + result = tool.execute_action("test_action", param1="val1") + + mock_run.assert_called_once_with("call_tool", "test_action", param1="val1") + assert result == {"key": "value"} + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + @patch("application.agents.tools.mcp_tool.MCPTool._run_async_operation") + def test_empty_kwargs_cleaned(self, mock_run, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + tool._client = MagicMock() + mock_run.return_value = {} + + tool.execute_action("test", param1="", param2=None, param3="real") + + call_kwargs = mock_run.call_args[1] + assert "param1" not in call_kwargs + assert "param2" not in call_kwargs + assert call_kwargs["param3"] == "real" + + +@pytest.mark.unit +class TestTestConnection: + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_no_server_url(self, mock_setup): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool({"server_url": "", "auth_type": "none"}) + result = tool.test_connection() + assert result["success"] is False + assert "No server URL" in result["message"] + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_invalid_scheme(self, mock_setup): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool( + {"server_url": "ftp://bad.com", "auth_type": "none"} + ) + result = tool.test_connection() + assert result["success"] is False + assert "Invalid URL scheme" in result["message"] + + +@pytest.mark.unit +class TestMapError: + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_timeout_error(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + import concurrent.futures + + tool = MCPTool(mcp_config) + err = tool._map_error("test", concurrent.futures.TimeoutError()) + assert "Timed out" in str(err) + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_connection_refused(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + err = tool._map_error("test", ConnectionRefusedError()) + assert "Connection refused" in str(err) + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_403_forbidden(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + err = tool._map_error("test", Exception("403 Forbidden")) + assert "Access denied" in str(err) + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_ssl_error(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + err = tool._map_error("test", Exception("SSL certificate verify failed")) + assert "SSL" in str(err) + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_unknown_error_passthrough(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + original = RuntimeError("something weird") + err = tool._map_error("test", original) + assert err is original + + +@pytest.mark.unit +class TestGetActionsMetadata: + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_empty_tools(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + tool.available_tools = [] + assert tool.get_actions_metadata() == [] + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_tools_with_input_schema(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + tool.available_tools = [ + { + "name": "search", + "description": "Search things", + "inputSchema": { + "type": "object", + "properties": {"query": {"type": "string"}}, + "required": ["query"], + }, + } + ] + meta = tool.get_actions_metadata() + assert len(meta) == 1 + assert meta[0]["name"] == "search" + assert "query" in meta[0]["parameters"]["properties"] + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_tools_without_schema(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + tool.available_tools = [{"name": "ping", "description": "Ping"}] + meta = tool.get_actions_metadata() + assert meta[0]["parameters"]["properties"] == {} + + @patch("application.agents.tools.mcp_tool.MCPTool._setup_client") + def test_config_requirements(self, mock_setup, mcp_config): + from application.agents.tools.mcp_tool import MCPTool + + tool = MCPTool(mcp_config) + reqs = tool.get_config_requirements() + assert "server_url" in reqs + assert "auth_type" in reqs + assert reqs["server_url"]["required"] is True + + +@pytest.mark.unit +class TestMCPOAuthManager: + def test_handle_callback_success(self): + from application.agents.tools.mcp_tool import MCPOAuthManager + + mock_redis = MagicMock() + manager = MCPOAuthManager(mock_redis) + + result = manager.handle_oauth_callback(state="abc123", code="auth_code") + + assert result is True + mock_redis.setex.assert_called() + + def test_handle_callback_no_redis(self): + from application.agents.tools.mcp_tool import MCPOAuthManager + + manager = MCPOAuthManager(None) + result = manager.handle_oauth_callback(state="abc", code="code") + assert result is False + + def test_handle_callback_error(self): + from application.agents.tools.mcp_tool import MCPOAuthManager + + mock_redis = MagicMock() + manager = MCPOAuthManager(mock_redis) + + result = manager.handle_oauth_callback( + state="abc", code="", error="access_denied" + ) + assert result is False + + def test_get_oauth_status_no_task(self): + from application.agents.tools.mcp_tool import MCPOAuthManager + + manager = MCPOAuthManager(MagicMock()) + result = manager.get_oauth_status("") + assert result["status"] == "not_started" + + +@pytest.mark.unit +class TestDBTokenStorage: + def test_get_base_url(self): + from application.agents.tools.mcp_tool import DBTokenStorage + + assert ( + DBTokenStorage.get_base_url("https://mcp.example.com/api/v1") + == "https://mcp.example.com" + ) + + def test_get_db_key(self): + from application.agents.tools.mcp_tool import DBTokenStorage + + mock_db = MagicMock() + storage = DBTokenStorage( + server_url="https://mcp.example.com/api", + user_id="user1", + db_client=mock_db, + ) + key = storage.get_db_key() + assert key["server_url"] == "https://mcp.example.com" + assert key["user_id"] == "user1" diff --git a/tests/agents/test_ntfy_tool.py b/tests/agents/test_ntfy_tool.py new file mode 100644 index 00000000..ab5f0bdc --- /dev/null +++ b/tests/agents/test_ntfy_tool.py @@ -0,0 +1,146 @@ +"""Tests for application/agents/tools/ntfy.py""" + +from unittest.mock import MagicMock, patch + +import pytest + +from application.agents.tools.ntfy import NtfyTool + + +@pytest.fixture +def tool(): + return NtfyTool(config={"token": "test_token"}) + + +@pytest.fixture +def tool_no_token(): + return NtfyTool(config={}) + + +@pytest.mark.unit +class TestNtfyExecuteAction: + def test_unknown_action_raises(self, tool): + with pytest.raises(ValueError, match="Unknown action"): + tool.execute_action("bad_action") + + @patch("application.agents.tools.ntfy.requests.post") + def test_send_message_basic(self, mock_post, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_post.return_value = mock_resp + + result = tool.execute_action( + "ntfy_send_message", + server_url="https://ntfy.sh", + message="Hello", + topic="test", + ) + + assert result["status_code"] == 200 + assert result["message"] == "Message sent" + mock_post.assert_called_once() + call_args = mock_post.call_args + assert call_args[0][0] == "https://ntfy.sh/test" + assert call_args[1]["data"] == b"Hello" + + @patch("application.agents.tools.ntfy.requests.post") + def test_send_with_title_and_priority(self, mock_post, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_post.return_value = mock_resp + + tool.execute_action( + "ntfy_send_message", + server_url="https://ntfy.sh", + message="Alert", + topic="urgent", + title="Warning", + priority=5, + ) + + headers = mock_post.call_args[1]["headers"] + assert headers["X-Title"] == "Warning" + assert headers["X-Priority"] == "5" + + @patch("application.agents.tools.ntfy.requests.post") + def test_auth_header_with_token(self, mock_post, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_post.return_value = mock_resp + + tool.execute_action( + "ntfy_send_message", + server_url="https://ntfy.sh", + message="Hi", + topic="t", + ) + + headers = mock_post.call_args[1]["headers"] + assert headers["Authorization"] == "Basic test_token" + + @patch("application.agents.tools.ntfy.requests.post") + def test_no_auth_without_token(self, mock_post, tool_no_token): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_post.return_value = mock_resp + + tool_no_token.execute_action( + "ntfy_send_message", + server_url="https://ntfy.sh", + message="Hi", + topic="t", + ) + + headers = mock_post.call_args[1]["headers"] + assert "Authorization" not in headers + + def test_invalid_priority_raises(self, tool): + with pytest.raises(ValueError, match="between 1 and 5"): + tool.execute_action( + "ntfy_send_message", + server_url="https://ntfy.sh", + message="Hi", + topic="t", + priority=10, + ) + + def test_non_numeric_priority_raises(self, tool): + with pytest.raises(ValueError, match="convertible to an integer"): + tool.execute_action( + "ntfy_send_message", + server_url="https://ntfy.sh", + message="Hi", + topic="t", + priority="abc", + ) + + @patch("application.agents.tools.ntfy.requests.post") + def test_trailing_slash_stripped(self, mock_post, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_post.return_value = mock_resp + + tool.execute_action( + "ntfy_send_message", + server_url="https://ntfy.sh/", + message="Hi", + topic="test", + ) + + assert mock_post.call_args[0][0] == "https://ntfy.sh/test" + + +@pytest.mark.unit +class TestNtfyMetadata: + def test_actions_metadata(self, tool): + meta = tool.get_actions_metadata() + assert len(meta) == 1 + assert meta[0]["name"] == "ntfy_send_message" + assert "server_url" in meta[0]["parameters"]["properties"] + assert "message" in meta[0]["parameters"]["properties"] + assert "topic" in meta[0]["parameters"]["properties"] + + def test_config_requirements(self, tool): + reqs = tool.get_config_requirements() + assert "token" in reqs + assert reqs["token"]["secret"] is True diff --git a/tests/agents/test_postgres_tool.py b/tests/agents/test_postgres_tool.py new file mode 100644 index 00000000..6610883a --- /dev/null +++ b/tests/agents/test_postgres_tool.py @@ -0,0 +1,146 @@ +"""Tests for application/agents/tools/postgres.py""" + +from unittest.mock import MagicMock, patch + +import pytest + +from application.agents.tools.postgres import PostgresTool + + +@pytest.fixture +def tool(): + return PostgresTool(config={"token": "postgresql://user:pass@localhost/testdb"}) + + +@pytest.mark.unit +class TestPostgresExecuteAction: + def test_unknown_action_raises(self, tool): + with pytest.raises(ValueError, match="Unknown action"): + tool.execute_action("invalid") + + @patch("application.agents.tools.postgres.psycopg2.connect") + def test_select_query(self, mock_connect, tool): + mock_conn = MagicMock() + mock_cur = MagicMock() + mock_cur.description = [("id",), ("name",)] + mock_cur.fetchall.return_value = [(1, "Alice"), (2, "Bob")] + mock_conn.cursor.return_value = mock_cur + mock_connect.return_value = mock_conn + + result = tool.execute_action( + "postgres_execute_sql", sql_query="SELECT id, name FROM users" + ) + + assert result["status_code"] == 200 + assert result["response_data"]["column_names"] == ["id", "name"] + assert len(result["response_data"]["data"]) == 2 + assert result["response_data"]["data"][0] == {"id": 1, "name": "Alice"} + mock_conn.close.assert_called_once() + + @patch("application.agents.tools.postgres.psycopg2.connect") + def test_insert_query(self, mock_connect, tool): + mock_conn = MagicMock() + mock_cur = MagicMock() + mock_cur.rowcount = 1 + mock_conn.cursor.return_value = mock_cur + mock_connect.return_value = mock_conn + + result = tool.execute_action( + "postgres_execute_sql", + sql_query="INSERT INTO users (name) VALUES ('Alice')", + ) + + assert result["status_code"] == 200 + assert "1 rows affected" in result["response_data"]["message"] + mock_conn.commit.assert_called_once() + mock_conn.close.assert_called_once() + + @patch("application.agents.tools.postgres.psycopg2.connect") + def test_db_error(self, mock_connect, tool): + import psycopg2 + + mock_connect.side_effect = psycopg2.Error("connection refused") + + result = tool.execute_action( + "postgres_execute_sql", sql_query="SELECT 1" + ) + + assert result["status_code"] == 500 + assert "Database error" in result["error"] + + @patch("application.agents.tools.postgres.psycopg2.connect") + def test_get_schema(self, mock_connect, tool): + mock_conn = MagicMock() + mock_cur = MagicMock() + mock_cur.fetchall.return_value = [ + ("users", "id", "integer", "nextval(...)", "NO"), + ("users", "name", "varchar", None, "YES"), + ("posts", "id", "integer", "nextval(...)", "NO"), + ] + mock_conn.cursor.return_value = mock_cur + mock_connect.return_value = mock_conn + + result = tool.execute_action("postgres_get_schema", db_name="testdb") + + assert result["status_code"] == 200 + assert "users" in result["schema"] + assert "posts" in result["schema"] + assert len(result["schema"]["users"]) == 2 + assert result["schema"]["users"][0]["column_name"] == "id" + mock_conn.close.assert_called_once() + + @patch("application.agents.tools.postgres.psycopg2.connect") + def test_get_schema_db_error(self, mock_connect, tool): + import psycopg2 + + mock_connect.side_effect = psycopg2.Error("auth failed") + + result = tool.execute_action("postgres_get_schema", db_name="testdb") + + assert result["status_code"] == 500 + assert "Database error" in result["error"] + + @patch("application.agents.tools.postgres.psycopg2.connect") + def test_connection_closed_on_error(self, mock_connect, tool): + import psycopg2 + + mock_conn = MagicMock() + mock_cur = MagicMock() + mock_cur.execute.side_effect = psycopg2.Error("syntax error") + mock_conn.cursor.return_value = mock_cur + mock_connect.return_value = mock_conn + + tool.execute_action("postgres_execute_sql", sql_query="BAD SQL") + + mock_conn.close.assert_called_once() + + @patch("application.agents.tools.postgres.psycopg2.connect") + def test_select_with_no_description(self, mock_connect, tool): + mock_conn = MagicMock() + mock_cur = MagicMock() + mock_cur.description = None + mock_cur.fetchall.return_value = [] + mock_conn.cursor.return_value = mock_cur + mock_connect.return_value = mock_conn + + result = tool.execute_action( + "postgres_execute_sql", sql_query="SELECT 1 WHERE false" + ) + + assert result["status_code"] == 200 + assert result["response_data"]["column_names"] == [] + + +@pytest.mark.unit +class TestPostgresMetadata: + def test_actions_metadata(self, tool): + meta = tool.get_actions_metadata() + assert len(meta) == 2 + names = {a["name"] for a in meta} + assert "postgres_execute_sql" in names + assert "postgres_get_schema" in names + + def test_config_requirements(self, tool): + reqs = tool.get_config_requirements() + assert "token" in reqs + assert reqs["token"]["secret"] is True diff --git a/tests/agents/test_read_webpage_tool.py b/tests/agents/test_read_webpage_tool.py new file mode 100644 index 00000000..7a568ba4 --- /dev/null +++ b/tests/agents/test_read_webpage_tool.py @@ -0,0 +1,86 @@ +"""Tests for application/agents/tools/read_webpage.py""" + +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from application.agents.tools.read_webpage import ReadWebpageTool + + +@pytest.fixture +def tool(): + return ReadWebpageTool() + + +@pytest.mark.unit +class TestReadWebpageExecuteAction: + def test_unknown_action(self, tool): + result = tool.execute_action("unknown_action") + assert "Error" in result + assert "Unknown action" in result + + def test_missing_url(self, tool): + result = tool.execute_action("read_webpage") + assert "Error" in result + assert "URL parameter is missing" in result + + @patch("application.agents.tools.read_webpage.validate_url") + @patch("application.agents.tools.read_webpage.requests.get") + def test_successful_fetch(self, mock_get, mock_validate, tool): + mock_validate.return_value = "https://example.com" + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.text = "

Title

Content

" + mock_get.return_value = mock_resp + + result = tool.execute_action("read_webpage", url="https://example.com") + + assert "Title" in result + assert "Content" in result + + @patch("application.agents.tools.read_webpage.validate_url") + @patch("application.agents.tools.read_webpage.requests.get") + def test_request_error(self, mock_get, mock_validate, tool): + mock_validate.return_value = "https://example.com" + mock_get.side_effect = requests.exceptions.ConnectionError("refused") + + result = tool.execute_action("read_webpage", url="https://example.com") + + assert "Error fetching URL" in result + + @patch("application.agents.tools.read_webpage.validate_url") + def test_ssrf_blocked(self, mock_validate, tool): + from application.core.url_validation import SSRFError + + mock_validate.side_effect = SSRFError("blocked") + + result = tool.execute_action("read_webpage", url="http://169.254.169.254/") + + assert "Error" in result + assert "validation failed" in result + + @patch("application.agents.tools.read_webpage.validate_url") + @patch("application.agents.tools.read_webpage.requests.get") + def test_http_error(self, mock_get, mock_validate, tool): + mock_validate.return_value = "https://example.com/404" + mock_resp = MagicMock() + mock_resp.raise_for_status.side_effect = requests.exceptions.HTTPError("404") + mock_get.return_value = mock_resp + + result = tool.execute_action("read_webpage", url="https://example.com/404") + + assert "Error fetching URL" in result + + +@pytest.mark.unit +class TestReadWebpageMetadata: + def test_actions_metadata(self, tool): + meta = tool.get_actions_metadata() + assert len(meta) == 1 + assert meta[0]["name"] == "read_webpage" + assert "url" in meta[0]["parameters"]["properties"] + assert "url" in meta[0]["parameters"]["required"] + + def test_config_requirements(self, tool): + assert tool.get_config_requirements() == {} diff --git a/tests/agents/test_spec_parser.py b/tests/agents/test_spec_parser.py new file mode 100644 index 00000000..cb6eb779 --- /dev/null +++ b/tests/agents/test_spec_parser.py @@ -0,0 +1,335 @@ +"""Tests for application/agents/tools/spec_parser.py""" + +import json + +import pytest + +from application.agents.tools.spec_parser import ( + _extract_metadata, + _generate_action_name, + _get_base_url, + _load_spec, + _param_to_property, + _resolve_ref, + _validate_spec, + parse_spec, +) + + +MINIMAL_OPENAPI = json.dumps( + { + "openapi": "3.0.0", + "info": {"title": "Test API", "version": "1.0.0"}, + "servers": [{"url": "https://api.example.com"}], + "paths": { + "/users": { + "get": { + "operationId": "listUsers", + "summary": "List all users", + "parameters": [ + { + "name": "limit", + "in": "query", + "schema": {"type": "integer"}, + } + ], + "responses": {"200": {"description": "OK"}}, + } + } + }, + } +) + +MINIMAL_SWAGGER = json.dumps( + { + "swagger": "2.0", + "info": {"title": "Swagger API", "version": "2.0.0"}, + "host": "api.example.com", + "basePath": "/v1", + "schemes": ["https"], + "paths": { + "/pets": { + "get": { + "operationId": "listPets", + "summary": "List pets", + "parameters": [], + "responses": {"200": {"description": "OK"}}, + } + } + }, + } +) + + +@pytest.mark.unit +class TestLoadSpec: + def test_load_json(self): + spec = _load_spec('{"openapi": "3.0.0"}') + assert spec["openapi"] == "3.0.0" + + def test_load_yaml(self): + spec = _load_spec("openapi: '3.0.0'\ninfo:\n title: Test") + assert spec["openapi"] == "3.0.0" + + def test_empty_raises(self): + with pytest.raises(ValueError, match="Empty"): + _load_spec("") + + def test_whitespace_only_raises(self): + with pytest.raises(ValueError, match="Empty"): + _load_spec(" \n ") + + def test_invalid_json_raises(self): + with pytest.raises(ValueError, match="Invalid"): + _load_spec("{invalid json}") + + +@pytest.mark.unit +class TestValidateSpec: + def test_valid_openapi(self): + spec = json.loads(MINIMAL_OPENAPI) + _validate_spec(spec) # should not raise + + def test_valid_swagger(self): + spec = json.loads(MINIMAL_SWAGGER) + _validate_spec(spec) + + def test_missing_version_raises(self): + with pytest.raises(ValueError, match="Unsupported"): + _validate_spec({"paths": {"/a": {}}}) + + def test_no_paths_raises(self): + with pytest.raises(ValueError, match="No API paths"): + _validate_spec({"openapi": "3.0.0"}) + + def test_empty_paths_raises(self): + with pytest.raises(ValueError, match="No API paths"): + _validate_spec({"openapi": "3.0.0", "paths": {}}) + + def test_non_dict_raises(self): + with pytest.raises(ValueError, match="valid object"): + _validate_spec("not a dict") + + +@pytest.mark.unit +class TestExtractMetadata: + def test_openapi_metadata(self): + spec = json.loads(MINIMAL_OPENAPI) + meta = _extract_metadata(spec, is_swagger=False) + assert meta["title"] == "Test API" + assert meta["version"] == "1.0.0" + assert meta["base_url"] == "https://api.example.com" + + def test_swagger_metadata(self): + spec = json.loads(MINIMAL_SWAGGER) + meta = _extract_metadata(spec, is_swagger=True) + assert meta["title"] == "Swagger API" + assert meta["base_url"] == "https://api.example.com/v1" + + def test_missing_info(self): + spec = {"openapi": "3.0.0", "paths": {"/a": {}}} + meta = _extract_metadata(spec, is_swagger=False) + assert meta["title"] == "Untitled API" + + def test_description_truncated(self): + spec = { + "openapi": "3.0.0", + "info": {"title": "T", "description": "x" * 1000}, + "paths": {"/a": {}}, + } + meta = _extract_metadata(spec, is_swagger=False) + assert len(meta["description"]) <= 500 + + +@pytest.mark.unit +class TestGetBaseUrl: + def test_openapi_servers(self): + spec = {"servers": [{"url": "https://api.example.com/v2/"}]} + assert _get_base_url(spec, is_swagger=False) == "https://api.example.com/v2" + + def test_openapi_no_servers(self): + assert _get_base_url({}, is_swagger=False) == "" + + def test_swagger_with_host(self): + spec = {"host": "api.test.com", "basePath": "/v1", "schemes": ["https"]} + assert _get_base_url(spec, is_swagger=True) == "https://api.test.com/v1" + + def test_swagger_no_host(self): + assert _get_base_url({}, is_swagger=True) == "" + + def test_swagger_default_scheme(self): + spec = {"host": "api.test.com", "basePath": ""} + assert _get_base_url(spec, is_swagger=True) == "https://api.test.com" + + +@pytest.mark.unit +class TestGenerateActionName: + def test_uses_operation_id(self): + assert _generate_action_name({"operationId": "getUser"}, "get", "/users") == "getUser" + + def test_fallback_to_method_path(self): + name = _generate_action_name({}, "get", "/users/{id}") + assert name.startswith("get_") + assert "users" in name + + def test_truncates_long_names(self): + name = _generate_action_name({}, "get", "/" + "a" * 200) + assert len(name) <= 64 + + def test_sanitizes_special_chars(self): + name = _generate_action_name({"operationId": "get.user@v2"}, "get", "/") + assert "." not in name + assert "@" not in name + + +@pytest.mark.unit +class TestParamToProperty: + def test_string_param(self): + param = {"name": "q", "in": "query", "schema": {"type": "string"}} + prop = _param_to_property(param) + assert prop["type"] == "string" + assert prop["required"] is False + + def test_integer_param(self): + param = { + "name": "limit", + "in": "query", + "required": True, + "schema": {"type": "integer"}, + } + prop = _param_to_property(param) + assert prop["type"] == "integer" + assert prop["required"] is True + assert prop["filled_by_llm"] is True + + def test_number_maps_to_integer(self): + param = {"name": "score", "in": "query", "schema": {"type": "number"}} + prop = _param_to_property(param) + assert prop["type"] == "integer" + + +@pytest.mark.unit +class TestResolveRef: + def test_no_ref(self): + obj = {"type": "string"} + assert _resolve_ref(obj, {}, {}) == obj + + def test_components_ref(self): + components = {"schemas": {"User": {"type": "object", "properties": {"name": {"type": "string"}}}}} + obj = {"$ref": "#/components/schemas/User"} + result = _resolve_ref(obj, components, {}) + assert result["type"] == "object" + + def test_definitions_ref(self): + definitions = {"Pet": {"type": "object"}} + obj = {"$ref": "#/definitions/Pet"} + result = _resolve_ref(obj, {}, definitions) + assert result["type"] == "object" + + def test_unsupported_ref(self): + obj = {"$ref": "#/external/something"} + assert _resolve_ref(obj, {}, {}) is None + + def test_non_dict_returns_none(self): + assert _resolve_ref("string", {}, {}) is None + assert _resolve_ref(42, {}, {}) is None + + +@pytest.mark.unit +class TestParseSpec: + def test_openapi_full_parse(self): + metadata, actions = parse_spec(MINIMAL_OPENAPI) + assert metadata["title"] == "Test API" + assert len(actions) == 1 + assert actions[0]["name"] == "listUsers" + assert actions[0]["method"] == "GET" + assert actions[0]["url"] == "https://api.example.com/users" + assert "limit" in actions[0]["query_params"]["properties"] + + def test_swagger_full_parse(self): + metadata, actions = parse_spec(MINIMAL_SWAGGER) + assert metadata["title"] == "Swagger API" + assert len(actions) == 1 + assert actions[0]["name"] == "listPets" + assert actions[0]["method"] == "GET" + + def test_multiple_methods(self): + spec = json.dumps( + { + "openapi": "3.0.0", + "info": {"title": "T", "version": "1"}, + "paths": { + "/items": { + "get": {"operationId": "listItems", "responses": {}}, + "post": { + "operationId": "createItem", + "requestBody": { + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "name": {"type": "string"} + }, + "required": ["name"], + } + } + } + }, + "responses": {}, + }, + } + }, + } + ) + metadata, actions = parse_spec(spec) + assert len(actions) == 2 + names = {a["name"] for a in actions} + assert "listItems" in names + assert "createItem" in names + + create = next(a for a in actions if a["name"] == "createItem") + assert "name" in create["body"]["properties"] + + def test_header_params(self): + spec = json.dumps( + { + "openapi": "3.0.0", + "info": {"title": "T", "version": "1"}, + "paths": { + "/data": { + "get": { + "operationId": "getData", + "parameters": [ + {"name": "X-API-Key", "in": "header", "schema": {"type": "string"}} + ], + "responses": {}, + } + } + }, + } + ) + _, actions = parse_spec(spec) + assert "X-API-Key" in actions[0]["headers"]["properties"] + + def test_invalid_spec_raises(self): + with pytest.raises(ValueError): + parse_spec("") + + def test_yaml_spec(self): + yaml_spec = """ +openapi: "3.0.0" +info: + title: YAML API + version: "1.0" +paths: + /health: + get: + operationId: healthCheck + responses: + "200": + description: OK +""" + metadata, actions = parse_spec(yaml_spec) + assert metadata["title"] == "YAML API" + assert actions[0]["name"] == "healthCheck" diff --git a/tests/agents/test_telegram_tool.py b/tests/agents/test_telegram_tool.py new file mode 100644 index 00000000..f0e8b0e9 --- /dev/null +++ b/tests/agents/test_telegram_tool.py @@ -0,0 +1,79 @@ +"""Tests for application/agents/tools/telegram.py""" + +from unittest.mock import MagicMock, patch + +import pytest + +from application.agents.tools.telegram import TelegramTool + + +@pytest.fixture +def tool(): + return TelegramTool(config={"token": "bot123:ABC"}) + + +@pytest.mark.unit +class TestTelegramExecuteAction: + def test_unknown_action_raises(self, tool): + with pytest.raises(ValueError, match="Unknown action"): + tool.execute_action("invalid") + + @patch("application.agents.tools.telegram.requests.post") + def test_send_message(self, mock_post, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_post.return_value = mock_resp + + result = tool.execute_action( + "telegram_send_message", text="Hello", chat_id="12345" + ) + + assert result["status_code"] == 200 + assert result["message"] == "Message sent" + call_args = mock_post.call_args + assert "bot123:ABC/sendMessage" in call_args[0][0] + assert call_args[1]["data"]["text"] == "Hello" + assert call_args[1]["data"]["chat_id"] == "12345" + + @patch("application.agents.tools.telegram.requests.post") + def test_send_image(self, mock_post, tool): + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_post.return_value = mock_resp + + result = tool.execute_action( + "telegram_send_image", image_url="https://img.com/cat.jpg", chat_id="12345" + ) + + assert result["status_code"] == 200 + assert result["message"] == "Image sent" + call_args = mock_post.call_args + assert "bot123:ABC/sendPhoto" in call_args[0][0] + assert call_args[1]["data"]["photo"] == "https://img.com/cat.jpg" + + @patch("application.agents.tools.telegram.requests.post") + def test_api_error_status(self, mock_post, tool): + mock_resp = MagicMock() + mock_resp.status_code = 403 + mock_post.return_value = mock_resp + + result = tool.execute_action( + "telegram_send_message", text="Hi", chat_id="999" + ) + + assert result["status_code"] == 403 + + +@pytest.mark.unit +class TestTelegramMetadata: + def test_actions_metadata(self, tool): + meta = tool.get_actions_metadata() + assert len(meta) == 2 + names = {a["name"] for a in meta} + assert "telegram_send_message" in names + assert "telegram_send_image" in names + + def test_config_requirements(self, tool): + reqs = tool.get_config_requirements() + assert "token" in reqs + assert reqs["token"]["secret"] is True diff --git a/tests/api/answer/routes/test_answer.py b/tests/api/answer/routes/test_answer.py new file mode 100644 index 00000000..53f2525d --- /dev/null +++ b/tests/api/answer/routes/test_answer.py @@ -0,0 +1,235 @@ +"""Tests for application/api/answer/routes/answer.py""" + +import json +from unittest.mock import MagicMock, patch + +import pytest +from bson import ObjectId + + +@pytest.fixture +def mock_stream_processor(): + """Create a mock StreamProcessor.""" + with patch( + "application.api.answer.routes.answer.StreamProcessor" + ) as MockProcessor: + processor = MagicMock() + processor.decoded_token = {"sub": "test_user"} + processor.conversation_id = str(ObjectId()) + processor.agent_config = {} + processor.agent_id = str(ObjectId()) + processor.is_shared_usage = False + processor.shared_token = None + processor.model_id = "gpt-4" + processor.build_agent.return_value = MagicMock() + MockProcessor.return_value = processor + yield processor + + +@pytest.fixture +def answer_client(mock_mongo_db, flask_app): + """Create a test client with the answer route registered.""" + from flask_restx import Api + + from application.api.answer.routes.answer import answer_ns + + api = Api(flask_app) + api.add_namespace(answer_ns) + flask_app.config["TESTING"] = True + return flask_app.test_client() + + +@pytest.mark.unit +class TestAnswerResourcePost: + def test_missing_question_returns_400(self, answer_client, mock_stream_processor): + resp = answer_client.post( + "/api/answer", + data=json.dumps({}), + content_type="application/json", + ) + assert resp.status_code == 400 + + def test_successful_answer(self, answer_client, mock_stream_processor): + conv_id = str(ObjectId()) + with patch.object( + mock_stream_processor.build_agent.return_value, + "gen", + return_value=iter([]), + ): + with patch( + "application.api.answer.routes.answer.AnswerResource.validate_request", + return_value=None, + ), patch( + "application.api.answer.routes.answer.AnswerResource.check_usage", + return_value=None, + ), patch( + "application.api.answer.routes.answer.AnswerResource.complete_stream", + return_value=iter( + [ + f'data: {json.dumps({"type": "answer", "answer": "Hello"})}\n\n', + f'data: {json.dumps({"type": "id", "id": conv_id})}\n\n', + f'data: {json.dumps({"type": "end"})}\n\n', + ] + ), + ), patch( + "application.api.answer.routes.answer.AnswerResource.process_response_stream", + return_value=(conv_id, "Hello", [], [], "", None), + ): + resp = answer_client.post( + "/api/answer", + data=json.dumps({"question": "What is Python?"}), + content_type="application/json", + ) + assert resp.status_code == 200 + data = resp.get_json() + assert data["answer"] == "Hello" + assert data["conversation_id"] == conv_id + + def test_unauthorized_returns_401(self, answer_client, mock_stream_processor): + mock_stream_processor.decoded_token = None + with patch( + "application.api.answer.routes.answer.AnswerResource.validate_request", + return_value=None, + ): + resp = answer_client.post( + "/api/answer", + data=json.dumps({"question": "test"}), + content_type="application/json", + ) + assert resp.status_code == 401 + assert resp.get_json()["error"] == "Unauthorized" + + def test_usage_exceeded_returns_error(self, answer_client, mock_stream_processor): + + with patch( + "application.api.answer.routes.answer.AnswerResource.validate_request", + return_value=None, + ), patch( + "application.api.answer.routes.answer.AnswerResource.check_usage", + ) as mock_check: + with flask_app_context(answer_client): + mock_check.return_value = ({"error": "Usage limit exceeded"}, 429) + + resp = answer_client.post( + "/api/answer", + data=json.dumps({"question": "test"}), + content_type="application/json", + ) + assert resp.status_code == 429 + + def test_stream_error_returns_400(self, answer_client, mock_stream_processor): + with patch( + "application.api.answer.routes.answer.AnswerResource.validate_request", + return_value=None, + ), patch( + "application.api.answer.routes.answer.AnswerResource.check_usage", + return_value=None, + ), patch( + "application.api.answer.routes.answer.AnswerResource.complete_stream", + return_value=iter([]), + ), patch( + "application.api.answer.routes.answer.AnswerResource.process_response_stream", + return_value=(None, None, None, None, None, "Stream error"), + ): + resp = answer_client.post( + "/api/answer", + data=json.dumps({"question": "test"}), + content_type="application/json", + ) + assert resp.status_code == 400 + assert resp.get_json()["error"] == "Stream error" + + def test_exception_returns_500(self, answer_client, mock_stream_processor): + with patch( + "application.api.answer.routes.answer.AnswerResource.validate_request", + return_value=None, + ), patch( + "application.api.answer.routes.answer.AnswerResource.check_usage", + return_value=None, + ), patch( + "application.api.answer.routes.answer.AnswerResource.complete_stream", + side_effect=RuntimeError("unexpected"), + ): + resp = answer_client.post( + "/api/answer", + data=json.dumps({"question": "test"}), + content_type="application/json", + ) + assert resp.status_code == 500 + assert "error" in resp.get_json() + + def test_structured_info_merged_into_result( + self, answer_client, mock_stream_processor + ): + conv_id = str(ObjectId()) + with patch( + "application.api.answer.routes.answer.AnswerResource.validate_request", + return_value=None, + ), patch( + "application.api.answer.routes.answer.AnswerResource.check_usage", + return_value=None, + ), patch( + "application.api.answer.routes.answer.AnswerResource.complete_stream", + return_value=iter([]), + ), patch( + "application.api.answer.routes.answer.AnswerResource.process_response_stream", + return_value=( + conv_id, + '{"key": "val"}', + [], + [], + "", + None, + {"structured": True, "schema": {"type": "object"}}, + ), + ): + resp = answer_client.post( + "/api/answer", + data=json.dumps({"question": "test"}), + content_type="application/json", + ) + assert resp.status_code == 200 + data = resp.get_json() + assert data["structured"] is True + assert data["schema"] == {"type": "object"} + + def test_result_contains_all_expected_fields( + self, answer_client, mock_stream_processor + ): + conv_id = str(ObjectId()) + with patch( + "application.api.answer.routes.answer.AnswerResource.validate_request", + return_value=None, + ), patch( + "application.api.answer.routes.answer.AnswerResource.check_usage", + return_value=None, + ), patch( + "application.api.answer.routes.answer.AnswerResource.complete_stream", + return_value=iter([]), + ), patch( + "application.api.answer.routes.answer.AnswerResource.process_response_stream", + return_value=( + conv_id, + "answer text", + [{"title": "src"}], + [{"tool": "t"}], + "thinking...", + None, + ), + ): + resp = answer_client.post( + "/api/answer", + data=json.dumps({"question": "test"}), + content_type="application/json", + ) + data = resp.get_json() + assert data["conversation_id"] == conv_id + assert data["answer"] == "answer text" + assert data["sources"] == [{"title": "src"}] + assert data["tool_calls"] == [{"tool": "t"}] + assert data["thought"] == "thinking..." + + +def flask_app_context(client): + """Helper to get app context from test client.""" + return client.application.app_context() diff --git a/tests/api/answer/routes/test_stream.py b/tests/api/answer/routes/test_stream.py new file mode 100644 index 00000000..f025bfd0 --- /dev/null +++ b/tests/api/answer/routes/test_stream.py @@ -0,0 +1,195 @@ +"""Tests for application/api/answer/routes/stream.py""" + +import json +from unittest.mock import MagicMock, patch + +import pytest +from bson import ObjectId + + +@pytest.fixture +def mock_stream_processor(): + """Create a mock StreamProcessor for stream tests.""" + with patch( + "application.api.answer.routes.stream.StreamProcessor" + ) as MockProcessor: + processor = MagicMock() + processor.decoded_token = {"sub": "test_user"} + processor.conversation_id = str(ObjectId()) + processor.agent_config = {} + processor.agent_id = str(ObjectId()) + processor.is_shared_usage = False + processor.shared_token = None + processor.model_id = "gpt-4" + processor.build_agent.return_value = MagicMock() + MockProcessor.return_value = processor + yield processor + + +@pytest.fixture +def stream_client(mock_mongo_db, flask_app): + """Create a test client with the stream route registered.""" + from flask_restx import Api + + from application.api.answer.routes.stream import answer_ns + + api = Api(flask_app) + api.add_namespace(answer_ns) + flask_app.config["TESTING"] = True + return flask_app.test_client() + + +@pytest.mark.unit +class TestStreamResourcePost: + def test_missing_question_returns_400(self, stream_client, mock_stream_processor): + resp = stream_client.post( + "/stream", + data=json.dumps({}), + content_type="application/json", + ) + assert resp.status_code == 400 + + def test_successful_stream(self, stream_client, mock_stream_processor): + def fake_stream(*args, **kwargs): + yield f'data: {json.dumps({"type": "answer", "answer": "Hi"})}\n\n' + yield f'data: {json.dumps({"type": "end"})}\n\n' + + with patch( + "application.api.answer.routes.stream.StreamResource.validate_request", + return_value=None, + ), patch( + "application.api.answer.routes.stream.StreamResource.check_usage", + return_value=None, + ), patch( + "application.api.answer.routes.stream.StreamResource.complete_stream", + side_effect=fake_stream, + ): + resp = stream_client.post( + "/stream", + data=json.dumps({"question": "What is Python?"}), + content_type="application/json", + ) + assert resp.status_code == 200 + assert "text/event-stream" in resp.content_type + data = resp.get_data(as_text=True) + assert '"type": "answer"' in data + assert '"answer": "Hi"' in data + + def test_unauthorized_returns_401_stream( + self, stream_client, mock_stream_processor + ): + mock_stream_processor.decoded_token = None + with patch( + "application.api.answer.routes.stream.StreamResource.validate_request", + return_value=None, + ): + resp = stream_client.post( + "/stream", + data=json.dumps({"question": "test"}), + content_type="application/json", + ) + assert resp.status_code == 401 + assert "text/event-stream" in resp.content_type + data = resp.get_data(as_text=True) + assert "Unauthorized" in data + + def test_usage_exceeded_returns_error( + self, stream_client, mock_stream_processor + ): + with patch( + "application.api.answer.routes.stream.StreamResource.validate_request", + return_value=None, + ), patch( + "application.api.answer.routes.stream.StreamResource.check_usage", + ) as mock_check: + mock_check.return_value = ({"error": "Usage limit exceeded"}, 429) + resp = stream_client.post( + "/stream", + data=json.dumps({"question": "test"}), + content_type="application/json", + ) + assert resp.status_code == 429 + + def test_value_error_returns_400_stream( + self, stream_client, mock_stream_processor + ): + mock_stream_processor.build_agent.side_effect = ValueError("bad data") + with patch( + "application.api.answer.routes.stream.StreamResource.validate_request", + return_value=None, + ): + resp = stream_client.post( + "/stream", + data=json.dumps({"question": "test"}), + content_type="application/json", + ) + assert resp.status_code == 400 + assert "text/event-stream" in resp.content_type + data = resp.get_data(as_text=True) + assert "Malformed request body" in data + + def test_general_exception_returns_400_stream( + self, stream_client, mock_stream_processor + ): + mock_stream_processor.build_agent.side_effect = RuntimeError("crash") + with patch( + "application.api.answer.routes.stream.StreamResource.validate_request", + return_value=None, + ): + resp = stream_client.post( + "/stream", + data=json.dumps({"question": "test"}), + content_type="application/json", + ) + assert resp.status_code == 400 + assert "text/event-stream" in resp.content_type + data = resp.get_data(as_text=True) + assert "Unknown error occurred" in data + + def test_index_in_data_requires_conversation_id( + self, stream_client, mock_stream_processor + ): + """When 'index' is present, validate_request is called with require_conversation_id=True.""" + resp = stream_client.post( + "/stream", + data=json.dumps({"question": "test", "index": 0}), + content_type="application/json", + ) + # Should get 400 since conversation_id is missing + assert resp.status_code == 400 + + def test_stream_passes_attachments_and_index( + self, stream_client, mock_stream_processor + ): + """Verify attachments and index params are forwarded to complete_stream.""" + + def fake_stream(*args, **kwargs): + yield f'data: {json.dumps({"type": "end"})}\n\n' + + conv_id = str(ObjectId()) + with patch( + "application.api.answer.routes.stream.StreamResource.validate_request", + return_value=None, + ), patch( + "application.api.answer.routes.stream.StreamResource.check_usage", + return_value=None, + ), patch( + "application.api.answer.routes.stream.StreamResource.complete_stream", + side_effect=fake_stream, + ) as mock_complete: + resp = stream_client.post( + "/stream", + data=json.dumps( + { + "question": "test", + "conversation_id": conv_id, + "index": 3, + "attachments": ["att1", "att2"], + } + ), + content_type="application/json", + ) + assert resp.status_code == 200 + call_kwargs = mock_complete.call_args + assert call_kwargs.kwargs.get("index") == 3 + assert call_kwargs.kwargs.get("attachment_ids") == ["att1", "att2"] diff --git a/tests/api/answer/services/compression/__init__.py b/tests/api/answer/services/compression/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/api/answer/services/compression/test_message_builder.py b/tests/api/answer/services/compression/test_message_builder.py new file mode 100644 index 00000000..9cb681d4 --- /dev/null +++ b/tests/api/answer/services/compression/test_message_builder.py @@ -0,0 +1,303 @@ +"""Tests for application/api/answer/services/compression/message_builder.py""" + + +import pytest + +from application.api.answer.services.compression.message_builder import MessageBuilder + + +@pytest.mark.unit +class TestBuildFromCompressedContext: + def test_no_compression_returns_system_only(self): + messages = MessageBuilder.build_from_compressed_context( + system_prompt="You are helpful.", + compressed_summary=None, + recent_queries=[], + ) + assert len(messages) == 1 + assert messages[0]["role"] == "system" + assert messages[0]["content"] == "You are helpful." + + def test_with_recent_queries_no_compression(self): + queries = [ + {"prompt": "Hello", "response": "Hi there!"}, + {"prompt": "How are you?", "response": "I'm fine."}, + ] + messages = MessageBuilder.build_from_compressed_context( + system_prompt="System prompt", + compressed_summary=None, + recent_queries=queries, + ) + # system + 2 * (user + assistant) = 5 + assert len(messages) == 5 + assert messages[1] == {"role": "user", "content": "Hello"} + assert messages[2] == {"role": "assistant", "content": "Hi there!"} + assert messages[3] == {"role": "user", "content": "How are you?"} + assert messages[4] == {"role": "assistant", "content": "I'm fine."} + + def test_with_compressed_summary_appended_to_system(self): + messages = MessageBuilder.build_from_compressed_context( + system_prompt="You are helpful.", + compressed_summary="Previous: user asked about Python.", + recent_queries=[{"prompt": "More?", "response": "Sure."}], + ) + system_content = messages[0]["content"] + assert "This session is being continued" in system_content + assert "Previous: user asked about Python." in system_content + + def test_mid_execution_context_type(self): + messages = MessageBuilder.build_from_compressed_context( + system_prompt="System", + compressed_summary="Summary here", + recent_queries=[{"prompt": "q", "response": "r"}], + context_type="mid_execution", + ) + system_content = messages[0]["content"] + assert "Context window limit reached" in system_content + + def test_include_tool_calls(self): + queries = [ + { + "prompt": "Search for X", + "response": "Found X", + "tool_calls": [ + { + "call_id": "call-1", + "action_name": "search", + "arguments": {"q": "X"}, + "result": "X found", + } + ], + } + ] + messages = MessageBuilder.build_from_compressed_context( + system_prompt="System", + compressed_summary=None, + recent_queries=queries, + include_tool_calls=True, + ) + # system + user + assistant + tool_call_assistant + tool_response = 5 + assert len(messages) == 5 + assert messages[3]["role"] == "assistant" + assert "function_call" in messages[3]["content"][0] + assert messages[4]["role"] == "tool" + assert "function_response" in messages[4]["content"][0] + + def test_tool_calls_not_included_by_default(self): + queries = [ + { + "prompt": "Search", + "response": "Found", + "tool_calls": [ + { + "call_id": "c1", + "action_name": "search", + "arguments": {}, + "result": "ok", + } + ], + } + ] + messages = MessageBuilder.build_from_compressed_context( + system_prompt="System", + compressed_summary=None, + recent_queries=queries, + include_tool_calls=False, + ) + # system + user + assistant = 3 (no tool messages) + assert len(messages) == 3 + + def test_tool_call_without_call_id_generates_uuid(self): + queries = [ + { + "prompt": "q", + "response": "r", + "tool_calls": [ + { + "action_name": "act", + "arguments": {}, + "result": "res", + } + ], + } + ] + messages = MessageBuilder.build_from_compressed_context( + system_prompt="S", + compressed_summary=None, + recent_queries=queries, + include_tool_calls=True, + ) + tool_msg = messages[3]["content"][0] + call_id = tool_msg["function_call"]["call_id"] + assert call_id is not None + assert len(call_id) > 0 + + def test_continuation_message_when_no_recent_queries_but_has_summary(self): + messages = MessageBuilder.build_from_compressed_context( + system_prompt="System", + compressed_summary="Everything was compressed", + recent_queries=[], + ) + # system + continuation user message = 2 + assert len(messages) == 2 + assert messages[1]["role"] == "user" + assert "continue" in messages[1]["content"].lower() + + def test_no_continuation_when_no_summary(self): + messages = MessageBuilder.build_from_compressed_context( + system_prompt="System", + compressed_summary=None, + recent_queries=[], + ) + assert len(messages) == 1 + + def test_queries_without_prompt_or_response_skipped(self): + queries = [ + {"other_field": "value"}, + {"prompt": "real", "response": "answer"}, + ] + messages = MessageBuilder.build_from_compressed_context( + system_prompt="S", + compressed_summary=None, + recent_queries=queries, + ) + # system + 1 valid query (user + assistant) = 3 + assert len(messages) == 3 + + +@pytest.mark.unit +class TestAppendCompressionContext: + def test_pre_request_context(self): + result = MessageBuilder._append_compression_context( + "Original prompt", "Summary text", "pre_request" + ) + assert "This session is being continued" in result + assert "Summary text" in result + assert result.startswith("Original prompt") + + def test_mid_execution_context(self): + result = MessageBuilder._append_compression_context( + "Original prompt", "Summary text", "mid_execution" + ) + assert "Context window limit reached" in result + assert "Summary text" in result + + def test_removes_existing_compression_context(self): + prompt_with_existing = ( + "Original prompt\n\n---\n\nThis session is being continued from old" + ) + result = MessageBuilder._append_compression_context( + prompt_with_existing, "New summary", "pre_request" + ) + # Should not contain old context twice + assert result.count("This session is being continued") == 1 + assert "New summary" in result + + def test_removes_mid_execution_context(self): + prompt_with_existing = ( + "Original\n\n---\n\nContext window limit reached during execution. Old." + ) + result = MessageBuilder._append_compression_context( + prompt_with_existing, "New", "mid_execution" + ) + assert result.count("Context window limit reached") == 1 + + +@pytest.mark.unit +class TestRebuildMessagesAfterCompression: + def test_basic_rebuild(self): + messages = [ + {"role": "system", "content": "You are helpful."}, + {"role": "user", "content": "old message"}, + {"role": "assistant", "content": "old reply"}, + ] + recent = [{"prompt": "new q", "response": "new r"}] + + result = MessageBuilder.rebuild_messages_after_compression( + messages=messages, + compressed_summary="Everything was compressed.", + recent_queries=recent, + ) + assert result is not None + # system + user + assistant = 3 + assert len(result) == 3 + assert "Context window limit reached" in result[0]["content"] + assert result[1] == {"role": "user", "content": "new q"} + assert result[2] == {"role": "assistant", "content": "new r"} + + def test_returns_none_without_system_message(self): + messages = [ + {"role": "user", "content": "hello"}, + ] + result = MessageBuilder.rebuild_messages_after_compression( + messages=messages, + compressed_summary="summary", + recent_queries=[], + ) + assert result is None + + def test_no_summary_keeps_system_unchanged(self): + messages = [{"role": "system", "content": "Be helpful."}] + result = MessageBuilder.rebuild_messages_after_compression( + messages=messages, + compressed_summary=None, + recent_queries=[], + ) + assert result is not None + assert result[0]["content"] == "Be helpful." + + def test_include_tool_calls_in_rebuild(self): + messages = [{"role": "system", "content": "S"}] + recent = [ + { + "prompt": "q", + "response": "r", + "tool_calls": [ + { + "call_id": "c1", + "action_name": "act", + "arguments": {"a": 1}, + "result": "done", + } + ], + } + ] + result = MessageBuilder.rebuild_messages_after_compression( + messages=messages, + compressed_summary="s", + recent_queries=recent, + include_tool_calls=True, + ) + # system + user + assistant + tool_call + tool_response = 5 + assert len(result) == 5 + + def test_continuation_added_when_no_recent_queries(self): + messages = [{"role": "system", "content": "S"}] + result = MessageBuilder.rebuild_messages_after_compression( + messages=messages, + compressed_summary="All compressed", + recent_queries=[], + ) + assert len(result) == 2 + assert result[1]["role"] == "user" + assert "continue" in result[1]["content"].lower() + + def test_include_current_execution_preserves_extra_messages(self): + messages = [ + {"role": "system", "content": "S"}, + {"role": "user", "content": "q1"}, + {"role": "assistant", "content": "r1"}, + {"role": "user", "content": "current execution msg"}, + ] + recent = [{"prompt": "q1", "response": "r1"}] + + result = MessageBuilder.rebuild_messages_after_compression( + messages=messages, + compressed_summary="summary", + recent_queries=recent, + include_current_execution=True, + ) + assert result is not None + # Should include the current execution message + contents = [m.get("content") for m in result] + assert "current execution msg" in contents diff --git a/tests/api/answer/services/compression/test_orchestrator.py b/tests/api/answer/services/compression/test_orchestrator.py new file mode 100644 index 00000000..67701614 --- /dev/null +++ b/tests/api/answer/services/compression/test_orchestrator.py @@ -0,0 +1,447 @@ +"""Tests for application/api/answer/services/compression/orchestrator.py""" + +from unittest.mock import MagicMock, patch + +import pytest + +from application.api.answer.services.compression.orchestrator import ( + CompressionOrchestrator, +) +from application.api.answer.services.compression.types import ( + CompressionMetadata, + CompressionResult, +) + + +@pytest.fixture +def mock_conversation_service(): + svc = MagicMock() + return svc + + +@pytest.fixture +def mock_threshold_checker(): + checker = MagicMock() + return checker + + +@pytest.fixture +def orchestrator(mock_conversation_service, mock_threshold_checker): + return CompressionOrchestrator( + conversation_service=mock_conversation_service, + threshold_checker=mock_threshold_checker, + ) + + +@pytest.fixture +def sample_conversation(): + return { + "queries": [ + {"prompt": "q0", "response": "r0"}, + {"prompt": "q1", "response": "r1"}, + {"prompt": "q2", "response": "r2"}, + ], + "compression_metadata": {}, + "agent_id": "agent-1", + } + + +@pytest.fixture +def decoded_token(): + return {"sub": "user123"} + + +@pytest.mark.unit +class TestCompressIfNeeded: + def test_conversation_not_found_returns_failure( + self, orchestrator, mock_conversation_service + ): + mock_conversation_service.get_conversation.return_value = None + + result = orchestrator.compress_if_needed( + conversation_id="conv1", + user_id="user1", + model_id="gpt-4", + decoded_token={"sub": "user1"}, + ) + + assert result.success is False + assert "not found" in result.error + + def test_no_compression_needed( + self, + orchestrator, + mock_conversation_service, + mock_threshold_checker, + sample_conversation, + ): + mock_conversation_service.get_conversation.return_value = sample_conversation + mock_threshold_checker.should_compress.return_value = False + + result = orchestrator.compress_if_needed( + conversation_id="conv1", + user_id="user1", + model_id="gpt-4", + decoded_token={"sub": "user1"}, + ) + + assert result.success is True + assert result.compression_performed is False + assert len(result.recent_queries) == 3 + + def test_compression_performed_successfully( + self, + orchestrator, + mock_conversation_service, + mock_threshold_checker, + sample_conversation, + decoded_token, + ): + mock_conversation_service.get_conversation.return_value = sample_conversation + mock_threshold_checker.should_compress.return_value = True + + mock_metadata = MagicMock(spec=CompressionMetadata) + mock_metadata.compression_ratio = 5.0 + mock_metadata.original_token_count = 1000 + mock_metadata.compressed_token_count = 200 + mock_metadata.to_dict.return_value = {"query_index": 2} + + with patch.object( + orchestrator, "_perform_compression" + ) as mock_perform: + mock_perform.return_value = CompressionResult.success_with_compression( + "compressed summary", + [{"prompt": "q2", "response": "r2"}], + mock_metadata, + ) + + result = orchestrator.compress_if_needed( + conversation_id="conv1", + user_id="user1", + model_id="gpt-4", + decoded_token=decoded_token, + ) + + assert result.success is True + assert result.compression_performed is True + assert result.compressed_summary == "compressed summary" + mock_perform.assert_called_once() + + def test_exception_returns_failure( + self, + orchestrator, + mock_conversation_service, + ): + mock_conversation_service.get_conversation.side_effect = RuntimeError("DB down") + + result = orchestrator.compress_if_needed( + conversation_id="conv1", + user_id="user1", + model_id="gpt-4", + decoded_token={"sub": "user1"}, + ) + + assert result.success is False + assert "DB down" in result.error + + def test_custom_query_tokens( + self, + orchestrator, + mock_conversation_service, + mock_threshold_checker, + sample_conversation, + ): + mock_conversation_service.get_conversation.return_value = sample_conversation + mock_threshold_checker.should_compress.return_value = False + + orchestrator.compress_if_needed( + conversation_id="conv1", + user_id="user1", + model_id="gpt-4", + decoded_token={"sub": "user1"}, + current_query_tokens=1000, + ) + + mock_threshold_checker.should_compress.assert_called_once_with( + sample_conversation, "gpt-4", 1000 + ) + + +@pytest.mark.unit +class TestPerformCompression: + @patch( + "application.api.answer.services.compression.orchestrator.get_provider_from_model_id" + ) + @patch( + "application.api.answer.services.compression.orchestrator.get_api_key_for_provider" + ) + @patch("application.api.answer.services.compression.orchestrator.LLMCreator") + @patch("application.api.answer.services.compression.orchestrator.CompressionService") + @patch("application.api.answer.services.compression.orchestrator.settings") + def test_successful_compression( + self, + mock_settings, + MockCompressionService, + MockLLMCreator, + mock_get_api_key, + mock_get_provider, + mock_conversation_service, + mock_threshold_checker, + sample_conversation, + decoded_token, + ): + mock_settings.COMPRESSION_MODEL_OVERRIDE = None + mock_get_provider.return_value = "openai" + mock_get_api_key.return_value = "sk-test" + MockLLMCreator.create_llm.return_value = MagicMock() + + mock_metadata = MagicMock(spec=CompressionMetadata) + mock_metadata.compression_ratio = 5.0 + mock_metadata.original_token_count = 500 + mock_metadata.compressed_token_count = 100 + + mock_svc_instance = MagicMock() + mock_svc_instance.compress_and_save.return_value = mock_metadata + mock_svc_instance.get_compressed_context.return_value = ( + "compressed text", + [{"prompt": "q2", "response": "r2"}], + ) + MockCompressionService.return_value = mock_svc_instance + + # After compression, reload conversation + mock_conversation_service.get_conversation.return_value = sample_conversation + + orch = CompressionOrchestrator( + conversation_service=mock_conversation_service, + threshold_checker=mock_threshold_checker, + ) + + result = orch._perform_compression( + "conv1", sample_conversation, "gpt-4", decoded_token + ) + + assert result.success is True + assert result.compression_performed is True + assert result.compressed_summary == "compressed text" + mock_svc_instance.compress_and_save.assert_called_once() + + @patch( + "application.api.answer.services.compression.orchestrator.get_provider_from_model_id" + ) + @patch( + "application.api.answer.services.compression.orchestrator.get_api_key_for_provider" + ) + @patch("application.api.answer.services.compression.orchestrator.LLMCreator") + @patch("application.api.answer.services.compression.orchestrator.settings") + def test_uses_compression_model_override( + self, + mock_settings, + MockLLMCreator, + mock_get_api_key, + mock_get_provider, + mock_conversation_service, + mock_threshold_checker, + decoded_token, + ): + mock_settings.COMPRESSION_MODEL_OVERRIDE = "gpt-3.5-turbo" + mock_get_provider.return_value = "openai" + mock_get_api_key.return_value = "sk-test" + MockLLMCreator.create_llm.return_value = MagicMock() + + conversation = {"queries": [{"prompt": "q", "response": "r"}], "agent_id": "a"} + + with patch( + "application.api.answer.services.compression.orchestrator.CompressionService" + ) as MockCS: + mock_svc = MagicMock() + mock_svc.compress_and_save.return_value = MagicMock( + compression_ratio=3.0, + original_token_count=300, + compressed_token_count=100, + ) + mock_svc.get_compressed_context.return_value = ("s", []) + MockCS.return_value = mock_svc + + mock_conversation_service.get_conversation.return_value = conversation + + orch = CompressionOrchestrator( + conversation_service=mock_conversation_service, + threshold_checker=mock_threshold_checker, + ) + orch._perform_compression("c1", conversation, "gpt-4", decoded_token) + + # Verify the override model was used + mock_get_provider.assert_called_with("gpt-3.5-turbo") + + @patch( + "application.api.answer.services.compression.orchestrator.get_provider_from_model_id" + ) + @patch( + "application.api.answer.services.compression.orchestrator.get_api_key_for_provider" + ) + @patch("application.api.answer.services.compression.orchestrator.LLMCreator") + @patch("application.api.answer.services.compression.orchestrator.CompressionService") + @patch("application.api.answer.services.compression.orchestrator.settings") + def test_no_queries_returns_no_compression( + self, + mock_settings, + MockCompressionService, + MockLLMCreator, + mock_get_api_key, + mock_get_provider, + mock_conversation_service, + mock_threshold_checker, + decoded_token, + ): + mock_settings.COMPRESSION_MODEL_OVERRIDE = None + mock_get_provider.return_value = "openai" + mock_get_api_key.return_value = "sk-test" + MockLLMCreator.create_llm.return_value = MagicMock() + + conversation = {"queries": [], "agent_id": "a"} + + orch = CompressionOrchestrator( + conversation_service=mock_conversation_service, + threshold_checker=mock_threshold_checker, + ) + result = orch._perform_compression("c1", conversation, "gpt-4", decoded_token) + + assert result.success is True + assert result.compression_performed is False + + def test_exception_returns_failure( + self, + mock_conversation_service, + mock_threshold_checker, + decoded_token, + ): + conversation = { + "queries": [{"prompt": "q", "response": "r"}], + "agent_id": "a", + } + + with patch( + "application.api.answer.services.compression.orchestrator.settings" + ) as mock_settings, patch( + "application.api.answer.services.compression.orchestrator.get_provider_from_model_id", + side_effect=RuntimeError("provider error"), + ): + mock_settings.COMPRESSION_MODEL_OVERRIDE = None + + orch = CompressionOrchestrator( + conversation_service=mock_conversation_service, + threshold_checker=mock_threshold_checker, + ) + result = orch._perform_compression( + "c1", conversation, "gpt-4", decoded_token + ) + + assert result.success is False + assert "provider error" in result.error + + +@pytest.mark.unit +class TestCompressMidExecution: + def test_with_provided_conversation( + self, + orchestrator, + sample_conversation, + decoded_token, + ): + with patch.object( + orchestrator, "_perform_compression" + ) as mock_perform: + mock_perform.return_value = CompressionResult.success_no_compression([]) + + orchestrator.compress_mid_execution( + conversation_id="conv1", + user_id="user1", + model_id="gpt-4", + decoded_token=decoded_token, + current_conversation=sample_conversation, + ) + + mock_perform.assert_called_once_with( + "conv1", sample_conversation, "gpt-4", decoded_token + ) + + def test_loads_conversation_when_not_provided( + self, + orchestrator, + mock_conversation_service, + sample_conversation, + decoded_token, + ): + mock_conversation_service.get_conversation.return_value = sample_conversation + + with patch.object( + orchestrator, "_perform_compression" + ) as mock_perform: + mock_perform.return_value = CompressionResult.success_no_compression([]) + + orchestrator.compress_mid_execution( + conversation_id="conv1", + user_id="user1", + model_id="gpt-4", + decoded_token=decoded_token, + ) + + mock_conversation_service.get_conversation.assert_called_once_with( + "conv1", "user1" + ) + mock_perform.assert_called_once() + + def test_conversation_not_found_returns_failure( + self, + orchestrator, + mock_conversation_service, + decoded_token, + ): + mock_conversation_service.get_conversation.return_value = None + + result = orchestrator.compress_mid_execution( + conversation_id="conv1", + user_id="user1", + model_id="gpt-4", + decoded_token=decoded_token, + ) + + assert result.success is False + assert "not found" in result.error + + def test_exception_returns_failure( + self, + orchestrator, + mock_conversation_service, + decoded_token, + ): + mock_conversation_service.get_conversation.side_effect = RuntimeError("fail") + + result = orchestrator.compress_mid_execution( + conversation_id="conv1", + user_id="user1", + model_id="gpt-4", + decoded_token=decoded_token, + ) + + assert result.success is False + assert "fail" in result.error + + +@pytest.mark.unit +class TestOrchestratorInit: + def test_default_threshold_checker(self, mock_conversation_service): + orch = CompressionOrchestrator( + conversation_service=mock_conversation_service + ) + assert orch.threshold_checker is not None + assert orch.conversation_service is mock_conversation_service + + def test_custom_threshold_checker( + self, mock_conversation_service, mock_threshold_checker + ): + orch = CompressionOrchestrator( + conversation_service=mock_conversation_service, + threshold_checker=mock_threshold_checker, + ) + assert orch.threshold_checker is mock_threshold_checker diff --git a/tests/api/answer/services/compression/test_service.py b/tests/api/answer/services/compression/test_service.py new file mode 100644 index 00000000..99042780 --- /dev/null +++ b/tests/api/answer/services/compression/test_service.py @@ -0,0 +1,423 @@ +"""Tests for application/api/answer/services/compression/service.py""" + +from unittest.mock import MagicMock, patch + +import pytest + +from application.api.answer.services.compression.service import CompressionService +from application.api.answer.services.compression.types import CompressionMetadata + + +@pytest.fixture +def mock_llm(): + llm = MagicMock() + llm.gen.return_value = "Compressed summary content" + return llm + + +@pytest.fixture +def mock_conversation_service(): + svc = MagicMock() + svc.update_compression_metadata = MagicMock() + return svc + + +@pytest.fixture +def sample_conversation(): + return { + "queries": [ + {"prompt": "What is Python?", "response": "A programming language."}, + {"prompt": "Tell me more.", "response": "It's versatile and popular."}, + { + "prompt": "What about tools?", + "response": "Python has many tools.", + "tool_calls": [ + { + "tool_name": "search", + "action_name": "web_search", + "arguments": {"q": "python tools"}, + "result": "Found 10 results", + "status": "success", + } + ], + }, + ], + "compression_metadata": {}, + } + + +@pytest.mark.unit +class TestCompressionServiceInit: + @patch("application.api.answer.services.compression.service.settings") + def test_default_prompt_builder(self, mock_settings, mock_llm): + mock_settings.COMPRESSION_PROMPT_VERSION = "v1.0" + with patch( + "application.api.answer.services.compression.service.CompressionPromptBuilder" + ): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + assert svc.llm is mock_llm + assert svc.model_id == "gpt-4" + + def test_custom_prompt_builder(self, mock_llm): + custom_builder = MagicMock() + svc = CompressionService( + llm=mock_llm, model_id="gpt-4", prompt_builder=custom_builder + ) + assert svc.prompt_builder is custom_builder + + +@pytest.mark.unit +class TestCompressConversation: + def test_successful_compression(self, mock_llm, sample_conversation): + mock_builder = MagicMock() + mock_builder.build_prompt.return_value = [ + {"role": "system", "content": "Compress"}, + {"role": "user", "content": "Conversation..."}, + ] + mock_builder.version = "v1.0" + + svc = CompressionService( + llm=mock_llm, model_id="gpt-4", prompt_builder=mock_builder + ) + + with patch( + "application.api.answer.services.compression.service.TokenCounter" + ) as MockTC: + MockTC.count_query_tokens.return_value = 1000 + MockTC.count_message_tokens.return_value = 100 + + result = svc.compress_conversation(sample_conversation, 2) + + assert isinstance(result, CompressionMetadata) + assert result.query_index == 2 + assert result.compressed_summary == "Compressed summary content" + assert result.original_token_count == 1000 + assert result.compressed_token_count == 100 + assert result.compression_ratio == 10.0 + assert result.model_used == "gpt-4" + assert result.compression_prompt_version == "v1.0" + + def test_invalid_index_negative(self, mock_llm, sample_conversation): + mock_builder = MagicMock() + mock_builder.version = "v1.0" + svc = CompressionService( + llm=mock_llm, model_id="gpt-4", prompt_builder=mock_builder + ) + + with pytest.raises(ValueError, match="Invalid compress_up_to_index"): + svc.compress_conversation(sample_conversation, -1) + + def test_invalid_index_too_large(self, mock_llm, sample_conversation): + mock_builder = MagicMock() + mock_builder.version = "v1.0" + svc = CompressionService( + llm=mock_llm, model_id="gpt-4", prompt_builder=mock_builder + ) + + with pytest.raises(ValueError, match="Invalid compress_up_to_index"): + svc.compress_conversation(sample_conversation, 10) + + def test_with_existing_compressions(self, mock_llm): + conversation = { + "queries": [ + {"prompt": "q1", "response": "r1"}, + {"prompt": "q2", "response": "r2"}, + ], + "compression_metadata": { + "compression_points": [ + { + "query_index": 0, + "compressed_summary": "Previous summary", + } + ] + }, + } + mock_builder = MagicMock() + mock_builder.build_prompt.return_value = [ + {"role": "system", "content": "Compress"}, + {"role": "user", "content": "..."}, + ] + mock_builder.version = "v1.0" + + svc = CompressionService( + llm=mock_llm, model_id="gpt-4", prompt_builder=mock_builder + ) + + with patch( + "application.api.answer.services.compression.service.TokenCounter" + ) as MockTC: + MockTC.count_query_tokens.return_value = 500 + MockTC.count_message_tokens.return_value = 50 + + result = svc.compress_conversation(conversation, 1) + assert isinstance(result, CompressionMetadata) + # Verify existing compressions were passed to prompt builder + call_args = mock_builder.build_prompt.call_args + assert call_args[0][1] == [ + {"query_index": 0, "compressed_summary": "Previous summary"} + ] + + def test_zero_compressed_tokens_ratio(self, mock_llm, sample_conversation): + mock_builder = MagicMock() + mock_builder.build_prompt.return_value = [ + {"role": "system", "content": "C"}, + {"role": "user", "content": "..."}, + ] + mock_builder.version = "v1.0" + + svc = CompressionService( + llm=mock_llm, model_id="gpt-4", prompt_builder=mock_builder + ) + + with patch( + "application.api.answer.services.compression.service.TokenCounter" + ) as MockTC: + MockTC.count_query_tokens.return_value = 1000 + MockTC.count_message_tokens.return_value = 0 + + result = svc.compress_conversation(sample_conversation, 2) + assert result.compression_ratio == 0 + + def test_llm_error_propagates(self, sample_conversation): + llm = MagicMock() + llm.gen.side_effect = RuntimeError("LLM error") + mock_builder = MagicMock() + mock_builder.build_prompt.return_value = [ + {"role": "system", "content": "C"}, + {"role": "user", "content": "..."}, + ] + mock_builder.version = "v1.0" + + svc = CompressionService( + llm=llm, model_id="gpt-4", prompt_builder=mock_builder + ) + + with patch( + "application.api.answer.services.compression.service.TokenCounter" + ) as MockTC: + MockTC.count_query_tokens.return_value = 100 + with pytest.raises(RuntimeError, match="LLM error"): + svc.compress_conversation(sample_conversation, 2) + + +@pytest.mark.unit +class TestCompressAndSave: + def test_saves_metadata_to_db( + self, mock_llm, mock_conversation_service, sample_conversation + ): + mock_builder = MagicMock() + mock_builder.build_prompt.return_value = [ + {"role": "system", "content": "C"}, + {"role": "user", "content": "..."}, + ] + mock_builder.version = "v1.0" + + svc = CompressionService( + llm=mock_llm, + model_id="gpt-4", + conversation_service=mock_conversation_service, + prompt_builder=mock_builder, + ) + + with patch( + "application.api.answer.services.compression.service.TokenCounter" + ) as MockTC: + MockTC.count_query_tokens.return_value = 500 + MockTC.count_message_tokens.return_value = 50 + + result = svc.compress_and_save("conv_123", sample_conversation, 2) + + assert isinstance(result, CompressionMetadata) + mock_conversation_service.update_compression_metadata.assert_called_once_with( + "conv_123", result.to_dict() + ) + + def test_raises_without_conversation_service(self, mock_llm, sample_conversation): + mock_builder = MagicMock() + mock_builder.version = "v1.0" + svc = CompressionService( + llm=mock_llm, model_id="gpt-4", prompt_builder=mock_builder + ) + + with pytest.raises(ValueError, match="conversation_service required"): + svc.compress_and_save("conv_123", sample_conversation, 2) + + +@pytest.mark.unit +class TestGetCompressedContext: + def test_no_compression_returns_full_history(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + conversation = { + "queries": [{"prompt": "q1", "response": "r1"}], + "compression_metadata": {}, + } + + summary, queries = svc.get_compressed_context(conversation) + + assert summary is None + assert queries == [{"prompt": "q1", "response": "r1"}] + + def test_no_compression_points_returns_full_history(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + conversation = { + "queries": [{"prompt": "q1", "response": "r1"}], + "compression_metadata": {"is_compressed": True, "compression_points": []}, + } + + summary, queries = svc.get_compressed_context(conversation) + assert summary is None + assert len(queries) == 1 + + def test_with_compression_returns_summary_and_recent(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + conversation = { + "queries": [ + {"prompt": "q0", "response": "r0"}, + {"prompt": "q1", "response": "r1"}, + {"prompt": "q2", "response": "r2"}, + ], + "compression_metadata": { + "is_compressed": True, + "compression_points": [ + { + "query_index": 1, + "compressed_summary": "Summary of q0 and q1", + "compressed_token_count": 50, + "original_token_count": 500, + } + ], + }, + } + + summary, queries = svc.get_compressed_context(conversation) + + assert summary == "Summary of q0 and q1" + assert len(queries) == 1 + assert queries[0]["prompt"] == "q2" + + def test_none_queries_returns_empty(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + conversation = { + "queries": None, + "compression_metadata": {}, + } + + summary, queries = svc.get_compressed_context(conversation) + assert summary is None + assert queries == [] + + def test_exception_falls_back_to_full_history(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + conversation = { + "queries": [{"prompt": "q", "response": "r"}], + "compression_metadata": { + "is_compressed": True, + "compression_points": "invalid", # This will cause an error + }, + } + + summary, queries = svc.get_compressed_context(conversation) + assert summary is None + assert queries == [{"prompt": "q", "response": "r"}] + + def test_exception_with_none_queries_returns_empty(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + # Force exception by making compression_points non-iterable + conversation = { + "queries": None, + "compression_metadata": { + "is_compressed": True, + "compression_points": "bad", + }, + } + + summary, queries = svc.get_compressed_context(conversation) + assert summary is None + assert queries == [] + + +@pytest.mark.unit +class TestExtractSummary: + def test_extracts_from_summary_tags(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + response = "Some analysisThe actual summary" + result = svc._extract_summary(response) + assert result == "The actual summary" + + def test_removes_analysis_tags_when_no_summary(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + response = "analysis textRaw summary text here" + result = svc._extract_summary(response) + assert result == "Raw summary text here" + + def test_returns_full_response_when_no_tags(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + response = "Just a plain text response" + result = svc._extract_summary(response) + assert result == "Just a plain text response" + + def test_multiline_summary(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + response = "Line 1\nLine 2\nLine 3" + result = svc._extract_summary(response) + assert "Line 1" in result + assert "Line 3" in result + + def test_strips_whitespace(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + response = " Trimmed " + result = svc._extract_summary(response) + assert result == "Trimmed" + + +@pytest.mark.unit +class TestLogToolCallStats: + def test_no_tool_calls(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + queries = [{"prompt": "q", "response": "r"}] + # Should not raise + svc._log_tool_call_stats(queries) + + def test_with_tool_calls(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + queries = [ + { + "prompt": "q", + "response": "r", + "tool_calls": [ + { + "tool_name": "search", + "action_name": "web", + "result": "result text", + }, + { + "tool_name": "search", + "action_name": "web", + "result": "more text", + }, + ], + } + ] + # Should not raise - just logs + svc._log_tool_call_stats(queries) + + def test_empty_queries(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + svc._log_tool_call_stats([]) + + def test_tool_call_with_none_result(self, mock_llm): + svc = CompressionService(llm=mock_llm, model_id="gpt-4") + queries = [ + { + "prompt": "q", + "response": "r", + "tool_calls": [ + { + "tool_name": "t", + "action_name": "a", + "result": None, + } + ], + } + ] + svc._log_tool_call_stats(queries) diff --git a/tests/api/answer/services/compression/test_types.py b/tests/api/answer/services/compression/test_types.py new file mode 100644 index 00000000..ef7a2396 --- /dev/null +++ b/tests/api/answer/services/compression/test_types.py @@ -0,0 +1,131 @@ +"""Tests for application/api/answer/services/compression/types.py""" + +from datetime import datetime, timezone + +import pytest + +from application.api.answer.services.compression.types import ( + CompressionMetadata, + CompressionResult, +) + + +@pytest.mark.unit +class TestCompressionMetadata: + def _make_metadata(self, **overrides): + defaults = dict( + timestamp=datetime(2025, 1, 1, tzinfo=timezone.utc), + query_index=5, + compressed_summary="Summary of conversation", + original_token_count=5000, + compressed_token_count=500, + compression_ratio=10.0, + model_used="gpt-4", + compression_prompt_version="v1.0", + ) + defaults.update(overrides) + return CompressionMetadata(**defaults) + + def test_to_dict_contains_all_fields(self): + meta = self._make_metadata() + d = meta.to_dict() + + assert d["timestamp"] == datetime(2025, 1, 1, tzinfo=timezone.utc) + assert d["query_index"] == 5 + assert d["compressed_summary"] == "Summary of conversation" + assert d["original_token_count"] == 5000 + assert d["compressed_token_count"] == 500 + assert d["compression_ratio"] == 10.0 + assert d["model_used"] == "gpt-4" + assert d["compression_prompt_version"] == "v1.0" + + def test_to_dict_returns_dict_type(self): + meta = self._make_metadata() + assert isinstance(meta.to_dict(), dict) + + def test_to_dict_field_count(self): + meta = self._make_metadata() + d = meta.to_dict() + assert len(d) == 8 + + def test_attributes_accessible(self): + meta = self._make_metadata(query_index=10, compression_ratio=5.5) + assert meta.query_index == 10 + assert meta.compression_ratio == 5.5 + + def test_zero_compressed_tokens(self): + meta = self._make_metadata(compressed_token_count=0, compression_ratio=0) + d = meta.to_dict() + assert d["compressed_token_count"] == 0 + assert d["compression_ratio"] == 0 + + +@pytest.mark.unit +class TestCompressionResult: + def test_success_with_compression(self): + meta = CompressionMetadata( + timestamp=datetime.now(timezone.utc), + query_index=3, + compressed_summary="summary", + original_token_count=1000, + compressed_token_count=100, + compression_ratio=10.0, + model_used="gpt-4", + compression_prompt_version="v1.0", + ) + queries = [{"prompt": "q1", "response": "r1"}] + result = CompressionResult.success_with_compression("summary", queries, meta) + + assert result.success is True + assert result.compressed_summary == "summary" + assert result.recent_queries == queries + assert result.metadata is meta + assert result.compression_performed is True + assert result.error is None + + def test_success_no_compression(self): + queries = [{"prompt": "q1", "response": "r1"}] + result = CompressionResult.success_no_compression(queries) + + assert result.success is True + assert result.compressed_summary is None + assert result.recent_queries == queries + assert result.metadata is None + assert result.compression_performed is False + assert result.error is None + + def test_failure(self): + result = CompressionResult.failure("something went wrong") + + assert result.success is False + assert result.error == "something went wrong" + assert result.compression_performed is False + assert result.compressed_summary is None + assert result.recent_queries == [] + assert result.metadata is None + + def test_as_history_extracts_prompt_response(self): + queries = [ + {"prompt": "Hello", "response": "Hi", "extra": "ignored"}, + {"prompt": "How?", "response": "Fine"}, + ] + result = CompressionResult.success_no_compression(queries) + history = result.as_history() + + assert len(history) == 2 + assert history[0] == {"prompt": "Hello", "response": "Hi"} + assert history[1] == {"prompt": "How?", "response": "Fine"} + + def test_as_history_empty_queries(self): + result = CompressionResult.success_no_compression([]) + assert result.as_history() == [] + + def test_default_recent_queries_is_empty_list(self): + result = CompressionResult(success=True) + assert result.recent_queries == [] + assert result.as_history() == [] + + def test_success_no_compression_with_empty_list(self): + result = CompressionResult.success_no_compression([]) + assert result.success is True + assert result.recent_queries == []