From 83248f2b325e3f61d78961af8dc24266b57ebd3b Mon Sep 17 00:00:00 2001 From: Otto Date: Fri, 6 Feb 2026 16:28:31 +0000 Subject: [PATCH] test: add tests for CoPilot block filtering logic Tests verify: - EXCLUDED_BLOCK_TYPES contains all graph-only block types - EXCLUDED_BLOCK_IDS contains SmartDecisionMakerBlock - FindBlockTool filters excluded blocks from search results - RunBlockTool returns error for excluded blocks - Non-excluded blocks pass through the filtering guard Addresses Copilot review comments on PR #11892. --- .../features/chat/tools/find_block_test.py | 134 ++++++++++++++++++ .../api/features/chat/tools/run_block_test.py | 111 +++++++++++++++ 2 files changed, 245 insertions(+) create mode 100644 autogpt_platform/backend/backend/api/features/chat/tools/find_block_test.py create mode 100644 autogpt_platform/backend/backend/api/features/chat/tools/run_block_test.py diff --git a/autogpt_platform/backend/backend/api/features/chat/tools/find_block_test.py b/autogpt_platform/backend/backend/api/features/chat/tools/find_block_test.py new file mode 100644 index 0000000000..42f453d436 --- /dev/null +++ b/autogpt_platform/backend/backend/api/features/chat/tools/find_block_test.py @@ -0,0 +1,134 @@ +"""Tests for block filtering in FindBlockTool.""" + +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from backend.api.features.chat.tools.find_block import FindBlockTool +from backend.data.block import EXCLUDED_BLOCK_IDS, EXCLUDED_BLOCK_TYPES, BlockType + +from ._test_data import make_session, setup_test_data + +# Prevent formatter from removing fixture imports +setup_test_data = setup_test_data + + +def make_mock_block( + block_id: str, name: str, block_type: BlockType, disabled: bool = False +): + """Create a mock block for testing.""" + mock = MagicMock() + mock.id = block_id + mock.name = name + mock.block_type = block_type + mock.disabled = disabled + mock.input_schema = MagicMock() + mock.input_schema.jsonschema.return_value = {"properties": {}, "required": []} + mock.output_schema = MagicMock() + mock.output_schema.jsonschema.return_value = {} + mock.categories = [] + return mock + + +class TestFindBlockFiltering: + """Tests for block filtering in FindBlockTool.""" + + def test_excluded_block_types_contains_expected_types(self): + """Verify EXCLUDED_BLOCK_TYPES contains all graph-only types.""" + assert BlockType.INPUT in EXCLUDED_BLOCK_TYPES + assert BlockType.OUTPUT in EXCLUDED_BLOCK_TYPES + assert BlockType.WEBHOOK in EXCLUDED_BLOCK_TYPES + assert BlockType.WEBHOOK_MANUAL in EXCLUDED_BLOCK_TYPES + assert BlockType.NOTE in EXCLUDED_BLOCK_TYPES + assert BlockType.HUMAN_IN_THE_LOOP in EXCLUDED_BLOCK_TYPES + assert BlockType.AGENT in EXCLUDED_BLOCK_TYPES + + def test_excluded_block_ids_contains_smart_decision_maker(self): + """Verify SmartDecisionMakerBlock is in EXCLUDED_BLOCK_IDS.""" + assert "3b191d9f-356f-482d-8238-ba04b6d18381" in EXCLUDED_BLOCK_IDS + + @pytest.mark.asyncio(loop_scope="session") + async def test_excluded_block_type_filtered_from_results(self, setup_test_data): + """Verify blocks with excluded BlockTypes are filtered from search results.""" + user = setup_test_data["user"] + session = make_session(user_id=user.id) + + # Mock search returns an INPUT block (excluded) and a STANDARD block (included) + search_results = [ + {"content_id": "input-block-id", "score": 0.9}, + {"content_id": "standard-block-id", "score": 0.8}, + ] + + input_block = make_mock_block("input-block-id", "Input Block", BlockType.INPUT) + standard_block = make_mock_block( + "standard-block-id", "HTTP Request", BlockType.STANDARD + ) + + def mock_get_block(block_id): + return { + "input-block-id": input_block, + "standard-block-id": standard_block, + }.get(block_id) + + with patch( + "backend.api.features.chat.tools.find_block.unified_hybrid_search", + new_callable=AsyncMock, + return_value=(search_results, 2), + ): + with patch( + "backend.api.features.chat.tools.find_block.get_block", + side_effect=mock_get_block, + ): + tool = FindBlockTool() + response = await tool._execute( + user_id=user.id, session=session, query="test" + ) + + # Should only return the standard block, not the INPUT block + assert hasattr(response, "blocks") + assert len(response.blocks) == 1 + assert response.blocks[0].id == "standard-block-id" + + @pytest.mark.asyncio(loop_scope="session") + async def test_excluded_block_id_filtered_from_results(self, setup_test_data): + """Verify SmartDecisionMakerBlock is filtered from search results.""" + user = setup_test_data["user"] + session = make_session(user_id=user.id) + + smart_decision_id = "3b191d9f-356f-482d-8238-ba04b6d18381" + search_results = [ + {"content_id": smart_decision_id, "score": 0.9}, + {"content_id": "normal-block-id", "score": 0.8}, + ] + + # SmartDecisionMakerBlock has STANDARD type but is excluded by ID + smart_block = make_mock_block( + smart_decision_id, "Smart Decision Maker", BlockType.STANDARD + ) + normal_block = make_mock_block( + "normal-block-id", "Normal Block", BlockType.STANDARD + ) + + def mock_get_block(block_id): + return { + smart_decision_id: smart_block, + "normal-block-id": normal_block, + }.get(block_id) + + with patch( + "backend.api.features.chat.tools.find_block.unified_hybrid_search", + new_callable=AsyncMock, + return_value=(search_results, 2), + ): + with patch( + "backend.api.features.chat.tools.find_block.get_block", + side_effect=mock_get_block, + ): + tool = FindBlockTool() + response = await tool._execute( + user_id=user.id, session=session, query="decision" + ) + + # Should only return normal block, not SmartDecisionMakerBlock + assert hasattr(response, "blocks") + assert len(response.blocks) == 1 + assert response.blocks[0].id == "normal-block-id" diff --git a/autogpt_platform/backend/backend/api/features/chat/tools/run_block_test.py b/autogpt_platform/backend/backend/api/features/chat/tools/run_block_test.py new file mode 100644 index 0000000000..27b99c29de --- /dev/null +++ b/autogpt_platform/backend/backend/api/features/chat/tools/run_block_test.py @@ -0,0 +1,111 @@ +"""Tests for block execution guards in RunBlockTool.""" + +import pytest +from unittest.mock import MagicMock, patch + +from backend.api.features.chat.tools.run_block import RunBlockTool +from backend.api.features.chat.tools.models import ErrorResponse +from backend.data.block import EXCLUDED_BLOCK_IDS, EXCLUDED_BLOCK_TYPES, BlockType + +from ._test_data import make_session, setup_test_data + +# Prevent formatter from removing fixture imports +setup_test_data = setup_test_data + + +def make_mock_block( + block_id: str, name: str, block_type: BlockType, disabled: bool = False +): + """Create a mock block for testing.""" + mock = MagicMock() + mock.id = block_id + mock.name = name + mock.block_type = block_type + mock.disabled = disabled + mock.input_schema = MagicMock() + mock.input_schema.jsonschema.return_value = {"properties": {}, "required": []} + mock.input_schema.get_credentials_fields_info.return_value = [] + return mock + + +class TestRunBlockFiltering: + """Tests for block execution guards in RunBlockTool.""" + + @pytest.mark.asyncio(loop_scope="session") + async def test_excluded_block_type_returns_error(self, setup_test_data): + """Attempting to execute a block with excluded BlockType returns error.""" + user = setup_test_data["user"] + session = make_session(user_id=user.id) + + input_block = make_mock_block( + "input-block-id", "Input Block", BlockType.INPUT + ) + + with patch( + "backend.api.features.chat.tools.run_block.get_block", + return_value=input_block, + ): + tool = RunBlockTool() + response = await tool._execute( + user_id=user.id, + session=session, + block_id="input-block-id", + input_data={}, + ) + + assert isinstance(response, ErrorResponse) + assert "cannot be run directly in CoPilot" in response.message + assert "designed for use within graphs only" in response.message + + @pytest.mark.asyncio(loop_scope="session") + async def test_excluded_block_id_returns_error(self, setup_test_data): + """Attempting to execute SmartDecisionMakerBlock returns error.""" + user = setup_test_data["user"] + session = make_session(user_id=user.id) + + smart_decision_id = "3b191d9f-356f-482d-8238-ba04b6d18381" + smart_block = make_mock_block( + smart_decision_id, "Smart Decision Maker", BlockType.STANDARD + ) + + with patch( + "backend.api.features.chat.tools.run_block.get_block", + return_value=smart_block, + ): + tool = RunBlockTool() + response = await tool._execute( + user_id=user.id, + session=session, + block_id=smart_decision_id, + input_data={}, + ) + + assert isinstance(response, ErrorResponse) + assert "cannot be run directly in CoPilot" in response.message + + @pytest.mark.asyncio(loop_scope="session") + async def test_non_excluded_block_passes_guard(self, setup_test_data): + """Non-excluded blocks pass the filtering guard (may fail later for other reasons).""" + user = setup_test_data["user"] + session = make_session(user_id=user.id) + + standard_block = make_mock_block( + "standard-id", "HTTP Request", BlockType.STANDARD + ) + + with patch( + "backend.api.features.chat.tools.run_block.get_block", + return_value=standard_block, + ): + tool = RunBlockTool() + response = await tool._execute( + user_id=user.id, + session=session, + block_id="standard-id", + input_data={}, + ) + + # Should NOT be an ErrorResponse about CoPilot exclusion + # (may be other errors like missing credentials, but not the exclusion guard) + if isinstance(response, ErrorResponse): + assert "cannot be run directly in CoPilot" not in response.message