f5c2733cfb
LangGraph-based release automation agent with: - PR discovery (webhook + polling) - AI code review via Claude Code CLI (subscription-based) - Auto-create Jira tickets for PRs without ticket ID - Jira ticket lifecycle management (code review -> staging -> done) - CI/CD pipeline trigger, polling, and approval gates - Slack interactive messages with approval buttons - Per-repo semantic versioning - PostgreSQL persistence (threads, staging, releases) - FastAPI API (webhooks, approvals, status, manual triggers) - Docker Compose deployment 1069 tests, 95%+ coverage.
295 lines
10 KiB
Python
295 lines
10 KiB
Python
"""Tests for graph/ci_nodes.py.
|
|
|
|
Written FIRST (TDD RED phase).
|
|
|
|
All external calls (azdo, slack, poll_until) are mocked.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from release_agent.graph.ci_nodes import notify_ci_result, poll_ci_build, trigger_ci_build
|
|
from release_agent.models.build import BuildStatus
|
|
from release_agent.models.pipeline import PipelineInfo
|
|
from tests.graph.conftest import build_config, build_mock_clients
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_pipeline(pipeline_id: int = 10, name: str = "CI-build") -> dict:
|
|
return {"id": pipeline_id, "name": name, "repo": "my-repo"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# trigger_ci_build
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestTriggerCiBuild:
|
|
"""Tests for trigger_ci_build node."""
|
|
|
|
async def test_triggers_pipeline_on_branch(self) -> None:
|
|
clients = build_mock_clients()
|
|
clients.azdo.list_build_pipelines.return_value = [
|
|
PipelineInfo(id=10, name="CI", repo="my-repo")
|
|
]
|
|
clients.azdo.trigger_pipeline.return_value = {"id": 555, "state": "inProgress"}
|
|
config = build_config(clients)
|
|
state = {"repo_name": "my-repo", "version": "v1.0.0"}
|
|
|
|
result = await trigger_ci_build(state, config)
|
|
|
|
clients.azdo.trigger_pipeline.assert_called_once()
|
|
assert "ci_build_id" in result
|
|
assert result["ci_build_id"] == 555
|
|
|
|
async def test_returns_ci_build_id(self) -> None:
|
|
clients = build_mock_clients()
|
|
clients.azdo.list_build_pipelines.return_value = [
|
|
PipelineInfo(id=20, name="build-and-test", repo="my-repo")
|
|
]
|
|
clients.azdo.trigger_pipeline.return_value = {"id": 999}
|
|
config = build_config(clients)
|
|
state = {"repo_name": "my-repo", "version": "v2.0.0"}
|
|
|
|
result = await trigger_ci_build(state, config)
|
|
assert result["ci_build_id"] == 999
|
|
|
|
async def test_appends_error_when_no_pipelines_found(self) -> None:
|
|
clients = build_mock_clients()
|
|
clients.azdo.list_build_pipelines.return_value = []
|
|
config = build_config(clients)
|
|
state = {"repo_name": "my-repo", "version": "v1.0.0"}
|
|
|
|
result = await trigger_ci_build(state, config)
|
|
|
|
assert "errors" in result
|
|
assert len(result["errors"]) >= 1
|
|
|
|
async def test_appends_error_on_trigger_failure(self) -> None:
|
|
from release_agent.exceptions import ServiceError
|
|
|
|
clients = build_mock_clients()
|
|
clients.azdo.list_build_pipelines.return_value = [
|
|
PipelineInfo(id=10, name="CI", repo="my-repo")
|
|
]
|
|
clients.azdo.trigger_pipeline.side_effect = ServiceError(
|
|
service="azdo", status_code=500, detail="Internal error"
|
|
)
|
|
config = build_config(clients)
|
|
state = {"repo_name": "my-repo", "version": "v1.0.0"}
|
|
|
|
result = await trigger_ci_build(state, config)
|
|
|
|
assert "errors" in result
|
|
|
|
async def test_uses_main_branch_when_no_version(self) -> None:
|
|
clients = build_mock_clients()
|
|
clients.azdo.list_build_pipelines.return_value = [
|
|
PipelineInfo(id=10, name="CI", repo="my-repo")
|
|
]
|
|
clients.azdo.trigger_pipeline.return_value = {"id": 1}
|
|
config = build_config(clients)
|
|
state = {"repo_name": "my-repo"}
|
|
|
|
result = await trigger_ci_build(state, config)
|
|
|
|
call_kwargs = clients.azdo.trigger_pipeline.call_args[1]
|
|
branch = call_kwargs.get("branch", "")
|
|
assert "main" in branch or "refs/heads" in branch
|
|
|
|
async def test_appends_message_on_success(self) -> None:
|
|
clients = build_mock_clients()
|
|
clients.azdo.list_build_pipelines.return_value = [
|
|
PipelineInfo(id=10, name="CI", repo="my-repo")
|
|
]
|
|
clients.azdo.trigger_pipeline.return_value = {"id": 123}
|
|
config = build_config(clients)
|
|
state = {"repo_name": "my-repo", "version": "v1.0.0"}
|
|
|
|
result = await trigger_ci_build(state, config)
|
|
|
|
assert "messages" in result
|
|
assert len(result["messages"]) >= 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# poll_ci_build
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPollCiBuild:
|
|
"""Tests for poll_ci_build node."""
|
|
|
|
async def test_returns_ci_build_status_and_result_on_completion(self) -> None:
|
|
clients = build_mock_clients()
|
|
completed_status = BuildStatus(status="completed", result="succeeded", build_url="https://build/1")
|
|
config = build_config(clients)
|
|
state = {"ci_build_id": 42, "repo_name": "my-repo"}
|
|
|
|
with patch(
|
|
"release_agent.graph.ci_nodes.poll_until",
|
|
return_value=(completed_status, True),
|
|
):
|
|
result = await poll_ci_build(state, config)
|
|
|
|
assert result["ci_build_status"] == "completed"
|
|
assert result["ci_build_result"] == "succeeded"
|
|
|
|
async def test_returns_build_url(self) -> None:
|
|
clients = build_mock_clients()
|
|
completed_status = BuildStatus(
|
|
status="completed",
|
|
result="succeeded",
|
|
build_url="https://dev.azure.com/build/42",
|
|
)
|
|
config = build_config(clients)
|
|
state = {"ci_build_id": 42, "repo_name": "my-repo"}
|
|
|
|
with patch(
|
|
"release_agent.graph.ci_nodes.poll_until",
|
|
return_value=(completed_status, True),
|
|
):
|
|
result = await poll_ci_build(state, config)
|
|
|
|
assert result.get("ci_build_url") == "https://dev.azure.com/build/42"
|
|
|
|
async def test_appends_error_on_timeout(self) -> None:
|
|
clients = build_mock_clients()
|
|
running_status = BuildStatus(status="inProgress", result=None, build_url=None)
|
|
config = build_config(clients)
|
|
state = {"ci_build_id": 42, "repo_name": "my-repo"}
|
|
|
|
with patch(
|
|
"release_agent.graph.ci_nodes.poll_until",
|
|
return_value=(running_status, False),
|
|
):
|
|
result = await poll_ci_build(state, config)
|
|
|
|
assert "errors" in result
|
|
|
|
async def test_appends_error_when_build_id_missing(self) -> None:
|
|
clients = build_mock_clients()
|
|
config = build_config(clients)
|
|
state = {"repo_name": "my-repo"} # no ci_build_id
|
|
|
|
result = await poll_ci_build(state, config)
|
|
|
|
assert "errors" in result
|
|
|
|
async def test_passes_correct_build_id_to_poll_fn(self) -> None:
|
|
clients = build_mock_clients()
|
|
clients.azdo.get_build_status.return_value = BuildStatus(
|
|
status="completed", result="succeeded", build_url=None
|
|
)
|
|
config = build_config(clients)
|
|
state = {"ci_build_id": 77, "repo_name": "my-repo"}
|
|
|
|
async def fake_poll_until(*, poll_fn, is_done, interval_seconds, max_wait_seconds, sleep_fn=None):
|
|
result = await poll_fn()
|
|
return result, True
|
|
|
|
with patch("release_agent.graph.ci_nodes.poll_until", side_effect=fake_poll_until):
|
|
await poll_ci_build(state, config)
|
|
|
|
clients.azdo.get_build_status.assert_called_once_with(build_id=77)
|
|
|
|
async def test_result_none_when_poll_returns_none(self) -> None:
|
|
clients = build_mock_clients()
|
|
config = build_config(clients)
|
|
state = {"ci_build_id": 42, "repo_name": "my-repo"}
|
|
|
|
with patch(
|
|
"release_agent.graph.ci_nodes.poll_until",
|
|
return_value=(None, False),
|
|
):
|
|
result = await poll_ci_build(state, config)
|
|
|
|
assert "errors" in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# notify_ci_result
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestNotifyCiResult:
|
|
"""Tests for notify_ci_result node."""
|
|
|
|
async def test_sends_notification_on_success(self) -> None:
|
|
clients = build_mock_clients()
|
|
clients.slack.send_notification.return_value = True
|
|
config = build_config(clients)
|
|
state = {
|
|
"repo_name": "my-repo",
|
|
"ci_build_status": "completed",
|
|
"ci_build_result": "succeeded",
|
|
"ci_build_url": "https://build/99",
|
|
}
|
|
|
|
result = await notify_ci_result(state, config)
|
|
|
|
clients.slack.send_notification.assert_called_once()
|
|
assert "messages" in result
|
|
|
|
async def test_sends_notification_on_failure(self) -> None:
|
|
clients = build_mock_clients()
|
|
clients.slack.send_notification.return_value = True
|
|
config = build_config(clients)
|
|
state = {
|
|
"repo_name": "my-repo",
|
|
"ci_build_status": "completed",
|
|
"ci_build_result": "failed",
|
|
"ci_build_url": None,
|
|
}
|
|
|
|
result = await notify_ci_result(state, config)
|
|
|
|
clients.slack.send_notification.assert_called_once()
|
|
|
|
async def test_handles_slack_error_gracefully(self) -> None:
|
|
from release_agent.exceptions import ServiceError
|
|
|
|
clients = build_mock_clients()
|
|
clients.slack.send_notification.side_effect = ServiceError(
|
|
service="slack", status_code=500, detail="Slack error"
|
|
)
|
|
config = build_config(clients)
|
|
state = {
|
|
"repo_name": "my-repo",
|
|
"ci_build_result": "succeeded",
|
|
"ci_build_url": None,
|
|
}
|
|
|
|
result = await notify_ci_result(state, config)
|
|
|
|
# Should not re-raise; should append error
|
|
assert "errors" in result
|
|
|
|
async def test_includes_repo_name_in_message(self) -> None:
|
|
clients = build_mock_clients()
|
|
clients.slack.send_notification.return_value = True
|
|
config = build_config(clients)
|
|
state = {
|
|
"repo_name": "super-service",
|
|
"ci_build_result": "succeeded",
|
|
"ci_build_url": None,
|
|
}
|
|
|
|
await notify_ci_result(state, config)
|
|
|
|
call_kwargs = clients.slack.send_notification.call_args[1]
|
|
text_or_blocks = str(call_kwargs)
|
|
assert "super-service" in text_or_blocks
|
|
|
|
async def test_returns_empty_dict_when_state_has_no_data(self) -> None:
|
|
clients = build_mock_clients()
|
|
clients.slack.send_notification.return_value = True
|
|
config = build_config(clients)
|
|
state = {}
|
|
|
|
result = await notify_ci_result(state, config)
|
|
|
|
# Should not crash; may return messages or empty dict
|
|
assert isinstance(result, dict)
|