Python github 模块，Repository() 实例源码

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'

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('')

def _extract_for_single_repo_single_extractor(self,
                                                  extractor: FeatureExtractor,
                                                  repo: Repository,
                                                  api_repo: GithubRepository) -> [Feature]:
        cached_features = []
        for feature in extractor.features:
            cached_feature = self._get_cached(repo, feature)
            if self.cache_features and cached_feature:
                cached_features.append(cached_feature)
            else:
                updated_features = extractor.extract(api_repo)
                for updated_feature in updated_features:
                    self._update_cached(repo, updated_feature)
                return updated_features
        return cached_features

def api_repo(self) -> GithubRepository:
        """
        :return: a lazily initialized Github API repository
        """
        if not self._api_repo:
            self._api_repo = GithubAuthentification().get_repo(self.repo.identifier)
        return self._api_repo

def extract(self, api_repo: GithubRepository = None) -> [Feature]:
        self._api_repo = api_repo
        try:
            self._extract()
        except UnknownObjectException:
            pass
        except GithubException as e:
            # Repository access blocked
            if e.status == 451:
                pass
            else:
                raise

        return self.features

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
        )