"""Tests for pipeline (CI/CD) tools."""
from unittest.mock import patch, MagicMock
from bitbucket_mcp.tools import pipelines as _mod
list_pipelines = _mod.list_pipelines.fn
get_pipeline = _mod.get_pipeline.fn
trigger_pipeline = _mod.trigger_pipeline.fn
stop_pipeline = _mod.stop_pipeline.fn
list_pipeline_steps = _mod.list_pipeline_steps.fn
get_pipeline_step_log = _mod.get_pipeline_step_log.fn
list_pipeline_variables = _mod.list_pipeline_variables.fn
create_pipeline_variable = _mod.create_pipeline_variable.fn
from tests.conftest import make_mock_response
class TestListPipelines:
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_success(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.get.return_value = make_mock_response(200, {
"values": [
{
"uuid": "{pipe-uuid}",
"build_number": 42,
"state": {"name": "COMPLETED", "result": {"name": "SUCCESSFUL"}},
"target": {
"ref_name": "main",
"commit": {"hash": "abc123def456789"},
},
"trigger": {"type": "push"},
"created_on": "2025-01-01",
"completed_on": "2025-01-01",
"duration_in_seconds": 120,
}
],
"size": 1, "page": 1, "pagelen": 25,
})
result = list_pipelines("repo")
assert result["success"] is True
assert len(result["pipelines"]) == 1
assert result["pipelines"][0]["build_number"] == 42
assert result["pipelines"][0]["state"] == "COMPLETED"
assert result["pipelines"][0]["result"] == "SUCCESSFUL"
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_not_found(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.get.return_value = make_mock_response(404)
result = list_pipelines("nonexistent")
assert result["success"] is False
assert "not found" in result["error"]
class TestGetPipeline:
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_success(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.get.return_value = make_mock_response(200, {
"uuid": "{pipe-uuid}",
"build_number": 42,
"state": {"name": "COMPLETED", "result": {"name": "SUCCESSFUL"}},
"target": {
"ref_name": "main",
"commit": {"hash": "abc123"},
},
"created_on": "2025-01-01",
"completed_on": "2025-01-01",
"duration_in_seconds": 120,
"has_variables": False,
})
result = get_pipeline("repo", "pipe-uuid")
assert result["success"] is True
assert result["build_number"] == 42
assert result["state"] == "COMPLETED"
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_not_found(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.get.return_value = make_mock_response(404)
result = get_pipeline("repo", "nonexistent")
assert result["success"] is False
assert "not found" in result["error"]
class TestTriggerPipeline:
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_success(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.post.return_value = make_mock_response(201, {
"uuid": "{new-pipe}",
"build_number": 43,
"state": {"name": "PENDING"},
"created_on": "2025-01-01",
})
result = trigger_pipeline("repo", branch="develop")
assert result["success"] is True
assert result["build_number"] == 43
assert result["state"] == "PENDING"
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_bad_request(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.post.return_value = make_mock_response(400, {
"error": {"message": "Pipeline configuration not found"}
})
result = trigger_pipeline("repo")
assert result["success"] is False
assert "Pipeline configuration not found" in result["error"]
class TestStopPipeline:
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_success(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.post.return_value = make_mock_response(204)
result = stop_pipeline("repo", "pipe-uuid")
assert result["success"] is True
assert "stop requested" in result["message"]
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_not_stoppable(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.post.return_value = make_mock_response(409)
result = stop_pipeline("repo", "pipe-uuid")
assert result["success"] is False
assert "not in a stoppable state" in result["error"]
class TestListPipelineSteps:
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_success(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.get.return_value = make_mock_response(200, {
"values": [
{
"uuid": "{step-uuid}",
"name": "Build",
"state": {"name": "COMPLETED", "result": {"name": "SUCCESSFUL"}},
"started_on": "2025-01-01",
"completed_on": "2025-01-01",
"duration_in_seconds": 60,
"image": {"name": "python:3.11"},
}
]
})
result = list_pipeline_steps("repo", "pipe-uuid")
assert result["success"] is True
assert len(result["steps"]) == 1
assert result["steps"][0]["name"] == "Build"
assert result["steps"][0]["image"] == "python:3.11"
class TestGetPipelineStepLog:
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_success(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
resp = make_mock_response(200)
resp.text = "+ pip install\n+ pytest\nAll tests passed"
client.get.return_value = resp
result = get_pipeline_step_log("repo", "pipe-uuid", "step-uuid")
assert result["success"] is True
assert "pytest" in result["log"]
assert result["truncated"] is False
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_not_found(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.get.return_value = make_mock_response(404)
result = get_pipeline_step_log("repo", "pipe-uuid", "bad-step")
assert result["success"] is False
assert "not found" in result["error"]
class TestListPipelineVariables:
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_success(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.get.return_value = make_mock_response(200, {
"values": [
{
"uuid": "{var-uuid}",
"key": "DEPLOY_ENV",
"value": "staging",
"secured": False,
}
],
"size": 1, "page": 1, "pagelen": 25,
})
result = list_pipeline_variables("repo")
assert result["success"] is True
assert len(result["variables"]) == 1
assert result["variables"][0]["key"] == "DEPLOY_ENV"
class TestCreatePipelineVariable:
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_success(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.post.return_value = make_mock_response(201, {
"uuid": "{new-var}",
"key": "API_KEY",
"secured": True,
})
result = create_pipeline_variable("repo", "API_KEY", "secret", secured=True)
assert result["success"] is True
assert result["key"] == "API_KEY"
assert result["secured"] is True
@patch("bitbucket_mcp.tools.pipelines.httpx.Client")
def test_already_exists(self, MockClient, saved_config):
client = MagicMock()
MockClient.return_value.__enter__ = MagicMock(return_value=client)
MockClient.return_value.__exit__ = MagicMock(return_value=False)
client.post.return_value = make_mock_response(409)
result = create_pipeline_variable("repo", "API_KEY", "val")
assert result["success"] is False
assert "already exists" in result["error"]
