From efbe121654c0b2d71ccda5aba2b74ebe11eaa8e1 Mon Sep 17 00:00:00 2001 From: Krrish Ghimire Date: Thu, 16 Apr 2026 11:30:46 +0545 Subject: [PATCH] test integrations --- backend/core/tests/test_custom_tool_parser.py | 434 ++++++++++++ backend/core/tests/test_github_tools.py | 667 ++++++++++++++++++ backend/core/tests/test_github_validator.py | 205 ++++++ .../core/tests/test_integration_dispatcher.py | 366 ++++++++++ 4 files changed, 1672 insertions(+) create mode 100644 backend/core/tests/test_custom_tool_parser.py create mode 100644 backend/core/tests/test_github_tools.py create mode 100644 backend/core/tests/test_github_validator.py create mode 100644 backend/core/tests/test_integration_dispatcher.py diff --git a/backend/core/tests/test_custom_tool_parser.py b/backend/core/tests/test_custom_tool_parser.py new file mode 100644 index 0000000..d1137a9 --- /dev/null +++ b/backend/core/tests/test_custom_tool_parser.py @@ -0,0 +1,434 @@ +import pytest + +from core.integrations.custom_tool_parser import ( + _derive_name, + _infer_type, + _minimal_schema, + parse_url_schema, + pretty_print_schema, +) + + +@pytest.mark.unit +class TestDeriveName: + def test_derive_name_simple(self): + result = _derive_name("Get Weather") + assert result == "get_weather" + + def test_derive_name_multiple_spaces(self): + result = _derive_name("Get User Profile") + assert result == "get_user_profile" + + def test_derive_name_special_characters(self): + result = _derive_name("Get User@Profile!") + assert result == "get_userprofile" + + def test_derive_name_numbers(self): + result = _derive_name("API v2 Request") + assert result == "api_v2_request" + + def test_derive_name_empty_string(self): + result = _derive_name("") + assert result == "custom_tool" + + def test_derive_name_only_special_chars(self): + result = _derive_name("@#$%^&*()") + assert result == "custom_tool" + + def test_derive_name_leading_trailing_spaces(self): + result = _derive_name(" Get Weather ") + assert result == "get_weather" + + def test_derive_name_underscores_in_title(self): + result = _derive_name("Get_User_Profile") + assert result == "get_user_profile" + + def test_derive_name_multiple_underscores(self): + result = _derive_name("Get___User___Profile") + assert result == "get_user_profile" + + +@pytest.mark.unit +class TestInferType: + def test_infer_type_string(self): + """Test inferring type from string.""" + result = _infer_type("hello") + assert result == "string" + + def test_infer_type_int(self): + result = _infer_type(42) + assert result == "number" + + def test_infer_type_float(self): + result = _infer_type(3.14) + assert result == "number" + + def test_infer_type_bool(self): + result = _infer_type(True) + assert result == "boolean" + + def test_infer_type_list(self): + result = _infer_type([1, 2, 3]) + assert result == "array" + + def test_infer_type_dict(self): + result = _infer_type({"key": "value"}) + assert result == "object" + + def test_infer_type_none(self): + result = _infer_type(None) + assert result == "string" + + def test_infer_type_unknown_type(self): + class CustomType: + pass + result = _infer_type(CustomType()) + assert result == "string" + + +@pytest.mark.unit +class TestMinimalSchema: + def test_minimal_schema_basic(self): + result = _minimal_schema("get_weather", "Get current weather") + assert result == { + "type": "function", + "function": { + "name": "get_weather", + "description": "Get current weather", + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + } + + def test_minimal_schema_empty_description(self): + result = _minimal_schema("tool_name", "") + assert result["function"]["description"] == "" + + def test_minimal_schema_special_chars_in_name(self): + result = _minimal_schema("tool-name", "description") + assert result["function"]["name"] == "tool-name" + + +@pytest.mark.unit +class TestParseUrlSchema: + def test_parse_url_schema_with_single_quotes(self): + url_schema = """curl -X POST https://api.example.com/endpoint --data '{"name": "John", "age": 30}'""" + result = parse_url_schema("Get User", "Get user information", url_schema) + + assert result["type"] == "function" + assert result["function"]["name"] == "get_user" + assert result["function"]["description"] == "Get user information" + assert "name" in result["function"]["parameters"]["properties"] + assert "age" in result["function"]["parameters"]["properties"] + assert result["function"]["parameters"]["properties"]["name"]["type"] == "string" + assert result["function"]["parameters"]["properties"]["age"]["type"] == "number" + + def test_parse_url_schema_with_double_quotes(self): + url_schema = '''curl -X POST https://api.example.com/endpoint --data "{\"city\": \"NYC\", \"temp\": 72.5}"''' + result = parse_url_schema("Get Weather", "Get weather data", url_schema) + + assert result["function"]["name"] == "get_weather" + assert result["function"]["description"] == "Get weather data" + + def test_parse_url_schema_with_unquoted_braces(self): + url_schema = """curl -X POST https://api.example.com/endpoint --data {"product": "widget", "quantity": 10}""" + result = parse_url_schema("Create Order", "Create a new order", url_schema) + + assert result["function"]["name"] == "create_order" + assert "product" in result["function"]["parameters"]["properties"] + assert "quantity" in result["function"]["parameters"]["properties"] + + def test_parse_url_schema_with_array_value(self): + url_schema = """curl -X POST https://api.example.com/endpoint --data '{"tags": ["tag1", "tag2"]}'""" + result = parse_url_schema("Add Tags", "Add tags to item", url_schema) + + assert result["function"]["parameters"]["properties"]["tags"]["type"] == "array" + assert "items" in result["function"]["parameters"]["properties"]["tags"] + + def test_parse_url_schema_with_object_value(self): + url_schema = """curl -X POST https://api.example.com/endpoint --data '{"metadata": {"key": "value"}}'""" + result = parse_url_schema("Set Metadata", "Set item metadata", url_schema) + + assert result["function"]["parameters"]["properties"]["metadata"]["type"] == "object" + + def test_parse_url_schema_with_boolean_value(self): + url_schema = """curl -X POST https://api.example.com/endpoint --data '{"active": true, "deleted": false}'""" + result = parse_url_schema("Update Status", "Update item status", url_schema) + + assert result["function"]["parameters"]["properties"]["active"]["type"] == "boolean" + assert result["function"]["parameters"]["properties"]["deleted"]["type"] == "boolean" + + def test_parse_url_schema_with_null_value(self): + url_schema = """curl -X POST https://api.example.com/endpoint --data '{"optional": null}'""" + result = parse_url_schema("Optional Field", "Test optional field", url_schema) + + assert result["function"]["parameters"]["properties"]["optional"]["type"] == "string" + + def test_parse_url_schema_no_data_flag(self): + url_schema = """curl -X POST https://api.example.com/endpoint""" + result = parse_url_schema("Simple Tool", "A simple tool", url_schema) + + assert result["function"]["name"] == "simple_tool" + assert result["function"]["parameters"]["properties"] == {} + + def test_parse_url_schema_invalid_json(self): + url_schema = """curl -X POST https://api.example.com/endpoint --data '{invalid json}'""" + result = parse_url_schema("Invalid Tool", "Tool with invalid JSON", url_schema) + + assert result["function"]["name"] == "invalid_tool" + assert result["function"]["parameters"]["properties"] == {} + + def test_parse_url_schema_json_not_dict(self): + url_schema = """curl -X POST https://api.example.com/endpoint --data '["item1", "item2"]'""" + result = parse_url_schema("Array Tool", "Tool with array", url_schema) + + assert result["function"]["name"] == "array_tool" + assert result["function"]["parameters"]["properties"] == {} + + def test_parse_url_schema_multiline_json(self): + url_schema = """curl -X POST https://api.example.com/endpoint --data '{ + "name": "John", + "age": 30 + }'""" + result = parse_url_schema("Multiline Tool", "Tool with multiline JSON", url_schema) + + assert result["function"]["name"] == "multiline_tool" + assert "name" in result["function"]["parameters"]["properties"] + assert "age" in result["function"]["parameters"]["properties"] + + def test_parse_url_schema_escaped_quotes(self): + url_schema = r"""curl -X POST https://api.example.com/endpoint --data '{"message": "Hello \"World\""}'""" + result = parse_url_schema("Escaped Tool", "Tool with escaped quotes", url_schema) + + assert result["function"]["name"] == "escaped_tool" + assert "message" in result["function"]["parameters"]["properties"] + + def test_parse_url_schema_empty_body(self): + url_schema = """curl -X POST https://api.example.com/endpoint --data '{}'""" + result = parse_url_schema("Empty Tool", "Tool with empty body", url_schema) + + assert result["function"]["name"] == "empty_tool" + assert result["function"]["parameters"]["properties"] == {} + + def test_parse_url_schema_complex_nested_structure(self): + url_schema = """curl -X POST https://api.example.com/endpoint --data '{"user": {"name": "John", "roles": ["admin", "user"]}, "active": true}'""" + result = parse_url_schema("Complex Tool", "Tool with complex structure", url_schema) + + assert result["function"]["name"] == "complex_tool" + assert "user" in result["function"]["parameters"]["properties"] + assert result["function"]["parameters"]["properties"]["user"]["type"] == "object" + assert "active" in result["function"]["parameters"]["properties"] + assert result["function"]["parameters"]["properties"]["active"]["type"] == "boolean" + + +@pytest.mark.unit +class TestPrettyPrintSchema: + def test_pretty_print_schema_no_properties(self): + schema = { + "type": "function", + "function": { + "name": "get_weather", + "description": "Get weather data", + "parameters": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + } + result = pretty_print_schema(schema) + assert result == "curl -X GET https://api.example.com/get_weather" + + def test_pretty_print_schema_with_string_property(self): + schema = { + "type": "function", + "function": { + "name": "create_user", + "description": "Create a user", + "parameters": { + "type": "object", + "properties": { + "name": {"type": "string", "description": "User name"}, + }, + "required": [], + }, + }, + } + result = pretty_print_schema(schema) + assert "curl -X POST https://api.example.com/create_user" in result + assert '"Content-Type: application/json"' in result + assert "-d" in result + assert '"name": "example"' in result + + def test_pretty_print_schema_with_number_property(self): + schema = { + "type": "function", + "function": { + "name": "set_age", + "description": "Set user age", + "parameters": { + "type": "object", + "properties": { + "age": {"type": "number", "description": "User age"}, + }, + "required": [], + }, + }, + } + result = pretty_print_schema(schema) + assert '"age": 0' in result + + def test_pretty_print_schema_with_boolean_property(self): + schema = { + "type": "function", + "function": { + "name": "toggle_active", + "description": "Toggle active status", + "parameters": { + "type": "object", + "properties": { + "active": {"type": "boolean", "description": "Active status"}, + }, + "required": [], + }, + }, + } + result = pretty_print_schema(schema) + assert '"active": true' in result + + def test_pretty_print_schema_with_array_property(self): + schema = { + "type": "function", + "function": { + "name": "add_tags", + "description": "Add tags", + "parameters": { + "type": "object", + "properties": { + "tags": {"type": "array", "description": "Tags list"}, + }, + "required": [], + }, + }, + } + result = pretty_print_schema(schema) + assert '"tags": []' in result + + def test_pretty_print_schema_with_object_property(self): + schema = { + "type": "function", + "function": { + "name": "set_metadata", + "description": "Set metadata", + "parameters": { + "type": "object", + "properties": { + "metadata": {"type": "object", "description": "Metadata object"}, + }, + "required": [], + }, + }, + } + result = pretty_print_schema(schema) + assert '"metadata": {}' in result + + def test_pretty_print_schema_multiple_properties(self): + schema = { + "type": "function", + "function": { + "name": "create_item", + "description": "Create an item", + "parameters": { + "type": "object", + "properties": { + "name": {"type": "string", "description": "Item name"}, + "count": {"type": "number", "description": "Item count"}, + "active": {"type": "boolean", "description": "Active status"}, + }, + "required": [], + }, + }, + } + result = pretty_print_schema(schema) + assert '"name": "example"' in result + assert '"count": 0' in result + assert '"active": true' in result + + def test_pretty_print_schema_missing_function_key(self): + schema = { + "type": "function", + } + result = pretty_print_schema(schema) + assert result == "curl -X GET https://api.example.com/tool" + + def test_pretty_print_schema_missing_name(self): + schema = { + "type": "function", + "function": { + "description": "A tool", + }, + } + result = pretty_print_schema(schema) + assert result == "curl -X GET https://api.example.com/tool" + + def test_pretty_print_schema_missing_parameters(self): + schema = { + "type": "function", + "function": { + "name": "my_tool", + "description": "My tool", + }, + } + result = pretty_print_schema(schema) + assert result == "curl -X GET https://api.example.com/my_tool" + + def test_pretty_print_schema_parameters_not_dict(self): + schema = { + "type": "function", + "function": { + "name": "my_tool", + "description": "My tool", + "parameters": "invalid", + }, + } + result = pretty_print_schema(schema) + assert result == "curl -X GET https://api.example.com/my_tool" + + def test_pretty_print_schema_property_not_dict(self): + schema = { + "type": "function", + "function": { + "name": "my_tool", + "description": "My tool", + "parameters": { + "type": "object", + "properties": { + "field": "not a dict", + }, + "required": [], + }, + }, + } + result = pretty_print_schema(schema) + assert '"field": "example"' in result + + def test_pretty_print_schema_unknown_type(self): + schema = { + "type": "function", + "function": { + "name": "my_tool", + "description": "My tool", + "parameters": { + "type": "object", + "properties": { + "field": {"type": "unknown_type", "description": "A field"}, + }, + "required": [], + }, + }, + } + result = pretty_print_schema(schema) + assert '"field": "example"' in result diff --git a/backend/core/tests/test_github_tools.py b/backend/core/tests/test_github_tools.py new file mode 100644 index 0000000..708b9ce --- /dev/null +++ b/backend/core/tests/test_github_tools.py @@ -0,0 +1,667 @@ +import pytest +from unittest.mock import Mock, patch, MagicMock + +from core.integrations.github_tools import ( + _headers, + _creds, + _gh_request, + list_commits, + list_pull_requests, + list_releases, + list_tickets, + create_ticket, + get_ticket, + update_ticket, + lock_ticket, + unlock_ticket, + GITHUB_API, + GITHUB_API_VERSION, + GITHUB_VC_HANDLERS, + GITHUB_PM_HANDLERS, +) + + +@pytest.mark.unit +class TestHeaders: + def test_headers_basic(self): + result = _headers("test_token_123") + assert result["Authorization"] == "Bearer test_token_123" + assert result["Accept"] == "application/vnd.github+json" + assert result["X-GitHub-Api-Version"] == GITHUB_API_VERSION + + def test_headers_empty_token(self): + result = _headers("") + assert result["Authorization"] == "Bearer " + assert result["Accept"] == "application/vnd.github+json" + + +@pytest.mark.unit +class TestCreds: + def test_creds_basic(self): + app_integration = Mock() + app_integration.integration.credentials = '{"token": "ghp_test_token"}' + app_integration.metadata = {"repo": "owner/repo"} + + token, repo = _creds(app_integration) + assert token == "ghp_test_token" + assert repo == "owner/repo" + + def test_creds_no_metadata(self): + app_integration = Mock() + app_integration.integration.credentials = '{"token": "ghp_test_token"}' + app_integration.metadata = None + + token, repo = _creds(app_integration) + assert token == "ghp_test_token" + assert repo == "" + + def test_creds_empty_metadata(self): + app_integration = Mock() + app_integration.integration.credentials = '{"token": "ghp_test_token"}' + app_integration.metadata = {} + + token, repo = _creds(app_integration) + assert token == "ghp_test_token" + assert repo == "" + + def test_creds_no_repo_in_metadata(self): + app_integration = Mock() + app_integration.integration.credentials = '{"token": "ghp_test_token"}' + app_integration.metadata = {"other_field": "value"} + + token, repo = _creds(app_integration) + assert token == "ghp_test_token" + assert repo == "" + + +@pytest.mark.unit +class TestGhRequest: + @patch("core.integrations.github_tools.requests.request") + def test_gh_request_get(self, mock_request): + mock_response = Mock() + mock_response.status_code = 200 + mock_response.headers = {"content-length": "100"} + mock_response.json.return_value = {"data": "test"} + mock_request.return_value = mock_response + + headers = {"Authorization": "Bearer token"} + result = _gh_request("get", f"{GITHUB_API}/test", headers=headers) + + mock_request.assert_called_once_with("get", f"{GITHUB_API}/test", headers=headers, params=None, json=None, timeout=15) + assert result.status_code == 200 + + @patch("core.integrations.github_tools.requests.request") + def test_gh_request_with_params(self, mock_request): + mock_response = Mock() + mock_response.status_code = 200 + mock_response.headers = {"content-length": "50"} + mock_response.json.return_value = [] + mock_request.return_value = mock_response + + headers = {"Authorization": "Bearer token"} + params = {"state": "open", "per_page": 10} + result = _gh_request("get", f"{GITHUB_API}/test", headers=headers, params=params) + + mock_request.assert_called_once_with("get", f"{GITHUB_API}/test", headers=headers, params=params, json=None, timeout=15) + + @patch("core.integrations.github_tools.requests.request") + def test_gh_request_post_with_json(self, mock_request): + mock_response = Mock() + mock_response.status_code = 201 + mock_response.headers = {"content-length": "200"} + mock_response.json.return_value = {"id": 1} + mock_request.return_value = mock_response + + headers = {"Authorization": "Bearer token"} + json_data = {"title": "Test Issue"} + result = _gh_request("post", f"{GITHUB_API}/test", headers=headers, json=json_data) + + mock_request.assert_called_once_with("post", f"{GITHUB_API}/test", headers=headers, params=None, json=json_data, timeout=15) + + @patch("core.integrations.github_tools.requests.request") + def test_gh_request_raises_on_error(self, mock_request): + mock_response = Mock() + mock_response.status_code = 404 + mock_response.headers = {} + mock_response.raise_for_status.side_effect = Exception("404 Not Found") + mock_request.return_value = mock_response + + headers = {"Authorization": "Bearer token"} + + with pytest.raises(Exception): + _gh_request("get", f"{GITHUB_API}/test", headers=headers) + + +@pytest.mark.unit +class TestListCommits: + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_list_commits_basic(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = [ + { + "sha": "abc123def456", + "commit": { + "message": "First commit\n\nSome details", + "author": {"name": "John Doe", "date": "2024-01-01T00:00:00Z"} + } + }, + { + "sha": "def456ghi789", + "commit": { + "message": "Second commit", + "author": {"name": "Jane Smith", "date": "2024-01-02T00:00:00Z"} + } + } + ] + mock_request.return_value = mock_response + + app_integration = Mock() + result = list_commits(app_integration) + + assert len(result) == 2 + assert result[0]["sha"] == "abc123d" + assert result[0]["message"] == "First commit" + assert result[0]["author"] == "John Doe" + assert result[1]["sha"] == "def456g" + assert result[1]["message"] == "Second commit" + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_list_commits_with_params(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = [ + { + "sha": "abc123def456", + "commit": { + "message": "Feature commit", + "author": {"name": "John Doe", "date": "2024-01-01T00:00:00Z"} + } + } + ] + mock_request.return_value = mock_response + + app_integration = Mock() + result = list_commits(app_integration, sha="main", path="src/", author="john") + + mock_request.assert_called_once() + call_args = mock_request.call_args + assert "sha" in call_args[1]["params"] + assert call_args[1]["params"]["sha"] == "main" + assert call_args[1]["params"]["path"] == "src/" + assert call_args[1]["params"]["author"] == "john" + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_list_commits_filters_none_params(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = [] + mock_request.return_value = mock_response + + app_integration = Mock() + list_commits(app_integration, sha="main", path=None, author=None) + + call_args = mock_request.call_args + assert call_args[1]["params"]["sha"] == "main" + assert "path" not in call_args[1]["params"] + assert "author" not in call_args[1]["params"] + + +@pytest.mark.unit +class TestListPullRequests: + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_list_pull_requests_basic(self, mock_creds, mock_request): + """Test listing pull requests with default state.""" + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = [ + { + "number": 1, + "title": "Add new feature", + "state": "open", + "user": {"login": "john"}, + "created_at": "2024-01-01T00:00:00Z" + } + ] + mock_request.return_value = mock_response + + app_integration = Mock() + result = list_pull_requests(app_integration) + + assert len(result) == 1 + assert result[0]["number"] == 1 + assert result[0]["title"] == "Add new feature" + assert result[0]["state"] == "open" + assert result[0]["author"] == "john" + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_list_pull_requests_default_state(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = [] + mock_request.return_value = mock_response + + app_integration = Mock() + list_pull_requests(app_integration) + + call_args = mock_request.call_args + assert call_args[1]["params"]["state"] == "open" + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_list_pull_requests_with_params(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = [] + mock_request.return_value = mock_response + + app_integration = Mock() + list_pull_requests(app_integration, state="closed", base="main", sort="updated") + + call_args = mock_request.call_args + assert call_args[1]["params"]["state"] == "closed" + assert call_args[1]["params"]["base"] == "main" + assert call_args[1]["params"]["sort"] == "updated" + + +@pytest.mark.unit +class TestListReleases: + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_list_releases_basic(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = [ + { + "tag_name": "v1.0.0", + "name": "First Release", + "draft": False, + "prerelease": False, + "published_at": "2024-01-01T00:00:00Z" + }, + { + "tag_name": "v2.0.0-beta", + "name": "Beta Release", + "draft": False, + "prerelease": True, + "published_at": "2024-02-01T00:00:00Z" + } + ] + mock_request.return_value = mock_response + + app_integration = Mock() + result = list_releases(app_integration) + + assert len(result) == 2 + assert result[0]["tag"] == "v1.0.0" + assert result[0]["name"] == "First Release" + assert result[0]["draft"] is False + assert result[0]["prerelease"] is False + assert result[1]["tag"] == "v2.0.0-beta" + assert result[1]["prerelease"] is True + + +@pytest.mark.unit +class TestListTickets: + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_list_tickets_basic(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = [ + { + "number": 1, + "title": "Bug report", + "state": "open", + "labels": [{"name": "bug"}, {"name": "high-priority"}], + "created_at": "2024-01-01T00:00:00Z" + }, + { + "number": 2, + "title": "Feature request", + "state": "open", + "labels": [{"name": "enhancement"}], + "created_at": "2024-01-02T00:00:00Z" + } + ] + mock_request.return_value = mock_response + + app_integration = Mock() + result = list_tickets(app_integration) + + assert len(result) == 2 + assert result[0]["number"] == 1 + assert result[0]["title"] == "Bug report" + assert result[0]["labels"] == ["bug", "high-priority"] + assert result[1]["labels"] == ["enhancement"] + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_list_tickets_filters_pull_requests(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = [ + { + "number": 1, + "title": "Regular issue", + "state": "open", + "labels": [], + "created_at": "2024-01-01T00:00:00Z" + }, + { + "number": 2, + "title": "Pull request", + "state": "open", + "labels": [], + "created_at": "2024-01-02T00:00:00Z", + "pull_request": {"url": "https://api.github.com/repos/owner/repo/pulls/2"} + } + ] + mock_request.return_value = mock_response + + app_integration = Mock() + result = list_tickets(app_integration) + + assert len(result) == 1 + assert result[0]["number"] == 1 + assert result[0]["title"] == "Regular issue" + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_list_tickets_default_state(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = [] + mock_request.return_value = mock_response + + app_integration = Mock() + list_tickets(app_integration) + + call_args = mock_request.call_args + assert call_args[1]["params"]["state"] == "open" + + +@pytest.mark.unit +class TestCreateTicket: + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_create_ticket_basic(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = { + "number": 1, + "title": "New Issue", + "state": "open" + } + mock_request.return_value = mock_response + + app_integration = Mock() + result = create_ticket(app_integration, title="New Issue") + + assert result["number"] == 1 + assert result["title"] == "New Issue" + assert result["state"] == "open" + + call_args = mock_request.call_args + assert call_args[0][0] == "post" + assert call_args[0][1] == f"{GITHUB_API}/repos/owner/repo/issues" + assert call_args[1]["json"]["title"] == "New Issue" + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_create_ticket_with_optional_fields(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = { + "number": 1, + "title": "New Issue", + "state": "open" + } + mock_request.return_value = mock_response + + app_integration = Mock() + result = create_ticket( + app_integration, + title="New Issue", + body="Issue description", + labels=["bug", "high-priority"], + assignees=["john"] + ) + + call_args = mock_request.call_args + assert call_args[1]["json"]["body"] == "Issue description" + assert call_args[1]["json"]["labels"] == ["bug", "high-priority"] + assert call_args[1]["json"]["assignees"] == ["john"] + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_create_ticket_ignores_none_fields(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = { + "number": 1, + "title": "New Issue", + "state": "open" + } + mock_request.return_value = mock_response + + app_integration = Mock() + create_ticket( + app_integration, + title="New Issue", + body=None, + labels=None, + assignees=None + ) + + call_args = mock_request.call_args + assert "body" not in call_args[1]["json"] + assert "labels" not in call_args[1]["json"] + assert "assignees" not in call_args[1]["json"] + + +@pytest.mark.unit +class TestGetTicket: + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_get_ticket_basic(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = { + "number": 1, + "title": "Issue Title", + "state": "open", + "body": "Issue description", + "labels": [{"name": "bug"}], + "assignees": [{"login": "john"}] + } + mock_request.return_value = mock_response + + app_integration = Mock() + result = get_ticket(app_integration, issue_number=1) + + assert result["number"] == 1 + assert result["title"] == "Issue Title" + assert result["state"] == "open" + assert result["body"] == "Issue description" + assert result["labels"] == ["bug"] + assert result["assignees"] == ["john"] + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_get_ticket_empty_body(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = { + "number": 1, + "title": "Issue Title", + "state": "open", + "labels": [], + "assignees": [] + } + mock_request.return_value = mock_response + + app_integration = Mock() + result = get_ticket(app_integration, issue_number=1) + + assert result["body"] == "" + assert result["labels"] == [] + assert result["assignees"] == [] + + +@pytest.mark.unit +class TestUpdateTicket: + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_update_ticket_basic(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = { + "number": 1, + "title": "Updated Title", + "state": "closed" + } + mock_request.return_value = mock_response + + app_integration = Mock() + result = update_ticket(app_integration, issue_number=1, title="Updated Title", state="closed") + + assert result["number"] == 1 + assert result["title"] == "Updated Title" + assert result["state"] == "closed" + + call_args = mock_request.call_args + assert call_args[0][0] == "patch" + assert call_args[0][1] == f"{GITHUB_API}/repos/owner/repo/issues/1" + assert call_args[1]["json"]["title"] == "Updated Title" + assert call_args[1]["json"]["state"] == "closed" + assert "issue_number" not in call_args[1]["json"] + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_update_ticket_filters_invalid_fields(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = { + "number": 1, + "title": "Updated Title", + "state": "open" + } + mock_request.return_value = mock_response + + app_integration = Mock() + update_ticket( + app_integration, + issue_number=1, + title="Updated Title", + body="New description", + invalid_field="should not be included" + ) + + call_args = mock_request.call_args + assert "title" in call_args[1]["json"] + assert "body" in call_args[1]["json"] + assert "invalid_field" not in call_args[1]["json"] + + +@pytest.mark.unit +class TestLockTicket: + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_lock_ticket_basic(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = {} + mock_request.return_value = mock_response + + app_integration = Mock() + result = lock_ticket(app_integration, issue_number=1) + + assert result["locked"] is True + assert result["issue_number"] == 1 + + call_args = mock_request.call_args + assert call_args[0][0] == "put" + assert call_args[0][1] == f"{GITHUB_API}/repos/owner/repo/issues/1/lock" + assert call_args[1]["json"] == {} + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_lock_ticket_with_reason(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = {} + mock_request.return_value = mock_response + + app_integration = Mock() + result = lock_ticket(app_integration, issue_number=1, lock_reason="spam") + + assert result["locked"] is True + + call_args = mock_request.call_args + assert call_args[1]["json"]["lock_reason"] == "spam" + + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_lock_ticket_without_reason(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = {} + mock_request.return_value = mock_response + + app_integration = Mock() + lock_ticket(app_integration, issue_number=1, lock_reason=None) + + call_args = mock_request.call_args + assert "lock_reason" not in call_args[1]["json"] + + +@pytest.mark.unit +class TestUnlockTicket: + @patch("core.integrations.github_tools._gh_request") + @patch("core.integrations.github_tools._creds") + def test_unlock_ticket(self, mock_creds, mock_request): + mock_creds.return_value = ("ghp_token", "owner/repo") + mock_response = Mock() + mock_response.json.return_value = {} + mock_request.return_value = mock_response + + app_integration = Mock() + result = unlock_ticket(app_integration, issue_number=1) + + assert result["locked"] is False + assert result["issue_number"] == 1 + + call_args = mock_request.call_args + assert call_args[0][0] == "delete" + assert call_args[0][1] == f"{GITHUB_API}/repos/owner/repo/issues/1/lock" + + +@pytest.mark.unit +class TestHandlerDictionaries: + def test_github_vc_handlers(self): + assert "list_commits" in GITHUB_VC_HANDLERS + assert "list_pull_requests" in GITHUB_VC_HANDLERS + assert "list_releases" in GITHUB_VC_HANDLERS + assert GITHUB_VC_HANDLERS["list_commits"] == list_commits + assert GITHUB_VC_HANDLERS["list_pull_requests"] == list_pull_requests + assert GITHUB_VC_HANDLERS["list_releases"] == list_releases + + def test_github_pm_handlers(self): + assert "list_tickets" in GITHUB_PM_HANDLERS + assert "create_ticket" in GITHUB_PM_HANDLERS + assert "get_ticket" in GITHUB_PM_HANDLERS + assert "update_ticket" in GITHUB_PM_HANDLERS + assert "lock_ticket" in GITHUB_PM_HANDLERS + assert "unlock_ticket" in GITHUB_PM_HANDLERS + assert GITHUB_PM_HANDLERS["list_tickets"] == list_tickets + assert GITHUB_PM_HANDLERS["create_ticket"] == create_ticket + assert GITHUB_PM_HANDLERS["get_ticket"] == get_ticket + assert GITHUB_PM_HANDLERS["update_ticket"] == update_ticket + assert GITHUB_PM_HANDLERS["lock_ticket"] == lock_ticket + assert GITHUB_PM_HANDLERS["unlock_ticket"] == unlock_ticket diff --git a/backend/core/tests/test_github_validator.py b/backend/core/tests/test_github_validator.py new file mode 100644 index 0000000..5faa105 --- /dev/null +++ b/backend/core/tests/test_github_validator.py @@ -0,0 +1,205 @@ +import pytest +from unittest.mock import Mock, patch + +from core.integrations.github_validator import validate_github_token + + +@pytest.mark.unit +class TestValidateGithubToken: + def test_validate_github_token_missing_token(self): + credentials = {} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is False + assert message == 'Token is required' + assert metadata == {} + + def test_validate_github_token_empty_token(self): + credentials = {'token': ''} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is False + assert message == 'Token is required' + assert metadata == {} + + def test_validate_github_token_none_token(self): + credentials = {'token': None} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is False + assert message == 'Token is required' + assert metadata == {} + + @patch("core.integrations.github_validator.requests.get") + def test_validate_github_token_success(self, mock_get): + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + 'login': 'testuser', + 'name': 'Test User', + 'avatar_url': 'https://github.com/testuser.png', + 'html_url': 'https://github.com/testuser' + } + mock_get.return_value = mock_response + + credentials = {'token': 'ghp_test_token_123'} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is True + assert message == '' + assert metadata['login'] == 'testuser' + assert metadata['name'] == 'Test User' + assert metadata['avatar_url'] == 'https://github.com/testuser.png' + assert metadata['html_url'] == 'https://github.com/testuser' + + mock_get.assert_called_once_with( + 'https://api.github.com/user', + headers={ + 'Authorization': 'Bearer ghp_test_token_123', + 'Accept': 'application/vnd.github+json', + }, + timeout=10, + ) + + @patch("core.integrations.github_validator.requests.get") + def test_validate_github_token_partial_metadata(self, mock_get): + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + 'login': 'testuser', + 'avatar_url': 'https://github.com/testuser.png' + } + mock_get.return_value = mock_response + + credentials = {'token': 'ghp_test_token_123'} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is True + assert message == '' + assert metadata['login'] == 'testuser' + assert metadata['name'] is None + assert metadata['avatar_url'] == 'https://github.com/testuser.png' + assert metadata['html_url'] is None + + @patch("core.integrations.github_validator.requests.get") + def test_validate_github_token_unauthorized(self, mock_get): + mock_response = Mock() + mock_response.status_code = 401 + mock_get.return_value = mock_response + + credentials = {'token': 'invalid_token'} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is False + assert message == 'GitHub returned 401' + assert metadata == {} + + @patch("core.integrations.github_validator.requests.get") + def test_validate_github_token_forbidden(self, mock_get): + mock_response = Mock() + mock_response.status_code = 403 + mock_get.return_value = mock_response + + credentials = {'token': 'forbidden_token'} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is False + assert message == 'GitHub returned 403' + assert metadata == {} + + @patch("core.integrations.github_validator.requests.get") + def test_validate_github_token_not_found(self, mock_get): + mock_response = Mock() + mock_response.status_code = 404 + mock_get.return_value = mock_response + + credentials = {'token': 'not_found_token'} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is False + assert message == 'GitHub returned 404' + assert metadata == {} + + @patch("core.integrations.github_validator.requests.get") + def test_validate_github_token_server_error(self, mock_get): + mock_response = Mock() + mock_response.status_code = 500 + mock_get.return_value = mock_response + + credentials = {'token': 'server_error_token'} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is False + assert message == 'GitHub returned 500' + assert metadata == {} + + @patch("core.integrations.github_validator.requests.get") + def test_validate_github_token_timeout(self, mock_get): + import requests + mock_get.side_effect = requests.Timeout("Connection timed out") + + credentials = {'token': 'timeout_token'} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is False + assert 'timed out' in message.lower() or 'timeout' in message.lower() + assert metadata == {} + + @patch("core.integrations.github_validator.requests.get") + def test_validate_github_token_connection_error(self, mock_get): + import requests + mock_get.side_effect = requests.ConnectionError("Failed to establish connection") + + credentials = {'token': 'connection_error_token'} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is False + assert 'connection' in message.lower() or 'failed' in message.lower() + assert metadata == {} + + @patch("core.integrations.github_validator.requests.get") + def test_validate_github_token_request_exception(self, mock_get): + import requests + mock_get.side_effect = requests.RequestException("Generic error") + + credentials = {'token': 'error_token'} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is False + assert message == 'Generic error' + assert metadata == {} + + @patch("core.integrations.github_validator.requests.get") + def test_validate_github_token_http_error(self, mock_get): + import requests + mock_get.side_effect = requests.HTTPError("HTTP Error occurred") + + credentials = {'token': 'http_error_token'} + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is False + assert 'http error' in message.lower() + assert metadata == {} + + @patch("core.integrations.github_validator.requests.get") + def test_validate_github_token_with_extra_credentials_fields(self, mock_get): + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + 'login': 'testuser', + 'name': 'Test User', + 'avatar_url': 'https://github.com/testuser.png', + 'html_url': 'https://github.com/testuser' + } + mock_get.return_value = mock_response + + credentials = { + 'token': 'ghp_test_token_123', + 'extra_field': 'ignored', + 'another_field': 'also_ignored' + } + is_valid, message, metadata = validate_github_token(credentials) + + assert is_valid is True + assert message == '' + assert metadata['login'] == 'testuser' diff --git a/backend/core/tests/test_integration_dispatcher.py b/backend/core/tests/test_integration_dispatcher.py new file mode 100644 index 0000000..ecb178f --- /dev/null +++ b/backend/core/tests/test_integration_dispatcher.py @@ -0,0 +1,366 @@ +import pytest +from unittest.mock import Mock, patch, MagicMock + +from core.integrations.integration_dispatcher import ( + get_enabled_tools_for_app, + execute_tool_call, +) + + +@pytest.mark.unit +class TestGetEnabledToolsForApp: + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.ToolConfig") + @patch("core.integrations.integration_dispatcher.INTEGRATION_TOOLS") + def test_get_enabled_tools_no_integrations(self, mock_tools, mock_tc, mock_ai): + mock_ai.objects.filter.return_value.select_related.return_value = [] + + result = get_enabled_tools_for_app("app-uuid-123") + + assert result == [] + mock_ai.objects.filter.assert_called_once_with(application__uuid="app-uuid-123", is_active=True) + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.ToolConfig") + @patch("core.integrations.integration_dispatcher.INTEGRATION_TOOLS") + def test_get_enabled_tools_with_builtin_tools_enabled(self, mock_tools, mock_tc, mock_ai): + mock_app_integration = Mock() + mock_app_integration.integration.provider = "github" + mock_app_integration.integration_type = "version_control" + mock_ai.objects.filter.return_value.select_related.return_value = [mock_app_integration] + + mock_tools.get.return_value = { + "list_commits": { + "id": "list_commits", + "title": "List Commits", + "schema": {"type": "function", "function": {"name": "list_commits"}} + } + } + + def filter_side_effect(*args, **kwargs): + if kwargs.get("is_builtin") is True: + return [ + Mock(tool_id="github_version_control:list_commits", is_enabled=True) + ] + return [] + + mock_tc.objects.filter.side_effect = filter_side_effect + + result = get_enabled_tools_for_app("app-uuid-123") + + assert len(result) == 1 + assert result[0]["function"]["name"] == "list_commits" + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.ToolConfig") + @patch("core.integrations.integration_dispatcher.INTEGRATION_TOOLS") + def test_get_enabled_tools_with_builtin_tools_disabled(self, mock_tools, mock_tc, mock_ai): + mock_app_integration = Mock() + mock_app_integration.integration.provider = "github" + mock_app_integration.integration_type = "version_control" + mock_ai.objects.filter.return_value.select_related.return_value = [mock_app_integration] + + mock_tools.get.return_value = { + "list_commits": { + "id": "list_commits", + "title": "List Commits", + "schema": {"type": "function", "function": {"name": "list_commits"}} + } + } + + def filter_side_effect(*args, **kwargs): + if kwargs.get("is_builtin") is True: + return [ + Mock(tool_id="github_version_control:list_commits", is_enabled=False) + ] + return [] + + mock_tc.objects.filter.side_effect = filter_side_effect + + result = get_enabled_tools_for_app("app-uuid-123") + + assert len(result) == 0 + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.ToolConfig") + @patch("core.integrations.integration_dispatcher.INTEGRATION_TOOLS") + @patch("core.integrations.integration_dispatcher.parse_url_schema") + def test_get_enabled_tools_with_custom_tools(self, mock_parse, mock_tools, mock_tc, mock_ai): + mock_app_integration = Mock() + mock_app_integration.integration.provider = "github" + mock_app_integration.integration_type = "version_control" + mock_ai.objects.filter.return_value.select_related.return_value = [mock_app_integration] + + mock_tools.get.return_value = {} + + def filter_side_effect(*args, **kwargs): + if kwargs.get("is_builtin") is True: + return [] + elif kwargs.get("is_builtin") is False and kwargs.get("is_enabled") is True: + mock_custom_tc = Mock() + mock_custom_tc.title = "Custom Tool" + mock_custom_tc.description = "A custom tool" + mock_custom_tc.url_schema = 'curl -X POST https://api.example.com/endpoint --data \'{"test": "value"}\'' + return [mock_custom_tc] + return [] + + mock_tc.objects.filter.side_effect = filter_side_effect + + mock_parse.return_value = { + "type": "function", + "function": {"name": "custom_tool", "description": "A custom tool"} + } + + result = get_enabled_tools_for_app("app-uuid-123") + + assert len(result) == 1 + assert result[0]["function"]["name"] == "custom_tool" + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.ToolConfig") + @patch("core.integrations.integration_dispatcher.INTEGRATION_TOOLS") + def test_get_enabled_tools_multiple_integrations(self, mock_tools, mock_tc, mock_ai): + mock_integration1 = Mock() + mock_integration1.integration.provider = "github" + mock_integration1.integration_type = "version_control" + + mock_integration2 = Mock() + mock_integration2.integration.provider = "github" + mock_integration2.integration_type = "project_management" + + mock_ai.objects.filter.return_value.select_related.return_value = [mock_integration1, mock_integration2] + + def tools_side_effect(key, default=None): + if key == "github_version_control": + return { + "list_commits": { + "id": "list_commits", + "title": "List Commits", + "schema": {"type": "function", "function": {"name": "list_commits"}} + } + } + elif key == "github_project_management": + return { + "list_tickets": { + "id": "list_tickets", + "title": "List Tickets", + "schema": {"type": "function", "function": {"name": "list_tickets"}} + } + } + return default or {} + + mock_tools.get.side_effect = tools_side_effect + + filter_call_count = [0] + + def tc_filter_side_effect(*args, **kwargs): + filter_call_count[0] += 1 + if kwargs.get("is_builtin") is True: + if filter_call_count[0] == 1: + return [ + Mock(tool_id="github_version_control:list_commits", is_enabled=True) + ] + else: + return [ + Mock(tool_id="github_project_management:list_tickets", is_enabled=True) + ] + return [] + + mock_tc.objects.filter.side_effect = tc_filter_side_effect + + result = get_enabled_tools_for_app("app-uuid-123") + + assert len(result) == 2 + tool_names = [t["function"]["name"] for t in result] + assert "list_commits" in tool_names + assert "list_tickets" in tool_names + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.ToolConfig") + @patch("core.integrations.integration_dispatcher.INTEGRATION_TOOLS") + def test_get_enabled_tools_unknown_integration_key(self, mock_tools, mock_tc, mock_ai): + mock_app_integration = Mock() + mock_app_integration.integration.provider = "unknown" + mock_app_integration.integration_type = "type" + mock_ai.objects.filter.return_value.select_related.return_value = [mock_app_integration] + + mock_tools.get.return_value = {} + + def filter_side_effect(*args, **kwargs): + return [] + + mock_tc.objects.filter.side_effect = filter_side_effect + + result = get_enabled_tools_for_app("app-uuid-123") + + assert result == [] + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.ToolConfig") + @patch("core.integrations.integration_dispatcher.INTEGRATION_TOOLS") + def test_get_enabled_tools_no_tool_configs(self, mock_tools, mock_tc, mock_ai): + mock_app_integration = Mock() + mock_app_integration.integration.provider = "github" + mock_app_integration.integration_type = "version_control" + mock_ai.objects.filter.return_value.select_related.return_value = [mock_app_integration] + + mock_tools.get.return_value = { + "list_commits": { + "id": "list_commits", + "title": "List Commits", + "schema": {"type": "function", "function": {"name": "list_commits"}} + } + } + + def filter_side_effect(*args, **kwargs): + return [] + + mock_tc.objects.filter.side_effect = filter_side_effect + + result = get_enabled_tools_for_app("app-uuid-123") + + assert result == [] + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.ToolConfig") + @patch("core.integrations.integration_dispatcher.INTEGRATION_TOOLS") + def test_get_enabled_tools_exception_returns_empty(self, mock_tools, mock_tc, mock_ai): + mock_ai.objects.filter.side_effect = Exception("Database error") + + result = get_enabled_tools_for_app("app-uuid-123") + + assert result == [] + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.ToolConfig") + @patch("core.integrations.integration_dispatcher.INTEGRATION_TOOLS") + def test_get_enabled_tools_default_disabled(self, mock_tools, mock_tc, mock_ai): + mock_app_integration = Mock() + mock_app_integration.integration.provider = "github" + mock_app_integration.integration_type = "version_control" + mock_ai.objects.filter.return_value.select_related.return_value = [mock_app_integration] + + mock_tools.get.return_value = { + "list_commits": { + "id": "list_commits", + "title": "List Commits", + "schema": {"type": "function", "function": {"name": "list_commits"}} + } + } + + def filter_side_effect(*args, **kwargs): + return [] + + mock_tc.objects.filter.side_effect = filter_side_effect + + result = get_enabled_tools_for_app("app-uuid-123") + + assert len(result) == 0 + + +@pytest.mark.unit +class TestExecuteToolCall: + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.INTEGRATION_HANDLERS") + def test_execute_tool_call_success(self, mock_handlers, mock_ai): + mock_app_integration = Mock() + mock_app_integration.integration.provider = "github" + mock_app_integration.integration_type = "version_control" + mock_ai.objects.filter.return_value.select_related.return_value = [mock_app_integration] + + mock_handler = Mock(return_value={"result": "success"}) + mock_handlers.get.return_value = {"list_commits": mock_handler} + + result = execute_tool_call("app-uuid-123", "list_commits", sha="main") + + assert result == {"result": "success"} + mock_handler.assert_called_once_with(mock_app_integration, sha="main") + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.INTEGRATION_HANDLERS") + def test_execute_tool_call_no_handler_found(self, mock_handlers, mock_ai): + mock_app_integration = Mock() + mock_app_integration.integration.provider = "github" + mock_app_integration.integration_type = "version_control" + mock_ai.objects.filter.return_value.select_related.return_value = [mock_app_integration] + + mock_handlers.get.return_value = {} + + with pytest.raises(ValueError) as exc_info: + execute_tool_call("app-uuid-123", "unknown_tool") + + assert "No handler found for tool 'unknown_tool'" in str(exc_info.value) + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.INTEGRATION_HANDLERS") + def test_execute_tool_call_no_integrations(self, mock_handlers, mock_ai): + mock_ai.objects.filter.return_value.select_related.return_value = [] + + with pytest.raises(ValueError) as exc_info: + execute_tool_call("app-uuid-123", "list_commits") + + assert "No handler found for tool 'list_commits'" in str(exc_info.value) + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.INTEGRATION_HANDLERS") + def test_execute_tool_call_multiple_integrations(self, mock_handlers, mock_ai): + mock_integration1 = Mock() + mock_integration1.integration.provider = "github" + mock_integration1.integration_type = "version_control" + + mock_integration2 = Mock() + mock_integration2.integration.provider = "github" + mock_integration2.integration_type = "project_management" + + mock_ai.objects.filter.return_value.select_related.return_value = [mock_integration1, mock_integration2] + + mock_handler = Mock(return_value={"result": "success"}) + mock_handlers.get.side_effect = lambda key, default=None: { + "github_version_control": {"list_commits": mock_handler}, + "github_project_management": {"list_tickets": Mock()} + }.get(key, default or {}) + + result = execute_tool_call("app-uuid-123", "list_commits", sha="main") + + assert result == {"result": "success"} + mock_handler.assert_called_once_with(mock_integration1, sha="main") + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.INTEGRATION_HANDLERS") + def test_execute_tool_call_with_arguments(self, mock_handlers, mock_ai): + mock_app_integration = Mock() + mock_app_integration.integration.provider = "github" + mock_app_integration.integration_type = "version_control" + mock_ai.objects.filter.return_value.select_related.return_value = [mock_app_integration] + + mock_handler = Mock(return_value={"result": "success"}) + mock_handlers.get.return_value = {"list_commits": mock_handler} + + result = execute_tool_call( + "app-uuid-123", + "list_commits", + sha="main", + path="src/", + author="john", + per_page=10 + ) + + assert result == {"result": "success"} + mock_handler.assert_called_once_with( + mock_app_integration, + sha="main", + path="src/", + author="john", + per_page=10 + ) + + @patch("core.integrations.integration_dispatcher.AppIntegration") + @patch("core.integrations.integration_dispatcher.INTEGRATION_HANDLERS") + def test_execute_tool_call_inactive_integration(self, mock_handlers, mock_ai): + mock_ai.objects.filter.return_value.select_related.return_value = [] + + with pytest.raises(ValueError) as exc_info: + execute_tool_call("app-uuid-123", "list_commits") + + assert "No handler found for tool 'list_commits'" in str(exc_info.value) + mock_ai.objects.filter.assert_called_once_with(application__uuid="app-uuid-123", is_active=True)