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