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.
This commit is contained in:
Otto
2026-02-06 16:28:31 +00:00
parent b20a907462
commit 83248f2b32
2 changed files with 245 additions and 0 deletions

View File

@@ -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"

View File

@@ -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