Publish to PR in event file only

python/publish/publisher.py

@@ -206,23 +206,33 @@ def publish(self,
         else:
             logger.info('Commenting on pull requests disabled')
 
+    def get_pull_from_event(self) -> Optional[PullRequest]:
+        number = self._settings.event.get('pull_request', {}).get('number')
+        repo = self._settings.event.get('pull_request', {}).get('base', {}).get('repo', {}).get('full_name')
+        if number is None or repo is None or repo != self._settings.repo:
+            return None
+
+        try:
+            return self._repo.get_pull(number)
+        except UnknownObjectException:
+            return None
+
     def get_all_pulls(self, commit: str) -> List[PullRequest]:
-        # totalCount of PaginatedList calls the GitHub API just to get the total number
-        # we have to retrieve them all anyway so better do this once by materialising the PaginatedList via list()
         if self._settings.search_pull_requests:
+            # totalCount of PaginatedList calls the GitHub API just to get the total number
+            # we have to retrieve them all anyway so better do this once by materialising the PaginatedList via list()
             issues = list(self._gh.search_issues(f'type:pr repo:"{self._settings.repo}" {commit}'))
             pull_requests = [issue.as_pull_request() for issue in issues]
         else:
-            try:
-                pull_requests = list(self._repo.get_commit(commit).get_pulls())
-            except UnknownObjectException:
-                pull_requests = []
+            pull_request = self.get_pull_from_event()
+            pull_requests = [pull_request] if pull_request is not None else []
 
         logger.debug(f'found {len(pull_requests)} pull requests in repo {self._settings.repo} containing commit {commit}')
         return pull_requests
 
     def get_pulls(self, commit: str) -> List[PullRequest]:
         # get all pull requests associated with this commit
+        # TODO: simplify to event pr only, breaking change for version 3.0
         pull_requests = self.get_all_pulls(commit)
 
         if logger.isEnabledFor(logging.DEBUG):

python/test/test_publisher.py

@@ -832,139 +832,136 @@ def test_reuse_comment_existing_updated(self):
         self.do_test_reuse_comment(earlier_body='comment already updated\n:recycle: Has been updated',
                                    expected_body='comment already updated\n:recycle: Has been updated')
 
+    def test_get_pull_from_event(self):
+        settings = self.create_settings()
+        gh, gha, req, repo, commit = self.create_mocks()
+        pr = self.create_github_pr(settings.repo, head_commit_sha=settings.commit)
+        repo.get_pull = mock.Mock(return_value=pr)
+
+        publisher = Publisher(settings, gh, gha)
+
+        actual = publisher.get_pull_from_event()
+        self.assertIsNone(actual)
+        repo.get_pull.assert_not_called()
+
+        # test with pull request in event file
+        settings = self.create_settings(event={'pull_request': {'number': 1234, 'base': {'repo': {'full_name': 'owner/repo'}}}})
+        publisher = Publisher(settings, gh, gha)
+
+        actual = publisher.get_pull_from_event()
+        self.assertIs(actual, pr)
+        repo.get_pull.assert_called_once_with(1234)
+
     def do_test_get_pulls(self,
-                         settings: Settings,
-                         pull_requests: mock.Mock,
-                         expected: List[mock.Mock]) -> mock.Mock:
+                          settings: Settings,
+                          pull_requests: mock.Mock,
+                          event_pull_request: Optional[mock.Mock],
+                          expected: List[mock.Mock]) -> mock.Mock:
         gh, gha, req, repo, commit = self.create_mocks()
 
         gh.search_issues = mock.Mock(return_value=pull_requests)
         commit.get_pulls = mock.Mock(return_value=pull_requests)
+        if event_pull_request is not None:
+            repo.get_pull = mock.Mock(return_value=event_pull_request)
 
         publisher = Publisher(settings, gh, gha)
 
         actual = publisher.get_pulls(settings.commit)
         self.assertEqual(expected, actual)
         if settings.search_pull_requests:
             gh.search_issues.assert_called_once_with('type:pr repo:"{}" {}'.format(settings.repo, settings.commit))
-            commit.get_pulls.assert_not_called()
         else:
             gh.search_issues.assert_not_called()
-            commit.get_pulls.assert_called_once_with()
+            if event_pull_request is not None and \
+                    settings.repo == settings.event.get('pull_request', {}).get('base', {}).get('repo', {}).get('full_name'):
+                repo.get_pull.assert_called_once_with(event_pull_request.number)
+            else:
+                repo.get_pull.assert_not_called()
+        commit.get_pulls.assert_not_called()
         return gha
 
-    def test_get_pulls(self):
+    def test_get_pulls_without_event(self):
         settings = self.create_settings()
         pr = self.create_github_pr(settings.repo, head_commit_sha=settings.commit)
         pull_requests = self.create_github_collection([pr])
-        gha = self.do_test_get_pulls(settings, pull_requests, [pr])
+        gha = self.do_test_get_pulls(settings, pull_requests, None, [])
         gha.warning.assert_not_called()
         gha.error.assert_not_called()
 
-    def test_get_pulls_no_search_results(self):
-        settings = self.create_settings()
-        pull_requests = self.create_github_collection([])
-        gha = self.do_test_get_pulls(settings, pull_requests, [])
+    def test_get_pulls_with_other_event_pr(self):
+        settings = self.create_settings(event={'pull_request': {'number': 1234, 'base': {'repo': {'full_name': 'owner/repo'}}}})
+        event_pr = self.create_github_pr(settings.repo, head_commit_sha=settings.commit, number=1234)
+        pr = self.create_github_pr(settings.repo, head_commit_sha=settings.commit, number=5678)
+        pull_requests = self.create_github_collection([pr])
+        gha = self.do_test_get_pulls(settings, pull_requests, event_pr, [event_pr])
         gha.warning.assert_not_called()
         gha.error.assert_not_called()
 
-    def test_get_pulls_one_closed_matches(self):
-        settings = self.create_settings()
-
-        pr = self.create_github_pr(settings.repo, state='closed', head_commit_sha=settings.commit)
+    def test_get_pulls_with_other_repo_event_pr(self):
+        settings = self.create_settings(event={'pull_request': {'number': 1234, 'base': {'repo': {'full_name': 'fork/repo'}}}})
+        event_pr = self.create_github_pr(settings.repo, head_commit_sha=settings.commit, number=1234)
+        pr = self.create_github_pr(settings.repo, head_commit_sha=settings.commit, number=5678)
         pull_requests = self.create_github_collection([pr])
-
-        gha = self.do_test_get_pulls(settings, pull_requests, [])
+        gha = self.do_test_get_pulls(settings, pull_requests, event_pr, [])
         gha.warning.assert_not_called()
         gha.error.assert_not_called()
 
-    def test_get_pulls_multiple_closed_matches(self):
-        settings = self.create_settings()
-
-        pr1 = self.create_github_pr(settings.repo, state='closed', head_commit_sha=settings.commit)
-        pr2 = self.create_github_pr(settings.repo, state='closed', head_commit_sha=settings.commit)
-        pull_requests = self.create_github_collection([pr1, pr2])
-
-        gha = self.do_test_get_pulls(settings, pull_requests, [])
+    def test_get_pulls_only_with_event_pr(self):
+        settings = self.create_settings(event={'pull_request': {'number': 1234, 'base': {'repo': {'full_name': 'owner/repo'}}}})
+        pr = self.create_github_pr(settings.repo, head_commit_sha=settings.commit, number=1234)
+        pull_requests = self.create_github_collection([])
+        gha = self.do_test_get_pulls(settings, pull_requests, pr, [pr])
         gha.warning.assert_not_called()
         gha.error.assert_not_called()
 
-    def test_get_pulls_one_closed_one_open_matches(self):
+    def test_get_pulls_no_pulls(self):
         settings = self.create_settings()
-
-        pr1 = self.create_github_pr(settings.repo, state='closed', head_commit_sha=settings.commit)
-        pr2 = self.create_github_pr(settings.repo, state='open', head_commit_sha=settings.commit)
-        pull_requests = self.create_github_collection([pr1, pr2])
-
-        gha = self.do_test_get_pulls(settings, pull_requests, [pr2])
+        pull_requests = self.create_github_collection([])
+        gha = self.do_test_get_pulls(settings, pull_requests, None, [])
         gha.warning.assert_not_called()
         gha.error.assert_not_called()
 
-    def test_get_pulls_multiple_open_one_matches_head_commit(self):
-        settings = self.create_settings()
-
-        pr1 = self.create_github_pr(settings.repo, state='open', head_commit_sha=settings.commit, merge_commit_sha='merge one')
-        pr2 = self.create_github_pr(settings.repo, state='open', head_commit_sha='other head commit', merge_commit_sha='merge two')
-        pull_requests = self.create_github_collection([pr1, pr2])
-
-        gha = self.do_test_get_pulls(settings, pull_requests, [pr1])
+    def test_get_pulls_closed_pull(self):
+        settings = self.create_settings(event={'pull_request': {'number': 1234, 'base': {'repo': {'full_name': 'owner/repo'}}}})
+        pr = self.create_github_pr(settings.repo, state='closed', head_commit_sha=settings.commit, number=1234)
+        pull_requests = self.create_github_collection([])
+        gha = self.do_test_get_pulls(settings, pull_requests, pr, [])
         gha.warning.assert_not_called()
         gha.error.assert_not_called()
 
-    def test_get_pulls_multiple_open_one_matches_merge_commit(self):
-        settings = self.create_settings()
-
-        pr1 = self.create_github_pr(settings.repo, state='open', head_commit_sha='one head commit', merge_commit_sha=settings.commit)
-        pr2 = self.create_github_pr(settings.repo, state='open', head_commit_sha='two head commit', merge_commit_sha='other merge commit')
-        pull_requests = self.create_github_collection([pr1, pr2])
-
-        gha = self.do_test_get_pulls(settings, pull_requests, [pr1])
+    def test_get_pulls_head_commit(self):
+        settings = self.create_settings(event={'pull_request': {'number': 1234, 'base': {'repo': {'full_name': 'owner/repo'}}}})
+        pr = self.create_github_pr(settings.repo, state='open', head_commit_sha=settings.commit, merge_commit_sha='merge', number=1234)
+        pull_requests = self.create_github_collection([])
+        gha = self.do_test_get_pulls(settings, pull_requests, pr, [pr])
         gha.warning.assert_not_called()
         gha.error.assert_not_called()
 
-    def test_get_pulls_multiple_open_both_match_head_commit(self):
-        settings = self.create_settings()
-
-        pr1 = self.create_github_pr(settings.repo, state='open', head_commit_sha=settings.commit, merge_commit_sha='merge one')
-        pr2 = self.create_github_pr(settings.repo, state='open', head_commit_sha=settings.commit, merge_commit_sha='merge two')
-        pull_requests = self.create_github_collection([pr1, pr2])
+    def test_get_pulls_merge_commit(self):
+        settings = self.create_settings(event={'pull_request': {'number': 1234, 'base': {'repo': {'full_name': 'owner/repo'}}}})
+        pr1 = self.create_github_pr(settings.repo, state='open', head_commit_sha='one head commit', merge_commit_sha=settings.commit, number=1234)
+        pr2 = self.create_github_pr(settings.repo, state='open', head_commit_sha='two head commit', merge_commit_sha='other merge commit', number=1234)
+        pull_requests = self.create_github_collection([])
 
-        gha = self.do_test_get_pulls(settings, pull_requests, [pr1, pr2])
+        gha = self.do_test_get_pulls(settings, pull_requests, pr1, [pr1])
         gha.warning.assert_not_called()
         gha.error.assert_not_called()
 
-    def test_get_pulls_multiple_open_both_match_merge_commit(self):
-        settings = self.create_settings()
-
-        pr1 = self.create_github_pr(settings.repo, state='open', head_commit_sha='one head commit', merge_commit_sha=settings.commit)
-        pr2 = self.create_github_pr(settings.repo, state='open', head_commit_sha='two head commit', merge_commit_sha=settings.commit)
-        pull_requests = self.create_github_collection([pr1, pr2])
-
-        gha = self.do_test_get_pulls(settings, pull_requests, [pr1, pr2])
+        gha = self.do_test_get_pulls(settings, pull_requests, pr2, [])
         gha.warning.assert_not_called()
         gha.error.assert_not_called()
 
     def test_get_pulls_forked_repo(self):
         settings = self.create_settings()
         fork = self.create_github_pr('other/fork', head_commit_sha=settings.commit)
-        pull_requests = self.create_github_collection([fork])
-        self.do_test_get_pulls(settings, pull_requests, [])
-
-    def test_get_pulls_forked_repos_and_own_repo(self):
-        settings = self.create_settings()
-
-        own = self.create_github_pr(settings.repo, head_commit_sha=settings.commit)
-        fork1 = self.create_github_pr('other/fork', head_commit_sha=settings.commit)
-        fork2 = self.create_github_pr('{}.fork'.format(settings.repo), head_commit_sha=settings.commit)
-        pull_requests = self.create_github_collection([own, fork1, fork2])
-
-        self.do_test_get_pulls(settings, pull_requests, [own])
+        pull_requests = self.create_github_collection([])
+        self.do_test_get_pulls(settings, pull_requests, fork, [])
 
     def test_get_pulls_via_search(self):
         settings = self.create_settings(search_pull_requests=True)
         pr = self.create_github_pr(settings.repo, head_commit_sha=settings.commit)
         search_issues = self.create_github_collection([pr])
-        gha = self.do_test_get_pulls(settings, search_issues, [pr])
+        gha = self.do_test_get_pulls(settings, search_issues, None, [pr])
         gha.warning.assert_not_called()
         gha.error.assert_not_called()