Skip to content

Commit

Permalink
Publish to PR in event file only (#428)
Browse files Browse the repository at this point in the history
  • Loading branch information
@EnricoMi
EnricoMi authored Mar 21, 2023
1 parent 4663c06 commit 763fc57
Show file tree
Hide file tree
Showing 2 changed files with 89 additions and 82 deletions.
22 changes: 16 additions & 6 deletions python/publish/publisher.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
149 changes: 73 additions & 76 deletions python/test/test_publisher.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down

0 comments on commit 763fc57

Please sign in to comment.