Files
billo-release-agent/tests/test_models.py
Yaojia Wang f5c2733cfb feat: initial commit — Billo Release Agent (LangGraph)
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.
2026-03-24 17:38:23 +01:00

636 lines
23 KiB
Python

"""Tests for Pydantic models. Written FIRST (TDD RED phase)."""
from datetime import date, datetime
import pytest
from pydantic import ValidationError
from release_agent.models.jira import JiraIssue, JiraTransition
from release_agent.models.pipeline import PipelineInfo, ReleasePipelineStage
from release_agent.models.pr import PRInfo
from release_agent.models.release import ArchivedRelease, StagingRelease
from release_agent.models.review import ReviewIssue, ReviewResult
from release_agent.models.ticket import TicketEntry
from release_agent.models.webhook import WebhookPayload, WebhookRepository, WebhookResource
# ---------------------------------------------------------------------------
# PRInfo tests
# ---------------------------------------------------------------------------
class TestPRInfo:
"""Tests for PRInfo model."""
def _make_pr(self, **kwargs) -> PRInfo:
defaults = {
"pr_id": "PR-1",
"pr_url": "https://dev.azure.com/org/project/_git/repo/pullrequest/1",
"repo_name": "my-repo",
"branch": "feature/ALLPOST-100_add-feature",
"pr_title": "Add new feature",
"pr_status": "active",
}
defaults.update(kwargs)
return PRInfo(**defaults)
def test_ticket_id_extracted_from_branch(self) -> None:
pr = self._make_pr(branch="feature/ALLPOST-100_add-feature")
assert pr.ticket_id == "ALLPOST-100"
assert pr.has_ticket is True
def test_branch_without_ticket(self) -> None:
pr = self._make_pr(branch="chore/update-dependencies")
assert pr.ticket_id is None
assert pr.has_ticket is False
def test_main_branch_no_ticket(self) -> None:
pr = self._make_pr(branch="main")
assert pr.ticket_id is None
assert pr.has_ticket is False
def test_refs_heads_branch_parsed(self) -> None:
pr = self._make_pr(branch="refs/heads/fix/BILL-42_fix-bug")
assert pr.ticket_id == "BILL-42"
assert pr.has_ticket is True
def test_pr_status_active(self) -> None:
pr = self._make_pr(pr_status="active")
assert pr.pr_status == "active"
def test_pr_status_completed(self) -> None:
pr = self._make_pr(pr_status="completed")
assert pr.pr_status == "completed"
def test_pr_status_abandoned(self) -> None:
pr = self._make_pr(pr_status="abandoned")
assert pr.pr_status == "abandoned"
def test_invalid_pr_status_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_pr(pr_status="unknown")
def test_model_is_frozen(self) -> None:
pr = self._make_pr()
with pytest.raises(ValidationError):
pr.pr_id = "modified" # type: ignore[misc]
def test_pr_url_is_valid_url(self) -> None:
pr = self._make_pr()
# HttpUrl should have been validated
assert "dev.azure.com" in str(pr.pr_url)
def test_invalid_url_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_pr(pr_url="not-a-url")
# ---------------------------------------------------------------------------
# TicketEntry tests
# ---------------------------------------------------------------------------
class TestTicketEntry:
"""Tests for TicketEntry model."""
def _make_ticket(self, **kwargs) -> TicketEntry:
defaults = {
"id": "ALLPOST-4229",
"summary": "Fix review bug",
"pr_id": "PR-42",
"pr_url": "https://dev.azure.com/org/project/_git/repo/pullrequest/42",
"pr_title": "Fix review",
"branch": "bug/ALLPOST-4229_fix-review",
"merged_at": date(2024, 1, 15),
}
defaults.update(kwargs)
return TicketEntry(**defaults)
def test_valid_ticket_entry(self) -> None:
ticket = self._make_ticket()
assert ticket.id == "ALLPOST-4229"
assert ticket.summary == "Fix review bug"
def test_valid_jira_id_format(self) -> None:
ticket = self._make_ticket(id="BILL-42")
assert ticket.id == "BILL-42"
def test_invalid_id_lowercase_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_ticket(id="allpost-4229")
def test_invalid_id_no_number_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_ticket(id="ALLPOST-")
def test_invalid_id_no_dash_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_ticket(id="ALLPOST4229")
def test_invalid_id_starts_with_number_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_ticket(id="4ALLPOST-4229")
def test_merged_at_is_date(self) -> None:
ticket = self._make_ticket()
assert isinstance(ticket.merged_at, date)
def test_model_is_frozen(self) -> None:
ticket = self._make_ticket()
with pytest.raises(ValidationError):
ticket.id = "OTHER-1" # type: ignore[misc]
def test_minimum_valid_id(self) -> None:
# Single uppercase letter prefix followed by dash and digits
ticket = self._make_ticket(id="A-1")
assert ticket.id == "A-1"
def test_numeric_in_project_key(self) -> None:
ticket = self._make_ticket(id="AB2-100")
assert ticket.id == "AB2-100"
# ---------------------------------------------------------------------------
# StagingRelease tests
# ---------------------------------------------------------------------------
class TestStagingRelease:
"""Tests for StagingRelease model."""
def _make_ticket(self, ticket_id: str = "ALLPOST-1") -> TicketEntry:
return TicketEntry(
id=ticket_id,
summary="Some ticket",
pr_id="PR-1",
pr_url="https://dev.azure.com/org/project/_git/repo/pullrequest/1",
pr_title="Some PR",
branch=f"feature/{ticket_id}_some-feature",
merged_at=date(2024, 1, 15),
)
def _make_release(self, **kwargs) -> StagingRelease:
defaults = {
"version": "v1.0.0",
"repo": "my-repo",
"started_at": date(2024, 1, 1),
"tickets": [],
}
defaults.update(kwargs)
return StagingRelease(**defaults)
def test_valid_release(self) -> None:
release = self._make_release()
assert release.version == "v1.0.0"
def test_version_must_match_pattern(self) -> None:
with pytest.raises(ValidationError):
self._make_release(version="1.0.0")
def test_version_missing_patch_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_release(version="v1.0")
def test_version_extra_segments_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_release(version="v1.0.0.1")
def test_version_letters_in_numbers_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_release(version="v1.a.0")
def test_add_ticket_returns_new_instance(self) -> None:
release = self._make_release()
ticket = self._make_ticket("ALLPOST-1")
new_release = release.add_ticket(ticket)
assert new_release is not release
def test_add_ticket_immutability(self) -> None:
release = self._make_release()
ticket = self._make_ticket("ALLPOST-1")
new_release = release.add_ticket(ticket)
assert len(release.tickets) == 0
assert len(new_release.tickets) == 1
def test_add_ticket_contains_ticket(self) -> None:
release = self._make_release()
ticket = self._make_ticket("ALLPOST-1")
new_release = release.add_ticket(ticket)
assert ticket in new_release.tickets
def test_has_ticket_true(self) -> None:
ticket = self._make_ticket("ALLPOST-1")
release = self._make_release(tickets=[ticket])
assert release.has_ticket("ALLPOST-1") is True
def test_has_ticket_false(self) -> None:
release = self._make_release()
assert release.has_ticket("ALLPOST-99") is False
def test_has_ticket_after_add(self) -> None:
release = self._make_release()
ticket = self._make_ticket("ALLPOST-5")
new_release = release.add_ticket(ticket)
assert new_release.has_ticket("ALLPOST-5") is True
def test_model_is_frozen(self) -> None:
release = self._make_release()
with pytest.raises(ValidationError):
release.version = "v2.0.0" # type: ignore[misc]
def test_multiple_tickets(self) -> None:
t1 = self._make_ticket("ALLPOST-1")
t2 = self._make_ticket("ALLPOST-2")
release = self._make_release(tickets=[t1, t2])
assert len(release.tickets) == 2
# ---------------------------------------------------------------------------
# ArchivedRelease tests
# ---------------------------------------------------------------------------
class TestArchivedRelease:
"""Tests for ArchivedRelease model."""
def _make_archived(self, **kwargs) -> ArchivedRelease:
defaults = {
"version": "v1.0.0",
"repo": "my-repo",
"started_at": date(2024, 1, 1),
"tickets": [],
"released_at": date(2024, 1, 10),
}
defaults.update(kwargs)
return ArchivedRelease(**defaults)
def test_valid_archived_release(self) -> None:
release = self._make_archived()
assert release.released_at == date(2024, 1, 10)
def test_released_at_same_as_started_at_is_valid(self) -> None:
release = self._make_archived(started_at=date(2024, 1, 1), released_at=date(2024, 1, 1))
assert release.released_at == release.started_at
def test_released_at_before_started_at_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_archived(
started_at=date(2024, 1, 10),
released_at=date(2024, 1, 1),
)
def test_model_is_frozen(self) -> None:
release = self._make_archived()
with pytest.raises(ValidationError):
release.released_at = date(2024, 12, 31) # type: ignore[misc]
def test_inherits_version_validation(self) -> None:
with pytest.raises(ValidationError):
self._make_archived(version="1.0.0")
# ---------------------------------------------------------------------------
# PipelineInfo tests
# ---------------------------------------------------------------------------
class TestPipelineInfo:
"""Tests for PipelineInfo model."""
def test_valid_pipeline_info(self) -> None:
pipeline = PipelineInfo(id=42, name="Release Pipeline", repo="my-repo")
assert pipeline.id == 42
assert pipeline.name == "Release Pipeline"
assert pipeline.repo == "my-repo"
def test_model_is_frozen(self) -> None:
pipeline = PipelineInfo(id=1, name="Test", repo="repo")
with pytest.raises(ValidationError):
pipeline.id = 2 # type: ignore[misc]
# ---------------------------------------------------------------------------
# ReleasePipelineStage tests
# ---------------------------------------------------------------------------
class TestReleasePipelineStage:
"""Tests for ReleasePipelineStage model."""
def test_valid_stage_without_approval(self) -> None:
stage = ReleasePipelineStage(
name="Build", rank=0, requires_approval=False, approval_id=None
)
assert stage.name == "Build"
assert stage.rank == 0
def test_valid_stage_with_approval(self) -> None:
stage = ReleasePipelineStage(
name="Production", rank=2, requires_approval=True, approval_id="approval-uuid-123"
)
assert stage.requires_approval is True
assert stage.approval_id == "approval-uuid-123"
def test_negative_rank_raises(self) -> None:
with pytest.raises(ValidationError):
ReleasePipelineStage(
name="Bad", rank=-1, requires_approval=False, approval_id=None
)
def test_requires_approval_false_with_approval_id_raises(self) -> None:
with pytest.raises(ValidationError):
ReleasePipelineStage(
name="Bad", rank=0, requires_approval=False, approval_id="some-id"
)
def test_requires_approval_true_without_approval_id_is_valid(self) -> None:
stage = ReleasePipelineStage(
name="Production", rank=2, requires_approval=True, approval_id=None
)
assert stage.requires_approval is True
assert stage.approval_id is None
def test_model_is_frozen(self) -> None:
stage = ReleasePipelineStage(name="Build", rank=0, requires_approval=False, approval_id=None)
with pytest.raises(ValidationError):
stage.name = "Changed" # type: ignore[misc]
# ---------------------------------------------------------------------------
# WebhookPayload tests
# ---------------------------------------------------------------------------
class TestWebhookPayload:
"""Tests for WebhookPayload and nested models."""
def _make_payload(self, **kwargs) -> WebhookPayload:
defaults = {
"subscription_id": "sub-123",
"event_type": "git.pullrequest.merged",
"resource": {
"repository": {
"id": "repo-uuid-456",
"name": "my-repo",
"web_url": "https://dev.azure.com/org/project/_git/my-repo",
},
"pull_request_id": 42,
"title": "Fix the bug",
"source_ref_name": "refs/heads/bug/ALLPOST-4229_fix-review",
"target_ref_name": "refs/heads/main",
"status": "completed",
"closed_date": None,
},
}
defaults.update(kwargs)
return WebhookPayload(**defaults)
def test_valid_payload(self) -> None:
payload = self._make_payload()
assert payload.subscription_id == "sub-123"
assert payload.event_type == "git.pullrequest.merged"
def test_resource_parsed(self) -> None:
payload = self._make_payload()
assert isinstance(payload.resource, WebhookResource)
assert payload.resource.pull_request_id == 42
assert payload.resource.title == "Fix the bug"
def test_repository_parsed(self) -> None:
payload = self._make_payload()
repo = payload.resource.repository
assert isinstance(repo, WebhookRepository)
assert repo.name == "my-repo"
def test_repository_web_url(self) -> None:
payload = self._make_payload()
assert "dev.azure.com" in str(payload.resource.repository.web_url)
def test_closed_date_none(self) -> None:
payload = self._make_payload()
assert payload.resource.closed_date is None
def test_closed_date_populated(self) -> None:
payload_data = {
"subscription_id": "sub-123",
"event_type": "git.pullrequest.merged",
"resource": {
"repository": {
"id": "repo-uuid-456",
"name": "my-repo",
"web_url": "https://dev.azure.com/org/project/_git/my-repo",
},
"pull_request_id": 42,
"title": "Fix the bug",
"source_ref_name": "refs/heads/bug/ALLPOST-4229_fix-review",
"target_ref_name": "refs/heads/main",
"status": "completed",
"closed_date": "2024-01-15T10:30:00Z",
},
}
payload = WebhookPayload(**payload_data)
assert payload.resource.closed_date is not None
assert isinstance(payload.resource.closed_date, datetime)
def test_model_is_frozen(self) -> None:
payload = self._make_payload()
with pytest.raises(ValidationError):
payload.subscription_id = "changed" # type: ignore[misc]
def test_source_ref_name_preserved(self) -> None:
payload = self._make_payload()
assert payload.resource.source_ref_name == "refs/heads/bug/ALLPOST-4229_fix-review"
# ---------------------------------------------------------------------------
# ReviewIssue tests
# ---------------------------------------------------------------------------
class TestReviewIssue:
"""Tests for ReviewIssue model."""
def _make_issue(self, **kwargs) -> ReviewIssue:
defaults = {
"severity": "warning",
"description": "Variable name is not descriptive",
}
defaults.update(kwargs)
return ReviewIssue(**defaults)
def test_valid_warning_issue(self) -> None:
issue = self._make_issue(severity="warning", description="Unclear variable")
assert issue.severity == "warning"
assert issue.description == "Unclear variable"
def test_valid_error_issue(self) -> None:
issue = self._make_issue(severity="error", description="Null pointer risk")
assert issue.severity == "error"
def test_valid_info_issue(self) -> None:
issue = self._make_issue(severity="info", description="Minor style note")
assert issue.severity == "info"
def test_valid_blocker_issue(self) -> None:
issue = self._make_issue(severity="blocker", description="Security vulnerability")
assert issue.severity == "blocker"
def test_invalid_severity_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_issue(severity="critical")
def test_file_path_optional_none_by_default(self) -> None:
issue = self._make_issue()
assert issue.file_path is None
def test_file_path_can_be_set(self) -> None:
issue = self._make_issue(file_path="src/foo.py")
assert issue.file_path == "src/foo.py"
def test_suggestion_optional_none_by_default(self) -> None:
issue = self._make_issue()
assert issue.suggestion is None
def test_suggestion_can_be_set(self) -> None:
issue = self._make_issue(suggestion="Rename to `user_count`")
assert issue.suggestion == "Rename to `user_count`"
def test_model_is_frozen(self) -> None:
issue = self._make_issue()
with pytest.raises(ValidationError):
issue.severity = "error" # type: ignore[misc]
def test_description_required(self) -> None:
with pytest.raises(ValidationError):
ReviewIssue(severity="warning") # type: ignore[call-arg]
# ---------------------------------------------------------------------------
# ReviewResult tests
# ---------------------------------------------------------------------------
class TestReviewResult:
"""Tests for ReviewResult model."""
def _make_blocker_issue(self) -> ReviewIssue:
return ReviewIssue(severity="blocker", description="Must fix this")
def _make_warning_issue(self) -> ReviewIssue:
return ReviewIssue(severity="warning", description="Minor issue")
def _make_result(self, **kwargs) -> ReviewResult:
defaults = {
"verdict": "approve",
"summary": "Looks good overall",
"issues": [],
}
defaults.update(kwargs)
return ReviewResult(**defaults)
def test_valid_approve_verdict(self) -> None:
result = self._make_result(verdict="approve")
assert result.verdict == "approve"
def test_valid_request_changes_verdict(self) -> None:
result = self._make_result(verdict="request_changes")
assert result.verdict == "request_changes"
def test_invalid_verdict_raises(self) -> None:
with pytest.raises(ValidationError):
self._make_result(verdict="reject")
def test_summary_stored(self) -> None:
result = self._make_result(summary="Great PR")
assert result.summary == "Great PR"
def test_issues_empty_by_default(self) -> None:
result = self._make_result()
assert len(result.issues) == 0
def test_has_blockers_false_with_no_issues(self) -> None:
result = self._make_result(issues=[])
assert result.has_blockers is False
def test_has_blockers_false_with_only_warnings(self) -> None:
result = self._make_result(issues=[self._make_warning_issue()])
assert result.has_blockers is False
def test_has_blockers_true_with_blocker_issue(self) -> None:
result = self._make_result(issues=[self._make_blocker_issue()])
assert result.has_blockers is True
def test_has_blockers_true_mixed_issues(self) -> None:
result = self._make_result(
issues=[self._make_warning_issue(), self._make_blocker_issue()]
)
assert result.has_blockers is True
def test_model_is_frozen(self) -> None:
result = self._make_result()
with pytest.raises(ValidationError):
result.verdict = "request_changes" # type: ignore[misc]
def test_multiple_issues_stored(self) -> None:
issues = [self._make_warning_issue(), self._make_blocker_issue()]
result = self._make_result(issues=issues)
assert len(result.issues) == 2
def test_has_blockers_is_computed(self) -> None:
# Verify has_blockers cannot be set directly (it's computed)
result = self._make_result(issues=[self._make_blocker_issue()])
assert result.has_blockers is True
# ---------------------------------------------------------------------------
# JiraTransition tests
# ---------------------------------------------------------------------------
class TestJiraTransition:
"""Tests for JiraTransition model."""
def test_valid_transition(self) -> None:
transition = JiraTransition(id="11", name="To Do")
assert transition.id == "11"
assert transition.name == "To Do"
def test_model_is_frozen(self) -> None:
transition = JiraTransition(id="11", name="To Do")
with pytest.raises(ValidationError):
transition.id = "22" # type: ignore[misc]
def test_id_required(self) -> None:
with pytest.raises(ValidationError):
JiraTransition(name="To Do") # type: ignore[call-arg]
def test_name_required(self) -> None:
with pytest.raises(ValidationError):
JiraTransition(id="11") # type: ignore[call-arg]
# ---------------------------------------------------------------------------
# JiraIssue tests
# ---------------------------------------------------------------------------
class TestJiraIssue:
"""Tests for JiraIssue model."""
def test_valid_issue(self) -> None:
issue = JiraIssue(key="ALLPOST-100", summary="Fix the bug", status="In Progress")
assert issue.key == "ALLPOST-100"
assert issue.summary == "Fix the bug"
assert issue.status == "In Progress"
def test_model_is_frozen(self) -> None:
issue = JiraIssue(key="ALLPOST-100", summary="Fix the bug", status="In Progress")
with pytest.raises(ValidationError):
issue.key = "ALLPOST-200" # type: ignore[misc]
def test_key_required(self) -> None:
with pytest.raises(ValidationError):
JiraIssue(summary="Fix the bug", status="In Progress") # type: ignore[call-arg]
def test_summary_required(self) -> None:
with pytest.raises(ValidationError):
JiraIssue(key="ALLPOST-100", status="In Progress") # type: ignore[call-arg]
def test_status_required(self) -> None:
with pytest.raises(ValidationError):
JiraIssue(key="ALLPOST-100", summary="Fix the bug") # type: ignore[call-arg]
def test_various_statuses(self) -> None:
statuses = ["To Do", "In Progress", "Done", "Released"]
for status in statuses:
issue = JiraIssue(key="ALLPOST-1", summary="Test", status=status)
assert issue.status == status