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.
474 lines
15 KiB
Python
474 lines
15 KiB
Python
"""Tests for api/slack_interactions.py endpoint.
|
|
|
|
Written FIRST (TDD RED phase).
|
|
|
|
Tests cover:
|
|
- Signature verification (HMAC-SHA256)
|
|
- Payload parsing
|
|
- Button routing
|
|
- _resume_graph invocation
|
|
- Error handling
|
|
"""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import time
|
|
import urllib.parse
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from release_agent.api.slack_interactions import router as slack_interactions_router
|
|
from release_agent.api.slack_interactions import _verify_slack_signature
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_TEST_SIGNING_SECRET = "test-signing-secret-abc"
|
|
|
|
|
|
def _make_slack_signature(*, signing_secret: str, timestamp: str, body: str) -> str:
|
|
"""Compute a valid Slack signing signature."""
|
|
base_string = f"v0:{timestamp}:{body}"
|
|
sig = hmac.new(
|
|
signing_secret.encode(),
|
|
base_string.encode(),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
return f"v0={sig}"
|
|
|
|
|
|
def _make_test_app(
|
|
*,
|
|
signing_secret: str = _TEST_SIGNING_SECRET,
|
|
thread_graph_name: str | None = "release",
|
|
graph_result: dict | None = None,
|
|
) -> FastAPI:
|
|
"""Return a FastAPI test app with mocked state for slack interactions."""
|
|
app = FastAPI()
|
|
app.include_router(slack_interactions_router)
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.slack_signing_secret.get_secret_value.return_value = signing_secret
|
|
mock_settings.operator_token.get_secret_value.return_value = ""
|
|
|
|
mock_graph = MagicMock()
|
|
mock_graph.ainvoke = AsyncMock(return_value=graph_result or {"messages": ["done"]})
|
|
|
|
mock_graphs = {
|
|
"release": mock_graph,
|
|
"pr_completed": MagicMock(),
|
|
}
|
|
mock_clients = MagicMock()
|
|
|
|
mock_pool = MagicMock()
|
|
mock_conn = AsyncMock()
|
|
mock_cursor = AsyncMock()
|
|
mock_cursor.fetchone = AsyncMock(
|
|
return_value=(thread_graph_name,) if thread_graph_name else None
|
|
)
|
|
mock_cursor.__aenter__ = AsyncMock(return_value=mock_cursor)
|
|
mock_cursor.__aexit__ = AsyncMock(return_value=False)
|
|
mock_conn.cursor = MagicMock(return_value=mock_cursor)
|
|
mock_conn.__aenter__ = AsyncMock(return_value=mock_conn)
|
|
mock_conn.__aexit__ = AsyncMock(return_value=False)
|
|
mock_pool.connection = MagicMock(return_value=mock_conn)
|
|
|
|
app.state.settings = mock_settings
|
|
app.state.graphs = mock_graphs
|
|
app.state.tool_clients = mock_clients
|
|
app.state.db_pool = mock_pool
|
|
app.state.background_tasks = set()
|
|
|
|
return app
|
|
|
|
|
|
def _make_button_payload(
|
|
*,
|
|
thread_id: str = "test-thread-123",
|
|
value: str = "approve",
|
|
user_id: str = "U12345",
|
|
user_name: str = "alice",
|
|
) -> str:
|
|
"""Build a URL-encoded Slack button action payload."""
|
|
payload = {
|
|
"type": "block_actions",
|
|
"user": {"id": user_id, "name": user_name},
|
|
"actions": [
|
|
{
|
|
"type": "button",
|
|
"value": f"{thread_id}:{value}",
|
|
"action_id": f"approval_{value}_{thread_id}",
|
|
}
|
|
],
|
|
}
|
|
return urllib.parse.urlencode({"payload": json.dumps(payload)})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _verify_slack_signature pure function tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestVerifySlackSignature:
|
|
"""Tests for the _verify_slack_signature pure function."""
|
|
|
|
def test_returns_true_for_valid_signature(self) -> None:
|
|
timestamp = str(int(time.time()))
|
|
body = "test=body&data=here"
|
|
sig = _make_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=body,
|
|
)
|
|
assert _verify_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=body,
|
|
signature=sig,
|
|
) is True
|
|
|
|
def test_returns_false_for_wrong_secret(self) -> None:
|
|
timestamp = str(int(time.time()))
|
|
body = "test=body"
|
|
sig = _make_slack_signature(
|
|
signing_secret="wrong-secret",
|
|
timestamp=timestamp,
|
|
body=body,
|
|
)
|
|
assert _verify_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=body,
|
|
signature=sig,
|
|
) is False
|
|
|
|
def test_returns_false_for_tampered_body(self) -> None:
|
|
timestamp = str(int(time.time()))
|
|
original_body = "original=body"
|
|
sig = _make_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=original_body,
|
|
)
|
|
assert _verify_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body="tampered=body",
|
|
signature=sig,
|
|
) is False
|
|
|
|
def test_returns_false_for_wrong_timestamp(self) -> None:
|
|
body = "test=body"
|
|
sig = _make_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp="1000000",
|
|
body=body,
|
|
)
|
|
assert _verify_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp="9999999",
|
|
body=body,
|
|
signature=sig,
|
|
) is False
|
|
|
|
def test_returns_false_for_malformed_signature(self) -> None:
|
|
timestamp = str(int(time.time()))
|
|
assert _verify_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body="body",
|
|
signature="not-a-valid-sig",
|
|
) is False
|
|
|
|
def test_returns_false_for_empty_signature(self) -> None:
|
|
timestamp = str(int(time.time()))
|
|
assert _verify_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body="body",
|
|
signature="",
|
|
) is False
|
|
|
|
def test_uses_hmac_sha256(self) -> None:
|
|
timestamp = "1234567890"
|
|
body = "payload=data"
|
|
base = f"v0:{timestamp}:{body}"
|
|
expected_hash = hmac.new(
|
|
_TEST_SIGNING_SECRET.encode(),
|
|
base.encode(),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
sig = f"v0={expected_hash}"
|
|
|
|
# Inject current_time matching timestamp to bypass replay prevention
|
|
assert _verify_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=body,
|
|
signature=sig,
|
|
current_time=1234567890.0,
|
|
) is True
|
|
|
|
def test_rejects_stale_timestamp(self) -> None:
|
|
old_timestamp = "1000000000" # year 2001
|
|
body = "payload=data"
|
|
base = f"v0:{old_timestamp}:{body}"
|
|
expected_hash = hmac.new(
|
|
_TEST_SIGNING_SECRET.encode(),
|
|
base.encode(),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
sig = f"v0={expected_hash}"
|
|
|
|
# Valid signature but timestamp too old
|
|
assert _verify_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=old_timestamp,
|
|
body=body,
|
|
signature=sig,
|
|
) is False
|
|
|
|
def test_rejects_non_integer_timestamp(self) -> None:
|
|
assert _verify_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp="not-a-number",
|
|
body="body",
|
|
signature="v0=abc",
|
|
) is False
|
|
|
|
def test_signature_prefix_must_be_v0(self) -> None:
|
|
timestamp = "1234567890"
|
|
body = "payload=data"
|
|
base = f"v0:{timestamp}:{body}"
|
|
hash_val = hmac.new(
|
|
_TEST_SIGNING_SECRET.encode(),
|
|
base.encode(),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
wrong_prefix_sig = f"v1={hash_val}"
|
|
|
|
assert _verify_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=body,
|
|
signature=wrong_prefix_sig,
|
|
) is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /slack/interactions endpoint tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSlackInteractionsEndpoint:
|
|
"""Tests for POST /slack/interactions."""
|
|
|
|
def test_returns_200_for_valid_request(self) -> None:
|
|
app = _make_test_app()
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
timestamp = str(int(time.time()))
|
|
body = _make_button_payload(thread_id="t-abc", value="approve")
|
|
sig = _make_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=body,
|
|
)
|
|
|
|
response = client.post(
|
|
"/slack/interactions",
|
|
content=body,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"X-Slack-Request-Timestamp": timestamp,
|
|
"X-Slack-Signature": sig,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_returns_403_for_invalid_signature(self) -> None:
|
|
app = _make_test_app()
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
timestamp = str(int(time.time()))
|
|
body = _make_button_payload()
|
|
|
|
response = client.post(
|
|
"/slack/interactions",
|
|
content=body,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"X-Slack-Request-Timestamp": timestamp,
|
|
"X-Slack-Signature": "v0=invalid_signature",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
def test_returns_400_when_missing_timestamp_header(self) -> None:
|
|
app = _make_test_app()
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
body = _make_button_payload()
|
|
|
|
response = client.post(
|
|
"/slack/interactions",
|
|
content=body,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"X-Slack-Signature": "v0=something",
|
|
},
|
|
)
|
|
|
|
assert response.status_code in (400, 403, 422)
|
|
|
|
def test_rejects_when_signing_secret_not_configured(self) -> None:
|
|
app = _make_test_app(signing_secret="")
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
timestamp = str(int(time.time()))
|
|
body = _make_button_payload(thread_id="t-abc", value="approve")
|
|
|
|
response = client.post(
|
|
"/slack/interactions",
|
|
content=body,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"X-Slack-Request-Timestamp": timestamp,
|
|
"X-Slack-Signature": "v0=any_sig",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
|
|
def test_returns_200_with_approve_action(self) -> None:
|
|
app = _make_test_app()
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
timestamp = str(int(time.time()))
|
|
body = _make_button_payload(thread_id="thread-1", value="approve")
|
|
sig = _make_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=body,
|
|
)
|
|
|
|
response = client.post(
|
|
"/slack/interactions",
|
|
content=body,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"X-Slack-Request-Timestamp": timestamp,
|
|
"X-Slack-Signature": sig,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_returns_200_with_reject_action(self) -> None:
|
|
app = _make_test_app()
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
timestamp = str(int(time.time()))
|
|
body = _make_button_payload(thread_id="thread-2", value="reject")
|
|
sig = _make_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=body,
|
|
)
|
|
|
|
response = client.post(
|
|
"/slack/interactions",
|
|
content=body,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"X-Slack-Request-Timestamp": timestamp,
|
|
"X-Slack-Signature": sig,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_schedules_graph_resume_in_background(self) -> None:
|
|
app = _make_test_app()
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
timestamp = str(int(time.time()))
|
|
body = _make_button_payload(thread_id="t-bg", value="approve")
|
|
sig = _make_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=body,
|
|
)
|
|
|
|
response = client.post(
|
|
"/slack/interactions",
|
|
content=body,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"X-Slack-Request-Timestamp": timestamp,
|
|
"X-Slack-Signature": sig,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_returns_404_for_unknown_thread(self) -> None:
|
|
app = _make_test_app(thread_graph_name=None)
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
timestamp = str(int(time.time()))
|
|
body = _make_button_payload(thread_id="unknown-thread", value="approve")
|
|
sig = _make_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=body,
|
|
)
|
|
|
|
response = client.post(
|
|
"/slack/interactions",
|
|
content=body,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"X-Slack-Request-Timestamp": timestamp,
|
|
"X-Slack-Signature": sig,
|
|
},
|
|
)
|
|
|
|
# Should return 200 immediately (Slack requires immediate 200)
|
|
# but the background task may log an error
|
|
assert response.status_code == 200
|
|
|
|
def test_response_body_is_empty_or_ok(self) -> None:
|
|
app = _make_test_app()
|
|
client = TestClient(app, raise_server_exceptions=False)
|
|
|
|
timestamp = str(int(time.time()))
|
|
body = _make_button_payload(thread_id="t-ok", value="approve")
|
|
sig = _make_slack_signature(
|
|
signing_secret=_TEST_SIGNING_SECRET,
|
|
timestamp=timestamp,
|
|
body=body,
|
|
)
|
|
|
|
response = client.post(
|
|
"/slack/interactions",
|
|
content=body,
|
|
headers={
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
"X-Slack-Request-Timestamp": timestamp,
|
|
"X-Slack-Signature": sig,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
# Body may be empty or a simple JSON with ok=True
|
|
if response.content:
|
|
data = response.json()
|
|
assert data.get("ok") is True or "ok" not in data
|