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.
199 lines
6.8 KiB
Python
199 lines
6.8 KiB
Python
"""Tests for custom exception hierarchy. Written FIRST (TDD RED phase)."""
|
|
|
|
import pytest
|
|
|
|
from release_agent.exceptions import (
|
|
AuthenticationError,
|
|
NotFoundError,
|
|
RateLimitError,
|
|
ReleaseAgentError,
|
|
ServiceError,
|
|
ServiceUnavailableError,
|
|
)
|
|
|
|
|
|
class TestReleaseAgentError:
|
|
"""Tests for the base exception class."""
|
|
|
|
def test_is_exception(self) -> None:
|
|
err = ReleaseAgentError("something went wrong")
|
|
assert isinstance(err, Exception)
|
|
|
|
def test_message_stored(self) -> None:
|
|
err = ReleaseAgentError("something went wrong")
|
|
assert str(err) == "something went wrong"
|
|
|
|
def test_can_be_raised(self) -> None:
|
|
with pytest.raises(ReleaseAgentError):
|
|
raise ReleaseAgentError("boom")
|
|
|
|
|
|
class TestServiceError:
|
|
"""Tests for ServiceError with service name and status code."""
|
|
|
|
def test_is_release_agent_error(self) -> None:
|
|
err = ServiceError(service="jira", status_code=500, detail="Internal error")
|
|
assert isinstance(err, ReleaseAgentError)
|
|
|
|
def test_stores_service(self) -> None:
|
|
err = ServiceError(service="jira", status_code=500, detail="Internal error")
|
|
assert err.service == "jira"
|
|
|
|
def test_stores_status_code(self) -> None:
|
|
err = ServiceError(service="azdo", status_code=422, detail="Unprocessable")
|
|
assert err.status_code == 422
|
|
|
|
def test_stores_detail(self) -> None:
|
|
err = ServiceError(service="slack", status_code=400, detail="Bad payload")
|
|
assert err.detail == "Bad payload"
|
|
|
|
def test_str_includes_service_and_status(self) -> None:
|
|
err = ServiceError(service="jira", status_code=500, detail="Server error")
|
|
text = str(err)
|
|
assert "jira" in text
|
|
assert "500" in text
|
|
|
|
def test_can_be_raised(self) -> None:
|
|
with pytest.raises(ServiceError):
|
|
raise ServiceError(service="azdo", status_code=400, detail="bad request")
|
|
|
|
def test_detail_none_allowed(self) -> None:
|
|
err = ServiceError(service="jira", status_code=404, detail=None)
|
|
assert err.detail is None
|
|
|
|
|
|
class TestAuthenticationError:
|
|
"""Tests for AuthenticationError (401/403)."""
|
|
|
|
def test_is_service_error(self) -> None:
|
|
err = AuthenticationError(service="azdo")
|
|
assert isinstance(err, ServiceError)
|
|
|
|
def test_default_status_code_401(self) -> None:
|
|
err = AuthenticationError(service="azdo")
|
|
assert err.status_code == 401
|
|
|
|
def test_service_stored(self) -> None:
|
|
err = AuthenticationError(service="jira")
|
|
assert err.service == "jira"
|
|
|
|
def test_custom_status_code(self) -> None:
|
|
err = AuthenticationError(service="azdo", status_code=403)
|
|
assert err.status_code == 403
|
|
|
|
def test_str_contains_service(self) -> None:
|
|
err = AuthenticationError(service="slack")
|
|
assert "slack" in str(err)
|
|
|
|
def test_can_be_raised(self) -> None:
|
|
with pytest.raises(AuthenticationError):
|
|
raise AuthenticationError(service="azdo")
|
|
|
|
def test_caught_as_service_error(self) -> None:
|
|
with pytest.raises(ServiceError):
|
|
raise AuthenticationError(service="azdo")
|
|
|
|
|
|
class TestNotFoundError:
|
|
"""Tests for NotFoundError (404)."""
|
|
|
|
def test_is_service_error(self) -> None:
|
|
err = NotFoundError(service="azdo", detail="PR not found")
|
|
assert isinstance(err, ServiceError)
|
|
|
|
def test_status_code_is_404(self) -> None:
|
|
err = NotFoundError(service="jira", detail="Issue not found")
|
|
assert err.status_code == 404
|
|
|
|
def test_detail_stored(self) -> None:
|
|
err = NotFoundError(service="azdo", detail="PR 999 not found")
|
|
assert "PR 999" in err.detail
|
|
|
|
def test_can_be_raised(self) -> None:
|
|
with pytest.raises(NotFoundError):
|
|
raise NotFoundError(service="azdo", detail="not found")
|
|
|
|
def test_caught_as_release_agent_error(self) -> None:
|
|
with pytest.raises(ReleaseAgentError):
|
|
raise NotFoundError(service="jira", detail="issue missing")
|
|
|
|
|
|
class TestRateLimitError:
|
|
"""Tests for RateLimitError (429) with retry_after."""
|
|
|
|
def test_is_service_error(self) -> None:
|
|
err = RateLimitError(service="jira", retry_after=30)
|
|
assert isinstance(err, ServiceError)
|
|
|
|
def test_status_code_is_429(self) -> None:
|
|
err = RateLimitError(service="jira", retry_after=30)
|
|
assert err.status_code == 429
|
|
|
|
def test_stores_retry_after(self) -> None:
|
|
err = RateLimitError(service="slack", retry_after=60)
|
|
assert err.retry_after == 60
|
|
|
|
def test_retry_after_none_allowed(self) -> None:
|
|
err = RateLimitError(service="azdo", retry_after=None)
|
|
assert err.retry_after is None
|
|
|
|
def test_str_contains_service(self) -> None:
|
|
err = RateLimitError(service="jira", retry_after=5)
|
|
assert "jira" in str(err)
|
|
|
|
def test_can_be_raised(self) -> None:
|
|
with pytest.raises(RateLimitError):
|
|
raise RateLimitError(service="jira", retry_after=30)
|
|
|
|
|
|
class TestServiceUnavailableError:
|
|
"""Tests for ServiceUnavailableError (503)."""
|
|
|
|
def test_is_service_error(self) -> None:
|
|
err = ServiceUnavailableError(service="azdo")
|
|
assert isinstance(err, ServiceError)
|
|
|
|
def test_status_code_is_503(self) -> None:
|
|
err = ServiceUnavailableError(service="azdo")
|
|
assert err.status_code == 503
|
|
|
|
def test_service_stored(self) -> None:
|
|
err = ServiceUnavailableError(service="slack")
|
|
assert err.service == "slack"
|
|
|
|
def test_custom_detail(self) -> None:
|
|
err = ServiceUnavailableError(service="azdo", detail="Maintenance window")
|
|
assert "Maintenance" in err.detail
|
|
|
|
def test_can_be_raised(self) -> None:
|
|
with pytest.raises(ServiceUnavailableError):
|
|
raise ServiceUnavailableError(service="azdo")
|
|
|
|
def test_caught_as_service_error(self) -> None:
|
|
with pytest.raises(ServiceError):
|
|
raise ServiceUnavailableError(service="jira")
|
|
|
|
|
|
class TestExceptionHierarchyInheritance:
|
|
"""Tests verifying the full exception hierarchy is correct."""
|
|
|
|
def test_all_are_release_agent_errors(self) -> None:
|
|
errors = [
|
|
AuthenticationError(service="svc"),
|
|
NotFoundError(service="svc", detail="x"),
|
|
RateLimitError(service="svc", retry_after=1),
|
|
ServiceUnavailableError(service="svc"),
|
|
]
|
|
for err in errors:
|
|
assert isinstance(err, ReleaseAgentError), f"{type(err)} not ReleaseAgentError"
|
|
|
|
def test_all_are_service_errors(self) -> None:
|
|
errors = [
|
|
AuthenticationError(service="svc"),
|
|
NotFoundError(service="svc", detail="x"),
|
|
RateLimitError(service="svc", retry_after=1),
|
|
ServiceUnavailableError(service="svc"),
|
|
]
|
|
for err in errors:
|
|
assert isinstance(err, ServiceError), f"{type(err)} not ServiceError"
|