Add support for workflow jobs and steps (#1951)

Co-authored-by: Enrico Minack <github@enrico.minack.dev>
This commit is contained in:
Jeppe Fihl-Pearson
2023-06-20 14:37:56 +08:00
committed by GitHub
co-authored by Enrico Minack
parent cb50dec5ba
commit 804c3107c8
11 changed files with 537 additions and 2 deletions
+201
View File
@@ -0,0 +1,201 @@
############################ Copyrights and license ############################
# #
# Copyright 2021 Jeppe Fihl-Pearson <jeppe@tenzer.dk> #
# #
# This file is part of PyGithub. #
# http://pygithub.readthedocs.io/ #
# #
# PyGithub is free software: you can redistribute it and/or modify it under #
# the terms of the GNU Lesser General Public License as published by the Free #
# Software Foundation, either version 3 of the License, or (at your option) #
# any later version. #
# #
# PyGithub is distributed in the hope that it will be useful, but WITHOUT ANY #
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS #
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more #
# details. #
# #
# You should have received a copy of the GNU Lesser General Public License #
# along with PyGithub. If not, see <http://www.gnu.org/licenses/>. #
# #
################################################################################
import github.GithubObject
import github.WorkflowStep
class WorkflowJob(github.GithubObject.CompletableGithubObject):
"""
This class represents Workflow Jobs. The reference can be found here https://docs.github.com/en/rest/reference/actions#workflow-jobs
"""
def __repr__(self):
return self.get__repr__({"id": self._id.value, "url": self._url.value})
@property
def check_run_url(self):
"""
:type: string
"""
self._completeIfNotSet(self._check_run_url)
return self._check_run_url.value
@property
def completed_at(self):
"""
:type: datetime.datetime
"""
self._completeIfNotSet(self._completed_at)
return self._completed_at.value
@property
def conclusion(self):
"""
:type: string
"""
self._completeIfNotSet(self._conclusion)
return self._conclusion.value
@property
def head_sha(self):
"""
:type: string
"""
self._completeIfNotSet(self._head_sha)
return self._head_sha.value
@property
def html_url(self):
"""
:type: string
"""
self._completeIfNotSet(self._html_url)
return self._html_url.value
@property
def id(self):
"""
:type: int
"""
self._completeIfNotSet(self._id)
return self._id.value
@property
def name(self):
"""
:type: string
"""
self._completeIfNotSet(self._name)
return self._name.value
@property
def node_id(self):
"""
:type: string
"""
self._completeIfNotSet(self._node_id)
return self._node_id.value
@property
def run_id(self):
"""
:type: integer
"""
self._completeIfNotSet(self._run_id)
return self._run_id.value
@property
def run_url(self):
"""
:type: string
"""
self._completeIfNotSet(self._run_url)
return self._run_url.value
@property
def started_at(self):
"""
:type: datetime.datetime
"""
self._completeIfNotSet(self._started_at)
return self._started_at.value
@property
def status(self):
"""
:type: string
"""
self._completeIfNotSet(self._status)
return self._status.value
@property
def steps(self):
"""
:type: list of github.WorkflowStep.WorkflowStep
"""
self._completeIfNotSet(self._steps)
return self._steps.value
@property
def url(self):
"""
:type: string
"""
self._completeIfNotSet(self._url)
return self._url.value
def logs_url(self):
"""
:type: string
"""
print(f"{self.url}/logs")
headers, _ = self._requester.requestBlobAndCheck("GET", f"{self.url}/logs")
return headers["location"]
def _initAttributes(self):
self._check_run_url = github.GithubObject.NotSet
self._completed_at = github.GithubObject.NotSet
self._conclusion = github.GithubObject.NotSet
self._head_sha = github.GithubObject.NotSet
self._html_url = github.GithubObject.NotSet
self._id = github.GithubObject.NotSet
self._name = github.GithubObject.NotSet
self._node_id = github.GithubObject.NotSet
self._run_id = github.GithubObject.NotSet
self._run_url = github.GithubObject.NotSet
self._started_at = github.GithubObject.NotSet
self._status = github.GithubObject.NotSet
self._steps = github.GithubObject.NotSet
self._url = github.GithubObject.NotSet
def _useAttributes(self, attributes):
if "check_run_url" in attributes: # pragma no branch
self._check_run_url = self._makeStringAttribute(attributes["check_run_url"])
if "completed_at" in attributes: # pragma no branch
self._completed_at = self._makeDatetimeAttribute(attributes["completed_at"])
if "conclusion" in attributes: # pragma no branch
self._conclusion = self._makeStringAttribute(attributes["conclusion"])
if "head_sha" in attributes: # pragma no branch
self._head_sha = self._makeStringAttribute(attributes["head_sha"])
if "html_url" in attributes: # pragma no branch
self._html_url = self._makeStringAttribute(attributes["html_url"])
if "id" in attributes: # pragma no branch
self._id = self._makeIntAttribute(attributes["id"])
if "name" in attributes: # pragma no branch
self._name = self._makeStringAttribute(attributes["name"])
if "node_id" in attributes: # pragma no branch
self._node_id = self._makeStringAttribute(attributes["node_id"])
if "run_id" in attributes: # pragma no branch
self._run_id = self._makeIntAttribute(attributes["run_id"])
if "run_url" in attributes: # pragma no branch
self._run_url = self._makeStringAttribute(attributes["run_url"])
if "started_at" in attributes: # pragma no branch
self._started_at = self._makeDatetimeAttribute(attributes["started_at"])
if "status" in attributes: # pragma no branch
self._status = self._makeStringAttribute(attributes["status"])
if "steps" in attributes: # pragma no branch
self._steps = self._makeListOfClassesAttribute(
github.WorkflowStep.WorkflowStep, attributes["steps"]
)
if "url" in attributes: # pragma no branch
self._url = self._makeStringAttribute(attributes["url"])
+39
View File
@@ -0,0 +1,39 @@
from datetime import datetime
from typing import Any, Dict, List
from github.GithubObject import CompletableGithubObject
from github.WorkflowStep import WorkflowStep
class WorkflowJob(CompletableGithubObject):
def __repr__(self) -> str: ...
def _initAttributes(self) -> None: ...
def _useAttributes(self, attributes: Dict[str, Any]) -> None: ...
@property
def check_run_url(self) -> str: ...
@property
def completed_at(self) -> datetime: ...
@property
def conclusion(self) -> str: ...
@property
def head_sha(self) -> str: ...
@property
def html_url(self) -> str: ...
@property
def id(self) -> int: ...
@property
def name(self) -> str: ...
@property
def node_id(self) -> str: ...
@property
def run_id(self) -> int: ...
@property
def run_url(self) -> str: ...
@property
def started_at(self) -> datetime: ...
@property
def status(self) -> str: ...
@property
def steps(self) -> List[WorkflowStep]: ...
@property
def url(self) -> str: ...
def logs_url(self) -> str: ...
+23
View File
@@ -26,6 +26,7 @@ from collections import namedtuple
import github.Artifact
import github.GithubObject
import github.PullRequest
import github.WorkflowJob
class WorkflowRun(github.GithubObject.CompletableGithubObject):
@@ -302,6 +303,28 @@ class WorkflowRun(github.GithubObject.CompletableGithubObject):
status, _, _ = self._requester.requestJson("DELETE", self.url)
return status == 204
def jobs(self, _filter=github.GithubObject.NotSet):
"""
:calls "`GET /repos/{owner}/{repo}/actions/runs/{run_id}/jobs <https://docs.github.com/en/rest/reference/actions#list-jobs-for-a-workflow-run>`_
:param _filter: string `latest`, or `all`
:rtype: :class:`github.PaginatedList.PaginatedList` of :class:`github.WorkflowJob.WorkflowJob`
"""
assert _filter is github.GithubObject.NotSet or isinstance(
_filter, str
), _filter
url_parameters = dict()
if _filter is not github.GithubObject.NotSet:
url_parameters["filter"] = _filter
return github.PaginatedList.PaginatedList(
github.WorkflowJob.WorkflowJob,
self._requester,
self.jobs_url,
url_parameters,
list_item="jobs",
)
def _initAttributes(self):
self._id = github.GithubObject.NotSet
self._name = github.GithubObject.NotSet
+7 -2
View File
@@ -1,10 +1,12 @@
from datetime import datetime
from typing import Any, Dict, NamedTuple, List
from typing import Any, Dict, NamedTuple, List, Union
from github.GitCommit import GitCommit
from github.GithubObject import CompletableGithubObject
from github.GithubObject import CompletableGithubObject, _NotSetType
from github.PaginatedList import PaginatedList
from github.PullRequest import PullRequest
from github.Repository import Repository
from github.WorkflowJob import WorkflowJob
class TimingData(NamedTuple):
billable: Dict[str, Dict[str, int]]
@@ -71,5 +73,8 @@ class WorkflowRun(CompletableGithubObject):
@property
def head_repository(self) -> Repository: ...
def cancel(self) -> bool: ...
def jobs(
self, _filter: Union[str, _NotSetType] = ...
) -> PaginatedList[WorkflowJob]: ...
def rerun(self) -> bool: ...
def timing(self) -> TimingData: ...
+104
View File
@@ -0,0 +1,104 @@
############################ Copyrights and license ############################
# #
# Copyright 2021 Jeppe Fihl-Pearson <jeppe@tenzer.dk> #
# #
# This file is part of PyGithub. #
# http://pygithub.readthedocs.io/ #
# #
# PyGithub is free software: you can redistribute it and/or modify it under #
# the terms of the GNU Lesser General Public License as published by the Free #
# Software Foundation, either version 3 of the License, or (at your option) #
# any later version. #
# #
# PyGithub is distributed in the hope that it will be useful, but WITHOUT ANY #
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS #
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more #
# details. #
# #
# You should have received a copy of the GNU Lesser General Public License #
# along with PyGithub. If not, see <http://www.gnu.org/licenses/>. #
# #
################################################################################
import github.GithubObject
class WorkflowStep(github.GithubObject.CompletableGithubObject):
"""
This class represents steps in a Workflow Job. The reference can be found here https://docs.github.com/en/rest/reference/actions#workflow-jobs
"""
def __repr__(self):
return self.get__repr__(
{"number": self._number.value, "name": self._name.value}
)
@property
def completed_at(self):
"""
:type: datetime.datetime
"""
self._completeIfNotSet(self._completed_at)
return self._completed_at.value
@property
def conclusion(self):
"""
:type: string
"""
self._completeIfNotSet(self._conclusion)
return self._conclusion.value
@property
def name(self):
"""
:type: string
"""
self._completeIfNotSet(self._name)
return self._name.value
@property
def number(self):
"""
:type: integer
"""
self._completeIfNotSet(self._number)
return self._number.value
@property
def started_at(self):
"""
:type: datetime.datetime
"""
self._completeIfNotSet(self._started_at)
return self._started_at.value
@property
def status(self):
"""
:type: string
"""
self._completeIfNotSet(self._status)
return self._status.value
def _initAttributes(self):
self._completed_at = github.GithubObject.NotSet
self._conclusion = github.GithubObject.NotSet
self._name = github.GithubObject.NotSet
self._number = github.GithubObject.NotSet
self._started_at = github.GithubObject.NotSet
self._status = github.GithubObject.NotSet
def _useAttributes(self, attributes):
if "completed_at" in attributes: # pragma no branch
self._completed_at = self._makeDatetimeAttribute(attributes["completed_at"])
if "conclusion" in attributes: # pragma no branch
self._conclusion = self._makeStringAttribute(attributes["conclusion"])
if "name" in attributes: # pragma no branch
self._name = self._makeStringAttribute(attributes["name"])
if "number" in attributes: # pragma no branch
self._number = self._makeIntAttribute(attributes["number"])
if "started_at" in attributes: # pragma no branch
self._started_at = self._makeDatetimeAttribute(attributes["started_at"])
if "status" in attributes: # pragma no branch
self._status = self._makeStringAttribute(attributes["status"])
+21
View File
@@ -0,0 +1,21 @@
from datetime import datetime
from typing import Any, Dict
from github.GithubObject import CompletableGithubObject
class WorkflowStep(CompletableGithubObject):
def __repr__(self) -> str: ...
def _initAttributes(self) -> None: ...
def _useAttributes(self, attributes: Dict[str, Any]) -> None: ...
@property
def completed_at(self) -> datetime: ...
@property
def conclusion(self) -> str: ...
@property
def name(self) -> str: ...
@property
def number(self) -> int: ...
@property
def started_at(self) -> datetime: ...
@property
def status(self) -> str: ...
File diff suppressed because one or more lines are too long
@@ -0,0 +1,11 @@
https
GET
api.github.com
None
/repos/PyGithub/PyGithub/actions/jobs/11421878319/logs
{'Authorization': 'Basic login_and_password_removed', 'User-Agent': 'PyGithub/Python'}
None
302
[('Server', 'GitHub.com'), ('Date', 'Wed, 15 Mar 2023 17:01:58 GMT'), ('Content-Type', 'text/html;charset=utf-8'), ('Content-Length', '0'), ('Location', 'https://pipelines.actions.githubusercontent.com/serviceHosts/d560a817-28d4-4544-a539-eb35c2a56899/_apis/pipelines/1/runs/5/signedlogcontent/5?urlExpires=2023-03-15T17%3A02%3A58.1305046Z&urlSigningMethod=HMACV1&urlSignature=abcdefghijklmn'), ('x-github-api-version-selected', '2022-11-28'), ('X-RateLimit-Limit', '5000'), ('X-RateLimit-Remaining', '4996'), ('X-RateLimit-Reset', '1678903315'), ('X-RateLimit-Used', '4'), ('X-RateLimit-Resource', 'core'), ('Access-Control-Expose-Headers', 'ETag, Link, Location, Retry-After, X-GitHub-OTP, X-RateLimit-Limit, X-RateLimit-Remaining, X-RateLimit-Used, X-RateLimit-Resource, X-RateLimit-Reset, X-OAuth-Scopes, X-Accepted-OAuth-Scopes, X-Poll-Interval, X-GitHub-Media-Type, X-GitHub-SSO, X-GitHub-Request-Id, Deprecation, Sunset'), ('Access-Control-Allow-Origin', '*'), ('Strict-Transport-Security', 'max-age=31536000; includeSubdomains; preload'), ('X-Frame-Options', 'deny'), ('X-Content-Type-Options', 'nosniff'), ('X-XSS-Protection', '0'), ('Referrer-Policy', 'origin-when-cross-origin, strict-origin-when-cross-origin'), ('Content-Security-Policy', "default-src 'none'"), ('Vary', 'Accept-Encoding, Accept, X-Requested-With'), ('X-GitHub-Request-Id', 'A66C:CC26:9983F9A:9BF3547:6411FA05')]
File diff suppressed because one or more lines are too long
+80
View File
@@ -0,0 +1,80 @@
############################ Copyrights and license ############################
# #
# Copyright 2021 Jeppe Fihl-Pearson <jeppe@tenzer.dk> #
# #
# This file is part of PyGithub. #
# http://pygithub.readthedocs.io/ #
# #
# PyGithub is free software: you can redistribute it and/or modify it under #
# the terms of the GNU Lesser General Public License as published by the Free #
# Software Foundation, either version 3 of the License, or (at your option) #
# any later version. #
# #
# PyGithub is distributed in the hope that it will be useful, but WITHOUT ANY #
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS #
# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more #
# details. #
# #
# You should have received a copy of the GNU Lesser General Public License #
# along with PyGithub. If not, see <http://www.gnu.org/licenses/>. #
# #
################################################################################
import datetime
from . import Framework
class WorkflowJob(Framework.TestCase):
def setUp(self):
super().setUp()
self.repo = self.g.get_repo("PyGithub/PyGithub")
self.job = self.repo.get_workflow_run(4205440316).jobs()[0]
def testAttributes(self):
self.assertEqual(self.job.id, 11421878319)
self.assertEqual(self.job.run_id, 4205440316)
self.assertEqual(
self.job.run_url,
"https://api.github.com/repos/PyGithub/PyGithub/actions/runs/4205440316",
)
self.assertEqual(self.job.node_id, "CR_kwDOGpsAJ88AAAACqMwILw")
self.assertEqual(self.job.head_sha, "06ec040b2eeef6c0316dd5abcda0608525a3f205")
self.assertEqual(
self.job.url,
"https://api.github.com/repos/PyGithub/PyGithub/actions/jobs/11421878319",
)
self.assertEqual(
self.job.html_url,
"https://github.com/PyGithub/PyGithub/actions/runs/4205440316/jobs/7297536068",
)
self.assertEqual(self.job.status, "completed")
self.assertEqual(self.job.conclusion, "success")
started_at = datetime.datetime(2023, 2, 17, 16, 3, 46)
self.assertEqual(self.job.started_at, started_at)
completed_at = datetime.datetime(2023, 2, 17, 16, 4, 52)
self.assertEqual(self.job.completed_at, completed_at)
self.assertEqual(self.job.name, "test (Python 3.7)")
self.assertEqual(
self.job.check_run_url,
"https://api.github.com/repos/PyGithub/PyGithub/check-runs/11421878319",
)
self.assertListKeyEqual(
self.job.steps,
lambda s: s.name,
[
"Set up job",
"Run actions/checkout@v2",
"Set up Python",
"Install tox",
"Run tests",
"Upload coverage to Codecov",
"Post Set up Python",
"Post Run actions/checkout@v2",
"Complete job",
],
)
self.assertEqual(
self.job.logs_url(),
"https://pipelines.actions.githubusercontent.com/serviceHosts/d560a817-28d4-4544-a539-eb35c2a56899/_apis/pipelines/1/runs/5/signedlogcontent/5?urlExpires=2023-03-15T17%3A02%3A58.1305046Z&urlSigningMethod=HMACV1&urlSignature=abcdefghijklmn",
)
+7
View File
@@ -135,3 +135,10 @@ class WorkflowRun(Framework.TestCase):
def test_delete(self):
wr = self.repo.get_workflow_run(3881497935)
self.assertFalse(wr.delete())
def test_jobs(self):
self.assertListKeyEqual(
self.workflow_run.jobs(),
lambda j: j.id,
[10545727758, 10545727888, 10545728039, 10545728190, 10545728356],
)