def test__when_comment_data_is_valid__should_send_it_to_proper_github_endpoint(self):
proper_data = {'repository': 'repo', 'sha': '123abc', 'body': 'Comment body', 'jobName': 'job'}
request = mock({'headers': {'Authorization': f'Token {self.API_TOKEN}'}}, spec=aiohttp.web_request.Request)
github_client = mock(spec=github.Github)
github_repository = mock(spec=github.Repository)
github_commit = mock(spec=github.Commit)
pipeline_controller = PipelineController(github_client, mock(), self.API_TOKEN)
# given
when(request).json().thenReturn(async_value(proper_data))
when(CommentRequestData).is_valid_comment_data(proper_data).thenReturn(True)
when(github_client).get_repo('repo').thenReturn(github_repository)
when(github_repository).get_commit('123abc').thenReturn(github_commit)
when(github_commit).create_comment(body='job\nComments: Comment body')
# when
response: aiohttp.web.Response = await pipeline_controller.handle_comment(request)
# then
assert response.status == 200
assert response.text == 'Comment ACK'
python类Repository()的实例源码
def test__when_status_data_is_ok__should_call_github_client_to_create_status(self):
proper_data = {'repository': 'repo', 'sha': '123abc', 'state': 'State', 'description': 'Description', 'context': 'Context', 'url': 'pr_url'}
request = mock({'headers': {'Authorization': f'Token {self.API_TOKEN}'}}, spec=aiohttp.web_request.Request, strict=True)
github_client = mock(spec=github.Github, strict=True)
github_repository = mock(spec=github.Repository, strict=True)
github_commit = mock(spec=github.Commit, strict=True)
pipeline_controller = PipelineController(github_client, mock(), self.API_TOKEN)
# given
when(request).json().thenReturn(async_value(proper_data))
when(StatusRequestData).is_valid_status_data(proper_data).thenReturn(True)
when(github_client).get_repo('repo').thenReturn(github_repository)
when(github_repository).get_commit('123abc').thenReturn(github_commit)
when(github_commit).create_status(state='State', description='Description', target_url='pr_url', context='Context')
# when
response: aiohttp.web.Response = await pipeline_controller.handle_status(request)
# then
assert response.status == 200
assert response.text == 'Status ACK'
def test__when_github_entity_is_not_found__status_should_return_404_response_with_github_explanation(self):
parameters = {'repository': 'repo', 'sha': 'null', 'state': 'State', 'description': 'Description', 'context': 'Context', 'url': 'pr_url'}
request = mock({'headers': {'Authorization': f'Token {self.API_TOKEN}'}}, spec=aiohttp.web_request.Request, strict=True)
github_client = mock(spec=github.Github, strict=True)
github_repository = mock(spec=github.Repository, strict=True)
pipeline_controller = PipelineController(github_client, mock(), self.API_TOKEN)
# given
when(request).json().thenReturn(async_value(parameters))
when(StatusRequestData).is_valid_status_data(parameters).thenReturn(True)
when(github_client).get_repo('repo').thenReturn(github_repository)
github_exception_data = {'message': 'Not Found', 'documentation_url': 'https://developer.github.com/v3/'}
when(github_repository).get_commit('null').thenRaise(github.GithubException(404, github_exception_data))
# when
response: aiohttp.web.Response = await pipeline_controller.handle_status(request)
# then
assert response.status == 404
assert response.reason == str(github_exception_data)
def test__when_github_entity_is_not_found__comment_should_return_404_response_with_github_explanation(self):
parameters = {'repository': 'repo', 'sha': 'null', 'body': 'Comment body', 'jobName': 'job'}
request = mock({'headers': {'Authorization': f'Token {self.API_TOKEN}'}}, spec=aiohttp.web_request.Request, strict=True)
github_client = mock(spec=github.Github, strict=True)
github_repository = mock(spec=github.Repository, strict=True)
pipeline_controller = PipelineController(github_client, mock(), self.API_TOKEN)
# given
when(request).json().thenReturn(async_value(parameters))
when(CommentRequestData).is_valid_comment_data(parameters).thenReturn(True)
when(github_client).get_repo('repo').thenReturn(github_repository)
github_exception_data = {'message': 'Not Found', 'documentation_url': 'https://developer.github.com/v3/'}
when(github_repository).get_commit('null').thenRaise(github.GithubException(404, github_exception_data))
# when
response: aiohttp.web.Response = await pipeline_controller.handle_comment(request)
# then
assert response.status == 404
assert response.reason == str(github_exception_data)
def test__when_setting_pr_sync_label__if_github_raises_more_then_3_times__timeout_error_should_be_raised(self):
github_client: github.Github = mock(spec=github.Github, strict=True)
github_controller = GithubController(github_client, mock(), mock(), mock())
github_repository: github.Repository.Repository = mock(spec=github.Repository.Repository, strict=True)
mock(spec=asyncio)
# when
when(github_client).get_repo('repo')\
.thenRaise(github.GithubException(404, 'repo not found'))\
.thenRaise(github.GithubException(404, 'repo not found'))\
.thenReturn(github_repository)
when(github_repository).get_issue(43)\
.thenRaise(github.GithubException(404, 'repo not found'))
when(asyncio).sleep(1)\
.thenReturn(async_value(None))\
.thenReturn(async_value(None))\
.thenReturn(async_value(None))
# then
with pytest.raises(TriggearTimeoutError) as timeout_error:
await github_controller.set_pr_sync_label_with_retry('repo', 43)
assert str(timeout_error.value) == 'Failed to set label on PR #43 in repo repo after 3 retries'
def test__when_get_repo_labels_is_called__only_label_names_are_returned(self):
github_client: github.Github = mock(spec=github.Github, strict=True)
github_controller = GithubController(github_client, mock(), mock(), mock())
github_repository: github.Repository.Repository = mock(spec=github.Repository.Repository, strict=True)
label: github.Label.Label = mock({'name': 'label'}, spec=github.Label.Label, strict=True)
other_label: github.Label.Label = mock({'name': 'other_label'}, spec=github.Label.Label, strict=True)
# given
when(github_client).get_repo('repo')\
.thenReturn(github_repository)
when(github_repository).get_labels()\
.thenReturn([label, other_label])
# when
result: List[str] = github_controller.get_repo_labels('repo')
# then
assert result == ['label', 'other_label']
def test__get_pr_labels__should_return_only_label_names(self):
github_client: github.Github = mock(spec=github.Github, strict=True)
github_controller = GithubController(github_client, mock(), mock(), mock())
github_repository: github.Repository.Repository = mock(spec=github.Repository.Repository, strict=True)
label: github.Label.Label = mock({'name': 'label'}, spec=github.Label.Label, strict=True)
other_label: github.Label.Label = mock({'name': 'other_label'}, spec=github.Label.Label, strict=True)
github_issue: github.Issue.Issue = mock({'labels': [label, other_label]}, spec=github.Issue.Issue, strict=True)
when(github_client).get_repo('repo')\
.thenReturn(github_repository)
when(github_repository).get_issue(25)\
.thenReturn(github_issue)
labels: List[str] = github_controller.get_pr_labels('repo', 25)
assert ['label', 'other_label'] == labels
def parent(self):
"""
:type: :class:`github.Repository.Repository`
"""
self._completeIfNotSet(self._parent)
return self._parent.value
def source(self):
"""
:type: :class:`github.Repository.Repository`
"""
self._completeIfNotSet(self._source)
return self._source.value
def get_forks(self):
"""
:calls: `GET /repos/:owner/:repo/forks <http://developer.github.com/v3/repos/forks>`_
:rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Repository.Repository`
"""
return github.PaginatedList.PaginatedList(
Repository,
self._requester,
self.url + "/forks",
None
)
def parent(self):
"""
:type: :class:`github.Repository.Repository`
"""
self._completeIfNotSet(self._parent)
return self._parent.value
def source(self):
"""
:type: :class:`github.Repository.Repository`
"""
self._completeIfNotSet(self._source)
return self._source.value
def get_forks(self):
"""
:calls: `GET /repos/:owner/:repo/forks <http://developer.github.com/v3/repos/forks>`_
:rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Repository.Repository`
"""
return github.PaginatedList.PaginatedList(
Repository,
self._requester,
self.url + "/forks",
None
)
def test__get_latest_commit_sha__should_call_proper_github_entities(self):
github_client: github.Github = mock(spec=github.Github, strict=True)
github_repo: github.Repository.Repository = mock(spec=github.Repository.Repository, strict=True)
github_head: github.PullRequestPart.PullRequestPart = mock({'sha': '123zxc'}, spec=github.PullRequestPart, strict=True)
github_pull_request: github.PullRequest.PullRequest = mock({'head': github_head}, spec=github.PullRequest.PullRequest, strict=True)
github_controller = GithubController(github_client, mock(), mock(), mock())
expect(github_client).get_repo('triggear').thenReturn(github_repo)
expect(github_repo).get_pull(32).thenReturn(github_pull_request)
sha: str = github_controller.get_latest_commit_sha(32, 'triggear')
assert '123zxc' == sha
def test__get_pr_branch__should_call_proper_github_entities(self):
github_client: github.Github = mock(spec=github.Github, strict=True)
github_repo: github.Repository.Repository = mock(spec=github.Repository.Repository, strict=True)
github_head: github.PullRequestPart.PullRequestPart = mock({'ref': '123zxc'}, spec=github.PullRequestPart, strict=True)
github_pull_request: github.PullRequest.PullRequest = mock({'head': github_head}, spec=github.PullRequest.PullRequest, strict=True)
github_controller = GithubController(github_client, mock(), mock(), mock())
expect(github_client).get_repo('triggear').thenReturn(github_repo)
expect(github_repo).get_pull(32).thenReturn(github_pull_request)
sha: str = github_controller.get_pr_branch(32, 'triggear')
assert '123zxc' == sha
def test__are_files_in_repo__should_return_false_on_any_missing_file(self):
github_client: github.Github = mock(spec=github.Github, strict=True)
github_repo: github.Repository.Repository = mock(spec=github.Repository.Repository, strict=True)
github_controller = GithubController(github_client, mock(), mock(), mock())
hook_details = mock({'repository': 'repo', 'sha': '123qwe', 'branch': 'master'}, spec=HookDetails, strict=True)
files = ['app/main.py', '.gitignore']
expect(github_client).get_repo('repo').thenReturn(github_repo)
expect(github_repo).get_file_contents(path='app/main.py', ref='123qwe').thenReturn(None)
expect(github_repo).get_file_contents(path='.gitignore', ref='123qwe').thenRaise(GithubException(404, 'File not found'))
assert not await github_controller.are_files_in_repo(files, hook_details)
def test__create_pr_comment__calls_proper_github_entities(self):
github_client: github.Github = mock(spec=github.Github, strict=True)
github_repo: github.Repository.Repository = mock(spec=github.Repository.Repository, strict=True)
github_issue: github.Issue.Issue = mock(spec=github.Issue.Issue, strict=True)
github_controller = GithubController(github_client, mock(), mock(), mock())
expect(github_client).get_repo('triggear').thenReturn(github_repo)
expect(github_repo).get_issue(23).thenReturn(github_issue)
expect(github_issue).create_comment(body='Comment to send')
github_controller.create_pr_comment(23, 'triggear', 'Comment to send')
def test__create_github_build_status__calls_github_client_properly(self):
github_client: github.Github = mock(spec=github.Github, strict=True)
github_repo: github.Repository.Repository = mock(spec=github.Repository.Repository, strict=True)
github_commit: github.Commit.Commit = mock(spec=github.Commit.Commit, strict=True)
github_controller = GithubController(github_client, mock(), mock(), mock())
expect(github_client).get_repo('repo').thenReturn(github_repo)
expect(github_repo).get_commit('123456').thenReturn(github_commit)
expect(github_commit).create_status(state='pending',
target_url='http://example.com',
description='whatever you need',
context='job')
await github_controller.create_github_build_status('repo', '123456', 'pending', 'http://example.com', 'whatever you need', 'job')
def parent(self):
"""
:type: :class:`github.Repository.Repository`
"""
self._completeIfNotSet(self._parent)
return self._parent.value
def source(self):
"""
:type: :class:`github.Repository.Repository`
"""
self._completeIfNotSet(self._source)
return self._source.value
def get_forks(self):
"""
:calls: `GET /repos/:owner/:repo/forks <http://developer.github.com/v3/repos/forks>`_
:rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Repository.Repository`
"""
return github.PaginatedList.PaginatedList(
Repository,
self._requester,
self.url + "/forks",
None
)
def get_or_create_from_repo(cls, repo: Repository) -> 'Repo':
try:
return cls.objects.get(id=repo.id)
except cls.DoesNotExist:
owner = GitHubUser.get_or_create(username=repo.owner.login)
return cls.objects.create(owner=owner, name=repo.name, id=repo.id)
def parent(self):
"""
:type: :class:`github.Repository.Repository`
"""
self._completeIfNotSet(self._parent)
return self._parent.value
def source(self):
"""
:type: :class:`github.Repository.Repository`
"""
self._completeIfNotSet(self._source)
return self._source.value
def get_forks(self):
"""
:calls: `GET /repos/:owner/:repo/forks <http://developer.github.com/v3/repos/forks>`_
:rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.Repository.Repository`
"""
return github.PaginatedList.PaginatedList(
Repository,
self._requester,
self.url + "/forks",
None
)
def _get_data_entry(self, repo: GithubRepository):
pass
def _get_data_entry(self, repo: GithubRepository):
return repo.get_languages()
def _get_data_entry(self, repo: GithubRepository):
return repo.description
def _get_data_entry(self, repo: GithubRepository):
git_tree = repo.get_git_tree(repo.default_branch, recursive=True)
items = git_tree.tree if git_tree else []
files = [item for item in items if item.type == 'blob']
extensions = [os.path.splitext(file.path)[1] for file in files]
return extensions
def _get_data_entry(self, repo: GithubRepository):
return repo.get_dir_contents('')
