import html
import io
import json
import tempfile
import unittest
from contextlib import redirect_stdout
from pathlib import Path
from unittest.mock import Mock, patch
import notify
TEST_PRIME_USER_ID = "UPRIME123"
def edge(number: int, author: str = "contributor") -> dict:
return {
"cursor": f"cursor-{number}",
"node": {
"id": f"discussion-{number}",
"number": number,
"title": f"Discussion {number}",
"author": {"login": author},
"category": {"name": "Bug reports"},
},
}
def state() -> dict:
return notify.initial_state({"pageInfo": {"endCursor": "baseline"}})
def report(number: int, author: str = "contributor") -> dict:
return {
**notify.notification_item(edge(number, author)["node"]),
"body": f"Original report {number}\n\nReproduction steps and error output.",
}
def bodies_response(*numbers: int) -> str:
return json.dumps({"data": {"nodes": [report(number) for number in numbers]}})
def trusted_run() -> dict:
return {
"event": "schedule",
"path": notify.STATE_WORKFLOW,
"head_branch": "main",
"head_repository": {"full_name": notify.SOURCE_REPOSITORY},
"conclusion": "failure",
}
class NotificationTests(unittest.TestCase):
def setUp(self):
environment = patch.dict(
"os.environ", {"PRIME_SLACK_USER_ID": TEST_PRIME_USER_ID}
)
environment.start()
self.addCleanup(environment.stop)
def test_single_notification_previews_without_webhook_or_state(self):
response = {"data": {"repository": {"discussion": edge(1)["node"]}}}
with (
patch.object(
notify, "gh", side_effect=[json.dumps(response), bodies_response(1)]
),
patch.object(notify, "deliver") as deliver,
patch.object(notify, "webhook_path") as webhook,
patch.object(notify, "restore_state") as restore,
patch("sys.argv", ["notify.py", "test", "--discussion", "1"]),
redirect_stdout(io.StringIO()) as output,
):
notify.main()
self.assertIn("/discussions/1", output.getvalue())
self.assertIn("Original report 1", output.getvalue())
self.assertIn("Preview only", output.getvalue())
deliver.assert_not_called()
webhook.assert_not_called()
restore.assert_not_called()
def test_single_notification_sends_the_production_payload_once(self):
response = {"data": {"repository": {"discussion": edge(1)["node"]}}}
with (
patch.object(
notify, "gh", side_effect=[json.dumps(response), bodies_response(1)]
),
patch.object(notify, "deliver") as deliver,
patch.object(notify, "webhook_path", return_value="/services/test"),
patch.object(notify, "restore_state") as restore,
patch.dict("os.environ", {"DRY_RUN": "false"}),
patch("sys.argv", ["notify.py", "test", "--discussion", "1", "--send"]),
redirect_stdout(io.StringIO()) as output,
):
notify.main()
deliver.assert_called_once_with(report(1), "/services/test")
restore.assert_not_called()
self.assertIn("production state is unchanged", output.getvalue())
def test_single_notification_respects_dry_run(self):
with (
patch.dict("os.environ", {"DRY_RUN": "true"}),
patch.object(notify, "gh") as gh,
patch.object(notify, "deliver") as deliver,
):
with self.assertRaisesRegex(RuntimeError, "disabled during a dry run"):
notify.test_notification(1, send=True)
gh.assert_not_called()
deliver.assert_not_called()
def test_single_notification_rejects_missing_or_invalid_discussion(self):
with (
patch.object(
notify,
"gh",
return_value=json.dumps(
{
"data": {"repository": {"discussion": None}},
}
),
) as gh,
patch.object(notify, "deliver") as deliver,
):
with self.assertRaisesRegex(RuntimeError, "positive discussion number"):
notify.test_notification(0, send=False)
gh.assert_not_called()
with self.assertRaisesRegex(RuntimeError, "Cannot read discussion #1"):
notify.test_notification(1, send=False)
deliver.assert_not_called()
def test_first_run_does_not_backfill(self):
with (
tempfile.TemporaryDirectory() as directory,
patch.object(notify, "restore_state", return_value=None),
patch.object(
notify,
"fetch_page",
return_value={
"edges": [edge(1)],
"pageInfo": {"endCursor": "cursor-1"},
},
),
patch.object(notify, "deliver") as deliver,
):
notify.plan(Path(directory), dry_run=True)
saved = json.loads((Path(directory) / "state.json").read_text())
self.assertEqual(saved["cursor"], "cursor-1")
self.assertEqual(saved["pending"], [])
self.assertEqual(
json.loads((Path(directory) / "outbox.json").read_text()), []
)
deliver.assert_not_called()
def test_plan_fetches_selected_bodies_without_saving_them_in_state(self):
current = state()
with (
tempfile.TemporaryDirectory() as directory,
patch.object(notify, "restore_state", return_value=current),
patch.object(
notify,
"fetch_page",
return_value={"edges": [edge(i) for i in (1, 2, 3)]},
),
patch.object(notify, "gh", return_value=bodies_response(2, 1)) as gh,
):
notify.plan(Path(directory), dry_run=True)
saved = json.loads((Path(directory) / "state.json").read_text())
outbox = json.loads((Path(directory) / "outbox.json").read_text())
self.assertEqual(outbox, [report(1), report(2)])
self.assertEqual([item["number"] for item in saved["pending"]], [3])
self.assertNotIn('"body"', json.dumps(saved))
self.assertEqual(
gh.call_args.kwargs["payload"]["variables"]["ids"],
["discussion-1", "discussion-2"],
)
def test_missing_body_does_not_checkpoint_reservations(self):
for response in [
{"data": {"nodes": [None]}},
{"data": {"nodes": [{"id": "discussion-1", "body": None}]}},
{"data": {"nodes": []}, "errors": [{"message": "unavailable"}]},
]:
with (
self.subTest(response=response),
tempfile.TemporaryDirectory() as directory,
patch.object(notify, "restore_state", return_value=state()),
patch.object(notify, "fetch_page", return_value={"edges": [edge(1)]}),
patch.object(notify, "gh", return_value=json.dumps(response)),
):
with self.assertRaisesRegex(RuntimeError, "state was not advanced"):
notify.plan(Path(directory), dry_run=True)
self.assertFalse((Path(directory) / "state.json").exists())
self.assertFalse((Path(directory) / "outbox.json").exists())
def test_empty_selection_does_not_fetch_bodies(self):
with patch.object(notify, "gh") as gh:
self.assertEqual(notify.fetch_report_bodies([]), [])
gh.assert_not_called()
def test_per_run_limit_preserves_pending_items(self):
current = state()
notify.enqueue(
current, {"edges": [edge(i, f"author-{i}") for i in range(1, 9)]}
)
first = notify.reserve(current, 100_000)
self.assertEqual(len(first), 5)
self.assertEqual(len(current["pending"]), 3)
second = notify.reserve(current, 100_001)
self.assertEqual(len(second), 3)
self.assertFalse(
{item["id"] for item in first} & {item["id"] for item in second}
)
def test_author_limit_cannot_be_bypassed_with_case(self):
current = state()
notify.enqueue(
current,
{
"edges": [
edge(1, "Alice"),
edge(2, "alice"),
edge(3, "ALICE"),
edge(4, "bob"),
]
},
)
self.assertEqual(
[item["number"] for item in notify.reserve(current, 100_000)], [1, 2, 4]
)
self.assertEqual([item["number"] for item in current["pending"]], [3])
self.assertEqual(notify.reserve(current, 100_001), [])
self.assertEqual(
[item["number"] for item in notify.reserve(current, 186_400)], [3]
)
def test_rolling_daily_limit_across_runs(self):
current = state()
notify.enqueue(
current, {"edges": [edge(i, f"author-{i}") for i in range(1, 26)]}
)
selected = []
for offset in range(6):
selected.extend(notify.reserve(current, 100_000 + offset))
self.assertEqual(len(selected), 20)
self.assertEqual(len(current["pending"]), 5)
self.assertEqual(notify.reserve(current, 186_399), [])
self.assertEqual(len(notify.reserve(current, 186_405)), 5)
def test_duplicate_pages_do_not_repeat_mentions(self):
current = state()
page = {"edges": [edge(1), edge(2)]}
notify.enqueue(current, page)
notify.enqueue(current, page)
self.assertEqual(len(notify.reserve(current, 100_000)), 2)
notify.enqueue(current, page)
self.assertEqual(notify.reserve(current, 100_001), [])
def test_full_queue_does_not_advance_past_unread_discussions(self):
current = state()
notify.enqueue(
current, {"edges": [edge(i) for i in range(1, notify.MAX_PENDING + 2)]}
)
self.assertEqual(len(current["pending"]), notify.MAX_PENDING)
self.assertEqual(current["cursor"], f"cursor-{notify.MAX_PENDING}")
def test_missing_author_is_handled(self):
current = state()
item = edge(1)
item["node"]["author"] = None
notify.enqueue(current, {"edges": [item]})
self.assertEqual(current["pending"][0]["author"], "deleted-user")
def test_invalid_discussion_number_is_rejected(self):
current = state()
item = edge(1)
item["node"]["number"] = "1?redirect=evil"
with self.assertRaises(RuntimeError):
notify.enqueue(current, {"edges": [item]})
self.assertEqual(current["cursor"], "baseline")
def test_message_contains_full_report_and_only_one_real_mention(self):
item = report(1, "<@UOTHER>")
item.update(
{
"title": "<!channel> <@UOTHER> & " + "a" * 500,
"body": (
"## Report\n\n<!channel> <@UOTHER> <@" + TEST_PRIME_USER_ID + ">"
"\n\n```typescript\nconst a = b < 5 && c > 0;\n```\n" + "x" * 4_000
),
}
)
payload = notify.slack_payload(item)
encoded = json.dumps(payload)
self.assertEqual(encoded.count("<@"), 1)
self.assertNotIn("<!channel>", encoded)
self.assertEqual(html.unescape(payload["attachments"][0]["text"]), item["body"])
self.assertIn("untrusted evidence", payload["text"])
self.assertEqual(
payload["attachments"][0]["title_link"],
"https://github.com/PrimeIntellect-ai/prime-agent/discussions/1",
)
self.assertNotIn("thread_ts", payload)
self.assertFalse(payload["unfurl_links"])
def test_oversized_report_requests_manual_review_without_a_mention(self):
item = report(1)
item["body"] = "&" * (notify.MAX_REPORT_CHARS // 5)
payload = notify.slack_payload(item)
self.assertIn(f"<@{TEST_PRIME_USER_ID}>", payload["text"])
self.assertEqual(html.unescape(payload["attachments"][0]["text"]), item["body"])
item["body"] += "&"
payload = notify.slack_payload(item)
self.assertNotIn("<@", json.dumps(payload))
self.assertIn("manual review", payload["text"])
self.assertIn("no partial report", payload["attachments"][0]["text"])
self.assertTrue(
payload["attachments"][0]["title_link"].endswith("/discussions/1")
)
def test_bot_configuration_cannot_inject_mentions(self):
for value in ["", "prime", "<!channel>", "UPRIME123> <@UOTHER", "U123\n"]:
with (
self.subTest(value=value),
patch.dict("os.environ", {"PRIME_SLACK_USER_ID": value}),
):
with self.assertRaisesRegex(RuntimeError, "PRIME_SLACK_USER_ID"):
notify.slack_payload(report(1))
def test_missing_bot_configuration_does_not_reserve_discussions(self):
with (
patch.dict("os.environ", {"PRIME_SLACK_USER_ID": ""}),
patch.object(notify, "webhook_path", return_value="/services/test"),
patch.object(notify, "restore_state") as restore,
):
with self.assertRaisesRegex(RuntimeError, "PRIME_SLACK_USER_ID"):
notify.plan(Path("unused"), dry_run=False)
restore.assert_not_called()
def test_webhook_is_restricted_to_slack_incoming_webhooks(self):
for url in [
"",
"https://example.com/services/test",
"http://hooks.slack.com/services/test",
"https://hooks.slack.com/triggers/test",
"https://hooks.slack.com/services/test?token=secret",
]:
with (
self.subTest(url=url),
patch.dict("os.environ", {"SLACK_WEBHOOK_URL": url}),
):
with self.assertRaisesRegex(
RuntimeError, "Configure SLACK_DISCUSSION_WEBHOOK_URL"
):
notify.webhook_path()
with patch.dict(
"os.environ", {"SLACK_WEBHOOK_URL": "https://hooks.slack.com/services/test"}
):
self.assertEqual(notify.webhook_path(), "/services/test")
def test_ambiguous_slack_failure_is_not_retried(self):
current = state()
notify.enqueue(current, {"edges": [edge(1)]})
item = notify.reserve(current, 100_000)[0]
item["body"] = report(1)["body"]
with patch.object(notify.http.client, "HTTPSConnection") as connection:
connection.return_value.getresponse.side_effect = TimeoutError(
"sensitive endpoint"
)
with self.assertRaisesRegex(
RuntimeError, "delivery for discussion #1 is uncertain"
):
notify.deliver(item, "/services/secret")
connection.return_value.request.assert_called_once()
connection.return_value.close.assert_called_once()
self.assertEqual(notify.reserve(current, 100_001), [])
def test_slack_requires_success_status_and_body(self):
item = report(1)
for status, body, succeeds in [
(200, b"ok", True),
(200, b"invalid_payload", False),
(429, b"rate_limited", False),
]:
with (
self.subTest(status=status, body=body),
patch.object(notify.http.client, "HTTPSConnection") as connection,
):
connection.return_value.getresponse.return_value = Mock(
status=status, read=Mock(return_value=body)
)
if succeeds:
notify.deliver(item, "/services/test")
else:
with self.assertRaises(RuntimeError):
notify.deliver(item, "/services/test")
connection.return_value.request.assert_called_once()
def test_restores_latest_trusted_reservation_by_creation_time(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory)
saved = state()
(path / "state.json").write_text(json.dumps(saved))
artifacts = [
{
"id": artifact_id,
"expired": False,
"created_at": created_at,
"workflow_run": {"id": run_id, "head_branch": branch},
}
for artifact_id, run_id, branch, created_at in [
(300, 10, "main", "2026-09-12T01:00:00Z"),
(100, 30, "feature/test", "2026-09-12T03:00:00Z"),
(200, 20, "main", "2026-09-12T02:00:00Z"),
]
]
with patch.object(
notify,
"gh",
side_effect=[
json.dumps({"artifacts": artifacts}),
json.dumps(trusted_run()),
"",
],
) as gh:
self.assertEqual(notify.restore_state(path), saved)
self.assertIn("20", gh.call_args.args)
def test_untrusted_builds_cannot_replace_notification_state(self):
for change in [
{"event": "pull_request"},
{"event": "pull_request_target"},
{"path": ".github/workflows/other.yml"},
{"head_repository": {"full_name": "contributor/prime-agent"}},
{"head_branch": "feature/test"},
]:
with (
self.subTest(change=change),
tempfile.TemporaryDirectory() as directory,
):
path = Path(directory)
saved = state()
(path / "state.json").write_text(json.dumps(saved))
artifacts = [
{
"id": i,
"expired": False,
"created_at": f"2026-09-12T00:0{i}:00Z",
"workflow_run": {"id": i * 10, "head_branch": "main"},
}
for i in [1, 2]
]
with patch.object(
notify,
"gh",
side_effect=[
json.dumps({"artifacts": artifacts}),
json.dumps({**trusted_run(), **change}),
json.dumps({**trusted_run(), "event": "workflow_dispatch"}),
"",
],
) as gh:
self.assertEqual(notify.restore_state(path), saved)
downloads = [
call for call in gh.call_args_list if call.args[0] == "run"
]
self.assertEqual(len(downloads), 1)
self.assertEqual(downloads[0].args[2], "10")
def test_expired_state_is_not_silently_reset(self):
artifact = {
"id": 1,
"expired": True,
"created_at": "2026-09-12T00:01:00Z",
"workflow_run": {"id": 10, "head_branch": "main"},
}
with patch.object(
notify,
"gh",
side_effect=[
json.dumps({"artifacts": [artifact]}),
json.dumps(trusted_run()),
],
):
with self.assertRaisesRegex(RuntimeError, "state expired"):
notify.restore_state(Path("unused"))
if __name__ == "__main__":
unittest.main()