Coverage for jaypore_ci/remotes/github.py: 85%
58 statements
« prev ^ index » next coverage.py v7.2.2, created at 2023-03-30 09:04 +0000
1"""
2A github remote git host.
4This is used to report pipeline status to the remote.
5"""
6import os
8import requests
10from jaypore_ci.interfaces import Remote, RemoteApiFailed, Repo, RemoteInfo
11from jaypore_ci.logging import logger
14class Github(Remote): # pylint: disable=too-many-instance-attributes
15 """
16 The remote implementation for github.
17 """
19 def __headers__(self):
20 return {
21 "Authorization": f"Bearer {self.token}",
22 "Accept": "application/vnd.github+json",
23 "X-Github-Api-Version": "2022-11-28",
24 }
26 @classmethod
27 def from_env(cls, *, repo: Repo) -> "Github":
28 """
29 Creates a remote instance from the environment.
30 It will:
32 - Find the remote location using `git remote`.
33 - Find the current branch
34 - Create a new pull request for that branch
35 - Allow posting updates using the gitea token provided
36 """
37 rem = RemoteInfo.parse(repo.remote)
38 os.environ["JAYPORE_COMMIT_BRANCH"] = repo.branch
39 os.environ["JAYPORE_COMMIT_SHA"] = repo.sha
40 return cls(
41 root="https://api.github.com",
42 owner=rem.owner,
43 repo=rem.repo,
44 branch=repo.branch,
45 token=os.environ["JAYPORE_GITHUB_TOKEN"],
46 sha=repo.sha,
47 )
49 def __init__(
50 self, *, root, owner, repo, token, **kwargs
51 ): # pylint: disable=too-many-arguments
52 super().__init__(**kwargs)
53 # --- customer
54 self.root = root
55 self.api = root
56 self.owner = owner
57 self.repo = repo
58 self.token = token
59 self.timeout = 10
60 self.base_branch = "main"
62 def logging(self):
63 """
64 Return's a logging instance with information about gitea bound to it.
65 """
66 return logger.bind(
67 root=self.root, owner=self.owner, repo=self.repo, branch=self.branch
68 )
70 def get_pr_id(self):
71 """
72 Returns the pull request ID for the current branch.
73 """
74 r = requests.post(
75 f"{self.api}/repos/{self.owner}/{self.repo}/pulls",
76 headers=self.__headers__(),
77 timeout=self.timeout,
78 json={
79 "base": self.base_branch,
80 "body": "Branch auto created by JayporeCI",
81 "head": self.branch,
82 "title": self.branch,
83 },
84 )
85 self.logging().debug("Create PR", status_code=r.status_code)
86 if r.status_code == 201: 86 ↛ 87line 86 didn't jump to line 87, because the condition on line 86 was never true
87 return r.json()["number"]
88 r = requests.get(
89 f"{self.api}/repos/{self.owner}/{self.repo}/pulls",
90 headers=self.__headers__(),
91 timeout=self.timeout,
92 json={"base": self.base_branch, "head": self.branch, "draft": True},
93 )
94 self.logging().debug("Get PR", status_code=r.status_code)
95 if r.status_code == 200: 95 ↛ 98line 95 didn't jump to line 98, because the condition on line 95 was never false
96 if len(r.json()) == 1: 96 ↛ 98line 96 didn't jump to line 98, because the condition on line 96 was never false
97 return r.json()[0]["number"]
98 self.logging().debug(
99 "Failed github api",
100 api=self.api,
101 owner=self.owner,
102 repo=self.repo,
103 token=self.token,
104 branch=self.branch,
105 status=r.status_code,
106 response=r.text,
107 )
108 raise RemoteApiFailed(r)
110 def publish(self, report: str, status: str):
111 """
112 Will publish the report to the remote.
114 :param report: Report to write to remote.
115 :param status: One of ["pending", "success", "error", "failure"]
116 This is the dot/tick next to each commit in gitea.
117 """
118 assert status in ("pending", "success", "error", "failure")
119 issue_id = self.get_pr_id()
120 # Get existing PR body
121 r = requests.get(
122 f"{self.api}/repos/{self.owner}/{self.repo}/pulls/{issue_id}",
123 timeout=self.timeout,
124 headers=self.__headers__(),
125 )
126 self.logging().debug("Get existing body", status_code=r.status_code)
127 assert r.status_code == 200
128 body = r.json()["body"]
129 body = (line for line in body.split("\n"))
130 prefix = []
131 for line in body:
132 if "```jayporeci" in line: 132 ↛ 133line 132 didn't jump to line 133, because the condition on line 132 was never true
133 prefix = prefix[:-1]
134 break
135 prefix.append(line)
136 while prefix and prefix[-1].strip() == "": 136 ↛ 137line 136 didn't jump to line 137, because the condition on line 136 was never true
137 prefix = prefix[:-1]
138 prefix.append("")
139 # Post new body with report
140 report = "\n".join(prefix) + "\n" + report
141 r = requests.patch(
142 f"{self.api}/repos/{self.owner}/{self.repo}/pulls/{issue_id}",
143 json={"body": report},
144 timeout=self.timeout,
145 headers=self.__headers__(),
146 )
147 self.logging().debug("Published new report", status_code=r.status_code)
148 # Set commit status
149 r = requests.post(
150 f"{self.api}/repos/{self.owner}/{self.repo}/statuses/{self.sha}",
151 json={
152 "context": "JayporeCi",
153 "description": f"Pipeline status is: {status}",
154 "state": status,
155 },
156 timeout=self.timeout,
157 headers=self.__headers__(),
158 )
159 self.logging().debug(
160 "Published new status", status=status, status_code=r.status_code
161 )