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.
242 lines
9.1 KiB
Python
242 lines
9.1 KiB
Python
"""Tests for LangGraph state module. Written FIRST (TDD RED phase)."""
|
|
|
|
import json
|
|
|
|
from release_agent.state import ReleaseState, add_errors, add_messages
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ReleaseState tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestReleaseState:
|
|
"""Tests for ReleaseState TypedDict."""
|
|
|
|
def test_empty_state_is_valid(self) -> None:
|
|
# total=False means all fields are optional
|
|
state: ReleaseState = {}
|
|
assert state == {}
|
|
|
|
def test_partial_state_with_repo(self) -> None:
|
|
state: ReleaseState = {"repo_name": "my-repo"}
|
|
assert state["repo_name"] == "my-repo"
|
|
|
|
def test_partial_state_with_messages(self) -> None:
|
|
state: ReleaseState = {"messages": ["Hello"]}
|
|
assert state["messages"] == ["Hello"]
|
|
|
|
def test_partial_state_with_errors(self) -> None:
|
|
state: ReleaseState = {"errors": ["Something went wrong"]}
|
|
assert state["errors"] == ["Something went wrong"]
|
|
|
|
def test_state_with_pr_id(self) -> None:
|
|
state: ReleaseState = {"pr_id": "PR-42"}
|
|
assert state["pr_id"] == "PR-42"
|
|
|
|
def test_state_with_ticket_id(self) -> None:
|
|
state: ReleaseState = {"ticket_id": "ALLPOST-100"}
|
|
assert state["ticket_id"] == "ALLPOST-100"
|
|
|
|
def test_state_with_version(self) -> None:
|
|
state: ReleaseState = {"version": "v1.0.1"}
|
|
assert state["version"] == "v1.0.1"
|
|
|
|
# Phase 3 new fields
|
|
def test_state_with_webhook_payload(self) -> None:
|
|
state: ReleaseState = {"webhook_payload": {"event_type": "git.pullrequest.merged"}}
|
|
assert state["webhook_payload"]["event_type"] == "git.pullrequest.merged"
|
|
|
|
def test_state_with_pr_info(self) -> None:
|
|
state: ReleaseState = {"pr_info": {"pr_id": "42", "repo_name": "my-repo"}}
|
|
assert state["pr_info"]["repo_name"] == "my-repo"
|
|
|
|
def test_state_with_pr_diff(self) -> None:
|
|
state: ReleaseState = {"pr_diff": "edit: src/main.py"}
|
|
assert state["pr_diff"] == "edit: src/main.py"
|
|
|
|
def test_state_with_last_merge_source_commit(self) -> None:
|
|
state: ReleaseState = {"last_merge_source_commit": "abc123"}
|
|
assert state["last_merge_source_commit"] == "abc123"
|
|
|
|
def test_state_with_ticket_summary(self) -> None:
|
|
state: ReleaseState = {"ticket_summary": "Fix login bug"}
|
|
assert state["ticket_summary"] == "Fix login bug"
|
|
|
|
def test_state_with_has_ticket(self) -> None:
|
|
state: ReleaseState = {"has_ticket": True}
|
|
assert state["has_ticket"] is True
|
|
|
|
def test_state_with_review_result(self) -> None:
|
|
state: ReleaseState = {"review_result": {"verdict": "approve", "summary": "LGTM"}}
|
|
assert state["review_result"]["verdict"] == "approve"
|
|
|
|
def test_state_with_review_approved(self) -> None:
|
|
state: ReleaseState = {"review_approved": True}
|
|
assert state["review_approved"] is True
|
|
|
|
def test_state_with_staging(self) -> None:
|
|
state: ReleaseState = {"staging": {"version": "v1.0.0", "tickets": []}}
|
|
assert state["staging"]["version"] == "v1.0.0"
|
|
|
|
def test_state_with_pr_already_merged(self) -> None:
|
|
state: ReleaseState = {"pr_already_merged": False}
|
|
assert state["pr_already_merged"] is False
|
|
|
|
def test_state_with_release_pr_id(self) -> None:
|
|
state: ReleaseState = {"release_pr_id": "123"}
|
|
assert state["release_pr_id"] == "123"
|
|
|
|
def test_state_with_release_pr_commit(self) -> None:
|
|
state: ReleaseState = {"release_pr_commit": "deadbeef"}
|
|
assert state["release_pr_commit"] == "deadbeef"
|
|
|
|
def test_state_with_pipelines(self) -> None:
|
|
state: ReleaseState = {"pipelines": [{"id": 1, "name": "build"}]}
|
|
assert len(state["pipelines"]) == 1
|
|
|
|
def test_state_with_triggered_builds(self) -> None:
|
|
state: ReleaseState = {"triggered_builds": [{"id": 99}]}
|
|
assert state["triggered_builds"][0]["id"] == 99
|
|
|
|
def test_state_with_pending_approvals(self) -> None:
|
|
state: ReleaseState = {"pending_approvals": [{"approval_id": "aaa"}]}
|
|
assert state["pending_approvals"][0]["approval_id"] == "aaa"
|
|
|
|
def test_state_with_continue_to_release(self) -> None:
|
|
state: ReleaseState = {"continue_to_release": True}
|
|
assert state["continue_to_release"] is True
|
|
|
|
# Phase 5: CI/CD and approval fields
|
|
def test_state_with_ci_build_id(self) -> None:
|
|
state: ReleaseState = {"ci_build_id": 12345}
|
|
assert state["ci_build_id"] == 12345
|
|
|
|
def test_state_with_ci_build_status(self) -> None:
|
|
state: ReleaseState = {"ci_build_status": "inProgress"}
|
|
assert state["ci_build_status"] == "inProgress"
|
|
|
|
def test_state_with_ci_build_result(self) -> None:
|
|
state: ReleaseState = {"ci_build_result": "succeeded"}
|
|
assert state["ci_build_result"] == "succeeded"
|
|
|
|
def test_state_with_ci_build_url(self) -> None:
|
|
state: ReleaseState = {"ci_build_url": "https://dev.azure.com/org/proj/_build/results?buildId=99"}
|
|
assert "buildId=99" in state["ci_build_url"]
|
|
|
|
def test_state_with_release_definition_id(self) -> None:
|
|
state: ReleaseState = {"release_definition_id": 7}
|
|
assert state["release_definition_id"] == 7
|
|
|
|
def test_state_with_release_id(self) -> None:
|
|
state: ReleaseState = {"release_id": 456}
|
|
assert state["release_id"] == 456
|
|
|
|
def test_state_with_current_stage(self) -> None:
|
|
state: ReleaseState = {"current_stage": "sandbox_pending"}
|
|
assert state["current_stage"] == "sandbox_pending"
|
|
|
|
def test_state_with_approval_message_ts(self) -> None:
|
|
state: ReleaseState = {"approval_message_ts": "1234567890.123456"}
|
|
assert state["approval_message_ts"] == "1234567890.123456"
|
|
|
|
def test_state_with_slack_message_ts(self) -> None:
|
|
state: ReleaseState = {"slack_message_ts": "9876543210.000001"}
|
|
assert state["slack_message_ts"] == "9876543210.000001"
|
|
|
|
def test_state_json_serializable_empty(self) -> None:
|
|
state: ReleaseState = {}
|
|
serialized = json.dumps(state)
|
|
assert json.loads(serialized) == {}
|
|
|
|
def test_state_json_serializable_with_strings(self) -> None:
|
|
state: ReleaseState = {
|
|
"repo_name": "my-repo",
|
|
"pr_id": "PR-1",
|
|
"ticket_id": "ALLPOST-1",
|
|
"version": "v1.0.0",
|
|
}
|
|
serialized = json.dumps(state)
|
|
loaded = json.loads(serialized)
|
|
assert loaded["repo_name"] == "my-repo"
|
|
assert loaded["pr_id"] == "PR-1"
|
|
|
|
def test_state_json_serializable_with_lists(self) -> None:
|
|
state: ReleaseState = {
|
|
"messages": ["msg1", "msg2"],
|
|
"errors": ["err1"],
|
|
}
|
|
serialized = json.dumps(state)
|
|
loaded = json.loads(serialized)
|
|
assert loaded["messages"] == ["msg1", "msg2"]
|
|
assert loaded["errors"] == ["err1"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reducer tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAddMessages:
|
|
"""Tests for add_messages reducer."""
|
|
|
|
def test_accumulates_to_empty(self) -> None:
|
|
result = add_messages([], ["Hello"])
|
|
assert result == ["Hello"]
|
|
|
|
def test_accumulates_to_existing(self) -> None:
|
|
result = add_messages(["Hello"], ["World"])
|
|
assert result == ["Hello", "World"]
|
|
|
|
def test_accumulates_multiple(self) -> None:
|
|
result = add_messages(["A", "B"], ["C", "D"])
|
|
assert result == ["A", "B", "C", "D"]
|
|
|
|
def test_existing_unchanged(self) -> None:
|
|
existing = ["Hello"]
|
|
add_messages(existing, ["World"])
|
|
# Original should not be mutated
|
|
assert existing == ["Hello"]
|
|
|
|
def test_empty_new_messages(self) -> None:
|
|
result = add_messages(["Hello"], [])
|
|
assert result == ["Hello"]
|
|
|
|
def test_both_empty(self) -> None:
|
|
result = add_messages([], [])
|
|
assert result == []
|
|
|
|
def test_returns_new_list(self) -> None:
|
|
existing = ["Hello"]
|
|
new_msgs = ["World"]
|
|
result = add_messages(existing, new_msgs)
|
|
assert result is not existing
|
|
assert result is not new_msgs
|
|
|
|
|
|
class TestAddErrors:
|
|
"""Tests for add_errors reducer."""
|
|
|
|
def test_accumulates_to_empty(self) -> None:
|
|
result = add_errors([], ["Error occurred"])
|
|
assert result == ["Error occurred"]
|
|
|
|
def test_accumulates_to_existing(self) -> None:
|
|
result = add_errors(["First error"], ["Second error"])
|
|
assert result == ["First error", "Second error"]
|
|
|
|
def test_existing_unchanged(self) -> None:
|
|
existing = ["First error"]
|
|
add_errors(existing, ["Second error"])
|
|
assert existing == ["First error"]
|
|
|
|
def test_empty_new_errors(self) -> None:
|
|
result = add_errors(["Existing"], [])
|
|
assert result == ["Existing"]
|
|
|
|
def test_both_empty(self) -> None:
|
|
result = add_errors([], [])
|
|
assert result == []
|
|
|
|
def test_returns_new_list(self) -> None:
|
|
existing = ["Error"]
|
|
result = add_errors(existing, ["New error"])
|
|
assert result is not existing
|