"""Tests for MCP tool definitions."""
import pytest
from unittest.mock import Mock, MagicMock, patch
from src.mcp_tools import (
_poll_job_status,
_format_pipeline_response,
_format_job_response,
_format_error_response,
)
from src.gitlab_client import GitLabClientError
class TestPollJobStatus:
"""Tests for job status polling logic."""
def test_poll_job_success_immediate(self):
"""Test polling returns immediately when job is successful."""
mock_client = MagicMock()
mock_client.get_job_status.return_value = {
'id': 123,
'name': 'test',
'status': 'success',
'stage': 'test',
'started_at': '2026-02-10T10:00:00Z',
'finished_at': '2026-02-10T10:01:00Z',
'duration': 60,
'web_url': 'http://example.com/jobs/123',
}
result = _poll_job_status(
mock_client, "group/project", None, 123, timeout_seconds=30
)
assert result['status'] == 'success'
assert result['is_polling'] is False
assert result['polling_timeout'] is False
assert 'polling_duration_seconds' in result
def test_poll_job_by_name(self):
"""Test polling job identified by name."""
mock_client = MagicMock()
mock_client.get_job_status.return_value = {
'id': 456,
'name': 'deploy',
'status': 'failed',
'stage': 'deploy',
'started_at': '2026-02-10T10:05:00Z',
'finished_at': '2026-02-10T10:06:00Z',
'duration': 60,
'web_url': 'http://example.com/jobs/456',
}
result = _poll_job_status(
mock_client, "group/project", "deploy", None, timeout_seconds=30
)
assert result['name'] == 'deploy'
assert result['status'] == 'failed'
assert result['is_polling'] is False
def test_poll_job_timeout(self, monkeypatch):
"""Test polling times out after max duration."""
import time
monkeypatch.setattr(time, "sleep", lambda _: None)
mock_client = MagicMock()
mock_client.get_job_status.return_value = {
'id': 789,
'name': 'build',
'status': 'running',
'stage': 'build',
'started_at': '2026-02-10T10:00:00Z',
'finished_at': None,
'duration': None,
'web_url': 'http://example.com/jobs/789',
}
result = _poll_job_status(
mock_client, "group/project", "build", None,
timeout_seconds=0.1, poll_interval=0.05
)
assert result['is_polling'] is True
assert result['polling_timeout'] is True
assert result['status'] == 'running'
def test_poll_job_terminal_states(self):
"""Test that polling stops at terminal job states."""
mock_client = MagicMock()
for terminal_state in ['success', 'failed', 'canceled', 'skipped']:
mock_client.get_job_status.return_value = {
'id': 999,
'name': 'test',
'status': terminal_state,
'stage': 'test',
'started_at': '2026-02-10T10:00:00Z',
'finished_at': '2026-02-10T10:01:00Z',
'duration': 60,
'web_url': 'http://example.com/jobs/999',
}
result = _poll_job_status(
mock_client, "group/project", "test", None, timeout_seconds=30
)
assert result['status'] == terminal_state
assert result['is_polling'] is False
class TestFormatPipelineResponse:
"""Tests for pipeline response formatting."""
def test_format_pipeline_response_basic(self):
"""Test basic pipeline response formatting."""
pipeline_info = {
'id': 100,
'status': 'running',
'ref': 'main',
'sha': 'abcdef123456',
'created_at': '2026-02-10T10:00:00Z',
'updated_at': '2026-02-10T10:05:00Z',
'started_at': '2026-02-10T10:01:00Z',
'finished_at': None,
'duration': None,
'web_url': 'http://example.com/pipelines/100',
'jobs': [
{
'id': 1,
'name': 'build',
'status': 'success',
'stage': 'build',
'web_url': 'http://example.com/jobs/1',
},
{
'id': 2,
'name': 'test',
'status': 'running',
'stage': 'test',
'web_url': 'http://example.com/jobs/2',
},
]
}
response = _format_pipeline_response(pipeline_info)
assert 'Pipeline Status Report' in response
assert 'abcdef12' in response # Truncated SHA
assert 'build' in response
assert 'test' in response
assert 'running' in response
def test_format_pipeline_response_with_no_jobs(self):
"""Test pipeline response with empty jobs list."""
pipeline_info = {
'id': 200,
'status': 'success',
'ref': 'develop',
'sha': 'deadbeef0000',
'created_at': '2026-02-10T09:00:00Z',
'updated_at': '2026-02-10T09:30:00Z',
'started_at': '2026-02-10T09:01:00Z',
'finished_at': '2026-02-10T09:30:00Z',
'duration': 1740,
'web_url': 'http://example.com/pipelines/200',
'jobs': []
}
response = _format_pipeline_response(pipeline_info)
assert 'Pipeline Status Report' in response
assert '200' in response
class TestFormatJobResponse:
"""Tests for job response formatting."""
def test_format_job_response_completed(self):
"""Test formatting response for completed job."""
job_info = {
'id': 321,
'name': 'lint',
'status': 'success',
'stage': 'test',
'started_at': '2026-02-10T10:00:30Z',
'finished_at': '2026-02-10T10:01:00Z',
'duration': 30,
'web_url': 'http://example.com/jobs/321',
'is_polling': False,
'polling_timeout': False,
'polling_duration_seconds': 0.5,
}
response = _format_job_response(job_info, pipeline_id=999)
assert 'Job Status Report' in response
assert 'lint' in response
assert 'success' in response
assert '999' in response
# When is_polling is False, no polling_status is appended
assert 'Job Status Report' in response
def test_format_job_response_polling_timeout(self):
"""Test formatting response for polling timeout."""
job_info = {
'id': 555,
'name': 'deploy',
'status': 'running',
'stage': 'deploy',
'started_at': '2026-02-10T10:05:00Z',
'finished_at': None,
'duration': None,
'web_url': 'http://example.com/jobs/555',
'is_polling': True,
'polling_timeout': True,
'polling_duration_seconds': 30.5,
}
response = _format_job_response(job_info, pipeline_id=888)
assert 'Job Status Report' in response
assert 'deploy' in response
assert 'POLLING TIMEOUT' in response
assert '30.5' in response
def test_format_job_response_polling_completed(self):
"""Test formatting response for job completed via polling."""
job_info = {
'id': 777,
'name': 'integration-test',
'status': 'success',
'stage': 'test',
'started_at': '2026-02-10T10:10:00Z',
'finished_at': '2026-02-10T10:15:00Z',
'duration': 300,
'web_url': 'http://example.com/jobs/777',
'is_polling': True,
'polling_timeout': False,
'polling_duration_seconds': 5.2,
}
response = _format_job_response(job_info, pipeline_id=777)
assert 'Job Status Report' in response
assert 'integration-test' in response
assert 'Job completed after' in response
assert '5.2' in response
class TestFormatErrorResponse:
"""Tests for error response formatting."""
def test_format_error_response(self):
"""Test error response formatting."""
error_msg = "Project not found: invalid/path"
response = _format_error_response(error_msg)
assert 'Error' in response
assert 'Project not found' in response
assert 'invalid/path' in response
