import unittest from datetime import UTC, datetime, timedelta from unittest.mock import MagicMock, patch import stale_pr_review.core as spr_core # --------------------------------------------------------------------------- # Fixtures shared across test classes # --------------------------------------------------------------------------- STALE_PR = { "number": 42, "user": {"login": "dev"}, "created_at": "2020-01-01T00:00:00Z", # always in the past → always stale "updated_at": "2020-01-01T00:00:00Z", "labels": [], "html_url": "https://github.com/theorchard/terraform-infra/pull/42", } RECENT_PR = { "number": 99, "user": {"login": "dev"}, "created_at": "2099-01-01T00:00:00Z", # far in the future → never stale "updated_at": "2099-01-01T00:00:00Z", "labels": [], "html_url": "https://github.com/theorchard/terraform-infra/pull/99", } # --------------------------------------------------------------------------- # Unit tests for helper functions # --------------------------------------------------------------------------- class StalePRManagerTests(unittest.TestCase): @patch("stale_pr_review.core.api_get_paginated") def test_last_bot_comment_returns_none_if_no_bot_comment(self, mock_api_get_paginated): mock_api_get_paginated.return_value = [ {"body": "Not a bot comment", "created_at": "2025-09-20T12:00:00Z"} ] self.assertIsNone(spr_core.last_bot_comment(1)) @patch("stale_pr_review.core.api_get_paginated") def test_last_bot_comment_returns_latest_bot_comment(self, mock_api_get_paginated): mock_api_get_paginated.return_value = [ {"body": "Some comment", "created_at": "2025-09-20T12:00:00Z"}, {"body": "[stale\u2011check] Bot comment", "created_at": "2025-09-21T12:00:00Z"}, ] result = spr_core.last_bot_comment(1) expected = datetime.strptime("2025-09-21T12:00:00Z", "%Y-%m-%dT%H:%M:%SZ").replace( tzinfo=UTC ) self.assertEqual(result, expected) def test_has_label_true_and_false(self): pr = {"labels": [{"name": "foo"}, {"name": "bar"}]} self.assertTrue(spr_core.has_label(pr, "foo")) self.assertTrue(spr_core.has_label(pr, "BAR")) self.assertFalse(spr_core.has_label(pr, "baz")) @patch("stale_pr_review.core.DRY_RUN", False) @patch("stale_pr_review.core.api_post") def test_post_comment_calls_api_post(self, mock_api_post): spr_core.post_comment(1, "body text") mock_api_post.assert_called_once_with( "https://api.github.com/repos/theorchard/terraform-infra/issues/1/comments", {"body": "body text"}, ) # --------------------------------------------------------------------------- # Unit tests for members file email lookup # --------------------------------------------------------------------------- class TestGetUserEmail(unittest.TestCase): def setUp(self): spr_core._members_cache = None def tearDown(self): spr_core._members_cache = None @patch("stale_pr_review.core._load_members", return_value={"dev": "dev@company.com"}) def test_returns_email_for_known_user(self, mock_load): self.assertEqual(spr_core.get_user_email("dev"), "dev@company.com") @patch("stale_pr_review.core._load_members", return_value={}) def test_returns_none_for_unknown_user(self, mock_load): self.assertIsNone(spr_core.get_user_email("unknown")) @patch("stale_pr_review.core._load_members", return_value={"dev": "dev@company.com"}) def test_caches_result(self, mock_load): spr_core.get_user_email("dev") spr_core.get_user_email("dev") self.assertEqual(mock_load.call_count, 1) @patch.object(spr_core, "ORG_MEMBERS_B64GZ", "") def test_returns_none_when_env_var_empty(self): self.assertIsNone(spr_core.get_user_email("dev")) # --------------------------------------------------------------------------- # Unit tests for Slack DM # --------------------------------------------------------------------------- class TestSendSlackDm(unittest.TestCase): def _ok_response(self, extra=None): r = MagicMock() r.raise_for_status = lambda: None payload = {"ok": True} if extra: payload.update(extra) r.json.return_value = payload return r @patch("stale_pr_review.core.DRY_RUN", False) @patch("stale_pr_review.core.SLACK_BOT_TOKEN", None) def test_returns_false_when_token_missing(self): self.assertFalse(spr_core.send_slack_dm("dev@company.com", "hello")) @patch("stale_pr_review.core.DRY_RUN", False) @patch("stale_pr_review.core.SLACK_BOT_TOKEN", "xoxb-test") @patch("stale_pr_review.core.requests.post") @patch("stale_pr_review.core.requests.get") def test_happy_path_returns_true(self, mock_get, mock_post): mock_get.return_value = self._ok_response({"user": {"id": "U123"}}) mock_post.side_effect = [ self._ok_response({"channel": {"id": "D456"}}), self._ok_response(), ] self.assertTrue(spr_core.send_slack_dm("dev@company.com", "hello")) self.assertEqual(mock_get.call_count, 1) self.assertEqual(mock_post.call_count, 2) @patch("stale_pr_review.core.DRY_RUN", False) @patch("stale_pr_review.core.SLACK_BOT_TOKEN", "xoxb-test") @patch("stale_pr_review.core.requests.get") def test_returns_false_when_user_not_found(self, mock_get): r = MagicMock() r.raise_for_status = lambda: None r.json.return_value = {"ok": False, "error": "users_not_found"} mock_get.return_value = r self.assertFalse(spr_core.send_slack_dm("noreply@company.com", "hello")) self.assertEqual(mock_get.call_count, 1) @patch("stale_pr_review.core.DRY_RUN", False) @patch("stale_pr_review.core.SLACK_BOT_TOKEN", "xoxb-test") @patch("stale_pr_review.core.requests.get", side_effect=Exception("connection refused")) def test_returns_false_on_exception(self, mock_get): self.assertFalse(spr_core.send_slack_dm("dev@company.com", "hello")) # --------------------------------------------------------------------------- # Unit tests for notify() # --------------------------------------------------------------------------- class TestNotify(unittest.TestCase): @patch("stale_pr_review.core.send_slack_dm") @patch("stale_pr_review.core.get_user_email", return_value="dev@company.com") @patch("stale_pr_review.core.post_comment") @patch("stale_pr_review.core.SLACK_BOT_TOKEN", "xoxb-test") def test_calls_post_comment_and_slack(self, mock_post, mock_email, mock_dm): spr_core.notify(42, "dev", f"{spr_core.BOT_COMMENT_MARKER}\n\nHi @dev, test message.") mock_post.assert_called_once() mock_email.assert_called_once_with("dev") mock_dm.assert_called_once() dm_text = mock_dm.call_args[0][1] self.assertNotIn(spr_core.BOT_COMMENT_MARKER, dm_text) @patch("stale_pr_review.core.send_slack_dm") @patch("stale_pr_review.core.get_user_email", return_value="dev@company.com") @patch("stale_pr_review.core.post_comment") @patch("stale_pr_review.core.SLACK_BOT_TOKEN", "xoxb-test") def test_uses_slack_body_when_provided(self, mock_post, mock_email, mock_dm): slack_body = "Custom slack message with link: https://github.com/org/repo/pull/42" spr_core.notify( 42, "dev", f"{spr_core.BOT_COMMENT_MARKER}\n\nGitHub comment text.", slack_body=slack_body, ) mock_post.assert_called_once() dm_text = mock_dm.call_args[0][1] self.assertEqual(dm_text, slack_body) @patch("stale_pr_review.core.send_slack_dm") @patch("stale_pr_review.core.get_user_email", return_value=None) @patch("stale_pr_review.core.post_comment") @patch("stale_pr_review.core.SLACK_BOT_TOKEN", "xoxb-test") def test_skips_slack_when_no_email(self, mock_post, mock_email, mock_dm): spr_core.notify(42, "dev", "body") mock_post.assert_called_once() mock_dm.assert_not_called() @patch("stale_pr_review.core.send_slack_dm") @patch("stale_pr_review.core.post_comment") @patch("stale_pr_review.core.SLACK_BOT_TOKEN", None) def test_skips_slack_when_bot_token_missing(self, mock_post, mock_dm): spr_core.notify(42, "dev", "body") mock_post.assert_called_once() mock_dm.assert_not_called() # --------------------------------------------------------------------------- # Integration tests for main() / process_pr() # --------------------------------------------------------------------------- class TestMain(unittest.TestCase): def _run_main_with_mocks(self, prs, last_bot_comment_return=None): """Patch the external calls used by main()/process_pr() and run main(). Returns the notify mock for assertions. The simplified flow only depends on staleness, the comment-lease guard, and label opt-out (read from the PR dict directly), so the real `has_label` is used. """ with ( patch("stale_pr_review.core.DRY_RUN", False), patch("stale_pr_review.core.TEST_PR", None), patch("stale_pr_review.core.api_get_paginated", return_value=prs), patch("stale_pr_review.core.last_bot_comment", return_value=last_bot_comment_return), patch("stale_pr_review.core.notify") as mock_notify, ): spr_core.main() return mock_notify def test_main_skips_non_stale_pr(self): mock_notify = self._run_main_with_mocks([RECENT_PR]) mock_notify.assert_not_called() def test_main_notifies_stale_pr_once(self): mock_notify = self._run_main_with_mocks([STALE_PR]) mock_notify.assert_called_once() # positional args: (pr_number, author, body); kwarg: slack_body self.assertEqual(mock_notify.call_args[0][0], STALE_PR["number"]) body = mock_notify.call_args[0][2] self.assertIn("merge or", body) slack_body = mock_notify.call_args[1]["slack_body"] self.assertIn(STALE_PR["html_url"], slack_body) def test_main_notify_message_includes_pr_age_in_days(self): # The message must report how long the PR has been open, computed from # created_at. Use a PR created exactly 40 days ago. created = datetime.now(UTC) - timedelta(days=40) pr = {**STALE_PR, "created_at": created.strftime("%Y-%m-%dT%H:%M:%SZ")} mock_notify = self._run_main_with_mocks([pr]) mock_notify.assert_called_once() body = mock_notify.call_args[0][2] slack_body = mock_notify.call_args[1]["slack_body"] self.assertIn("open for 40 days", body) self.assertIn("open for 40 days", slack_body) def test_main_uses_created_at_not_updated_at(self): # Regression: GitHub bumps updated_at on any activity, including the bot's # own stale-check comment. Staleness must key off created_at so a PR that # is old by creation date is still nudged even when updated_at is recent. pr = { **STALE_PR, "created_at": "2020-01-01T00:00:00Z", # old → stale "updated_at": "2099-01-01T00:00:00Z", # freshly "updated" by bot comment } mock_notify = self._run_main_with_mocks([pr]) mock_notify.assert_called_once() def test_main_skips_dont_merge_label(self): pr = {**STALE_PR, "labels": [{"name": "Dont Merge"}]} mock_notify = self._run_main_with_mocks([pr]) mock_notify.assert_not_called() def test_main_skips_do_not_apply_label(self): pr = {**STALE_PR, "labels": [{"name": "do not apply"}]} mock_notify = self._run_main_with_mocks([pr]) mock_notify.assert_not_called() def test_main_skips_if_bot_commented_recently(self): recent_ts = datetime.now(UTC) - timedelta(days=1) # < 3-day lease mock_notify = self._run_main_with_mocks([STALE_PR], last_bot_comment_return=recent_ts) mock_notify.assert_not_called() def test_main_one_failing_pr_does_not_abort_the_rest(self): # A failure on one PR must be isolated so the rest still get processed. bad_pr = {**STALE_PR, "number": 1} good_pr = {**STALE_PR, "number": 2} def boom_for_pr_1(pr_number): if pr_number == 1: raise RuntimeError("simulated API failure") return with ( patch("stale_pr_review.core.DRY_RUN", False), patch("stale_pr_review.core.TEST_PR", None), patch("stale_pr_review.core.api_get_paginated", return_value=[bad_pr, good_pr]), patch("stale_pr_review.core.last_bot_comment", side_effect=boom_for_pr_1), patch("stale_pr_review.core.notify") as mock_notify, ): spr_core.main() # The good PR (#2) was still notified despite #1 raising. mock_notify.assert_called_once() self.assertEqual(mock_notify.call_args[0][0], 2) if __name__ == "__main__": unittest.main()