Repository: BeehiveInnovations/zen-mcp-server
File: tests/test_consensus.py (Python)

Code before:
        assert "status" in result

        # Test between model consultations
        request = Mock(step_number=2, current_model_index=1)
        response_data = {}

        result = tool.handle_work_continuation(response_data, request)
        assert "status" in result

    def test_customize_workflow_response(self):
        """Test response customization for consensus workflow."""
        tool = ConsensusTool()
        tool.accumulated_responses = [{"model": "test", "response": "data"}]

        # Test different step numbers (new workflow: 2 models = 2 steps)
        request = Mock(step_number=1, total_steps=2)
        response_data = {}
        result = tool.customize_workflow_response(response_data, request)
        assert result["consensus_workflow_status"] == "initial_analysis_complete"

        request = Mock(step_number=2, total_steps=2)
        response_data = {}
        result = tool.customize_workflow_response(response_data, request)
        assert result["consensus_workflow_status"] == "ready_for_synthesis"


if __name__ == "__main__":
    import unittest

    unittest.main()


Review comment (bug):
The method isn’t marked with `@pytest.mark.asyncio`. In this test suite we rely on pytest (not unittest) to run these tests, and elsewhere every async test is decorated for the asyncio plugin. Without that decorator pytest will treat the coroutine as a normal function, emit a "coroutine was never awaited" error, and the test will fail. Please add `@pytest.mark.asyncio` so the suite runs.

Code after:
        assert "status" in result

        # Test between model consultations
        request = Mock(step_number=2, current_model_index=1)
        response_data = {}

        result = tool.handle_work_continuation(response_data, request)
        assert "status" in result

    def test_customize_workflow_response(self):
        """Test response customization for consensus workflow."""
        tool = ConsensusTool()
        tool.accumulated_responses = [{"model": "test", "response": "data"}]

        # Test different step numbers (new workflow: 2 models = 2 steps)
        request = Mock(step_number=1, total_steps=2)
        response_data = {}
        result = tool.customize_workflow_response(response_data, request)
        assert result["consensus_workflow_status"] == "initial_analysis_complete"

        request = Mock(step_number=2, total_steps=2)
        response_data = {}
        result = tool.customize_workflow_response(response_data, request)
        assert result["consensus_workflow_status"] == "ready_for_synthesis"

    async def test_consensus_with_relevant_files_model_context_fix(self):
        """Test that consensus tool properly handles relevant_files without RuntimeError.

        This is a regression test for the bug where _prepare_file_content_for_prompt
        was called without model_context parameter, causing RuntimeError:
        'Model context not provided for file preparation'

        Bug details:
        - Occurred when consensus tool processed requests with relevant_files
        - _consult_model method called _prepare_file_content_for_prompt without model_context
        - Method expected model_context parameter but got None (default value)
        - Runtime validation in base_tool.py threw RuntimeError
        """
        from unittest.mock import AsyncMock, Mock, patch

        from utils.model_context import ModelContext

        tool = ConsensusTool()

        # Create a mock request with relevant_files (the trigger condition)
        mock_request = Mock()
        mock_request.relevant_files = ["/test/file1.py", "/test/file2.js"]
        mock_request.continuation_id = None

        # Mock model configuration
        model_config = {"model": "flash", "stance": "neutral"}