Coverage for jaypore_ci/interfaces.py: 82%

1""" 

2Defines interfaces for remotes and executors. 

3 

4Currently only gitea and docker are supported as remote and executor 

5respectively. 

6""" 

7from enum import Enum 

8from pathlib import Path 

9from urllib.parse import urlparse 

10from typing import NamedTuple, List 

11 

12 

13class TriggerFailed(Exception): 

14 "Failure to trigger a job" 

15 

16 

17class RemoteApiFailed(Exception): 

18 "Failure while working with a remote" 

19 

20 

21class JobStatus(NamedTuple): 

22 is_running: bool 

23 exit_code: int 

24 logs: str 

25 started_at: str 

26 finished_at: str 

27 

28 

29class Status(Enum): 

30 "Each pipeline can ONLY be in any one of these statuses" 

31 PENDING = 10 

32 RUNNING = 30 

33 FAILED = 40 

34 PASSED = 50 

35 TIMEOUT = 60 

36 SKIPPED = 70 

37 

38 

39class RemoteInfo(NamedTuple): 

40 """ 

41 Holds information about the remote irrespective of if the remote was ssh or 

42 https. 

43 """ 

44 

45 netloc: str 

46 owner: str 

47 repo: str 

48 original: str 

49 

50 @classmethod 

51 def parse(cls, remote: str) -> "RemoteInfo": 

52 """ 

53 Given a git remote url string, parses and breaks down information 

54 contained in the url. 

55 

56 Works with the following formats: 

57 

58 ssh://git@gitea.arjoonn.com:arjoonn/jaypore_ci.git 

59 ssh+git://git@gitea.arjoonn.com:arjoonn/jaypore_ci.git 

60 

61 git@gitea.arjoonn.com:arjoonn/jaypore_ci.git 

62 git@gitea.arjoonn.com:arjoonn/jaypore_ci.git 

63 

64 https://gitea.arjoonn.com/midpath/jaypore_ci.git 

65 http://gitea.arjoonn.com/midpath/jaypore_ci.git 

66 """ 

67 original = remote 

68 if ( 68 ↛ 73line 68 didn't jump to line 73

69 ("ssh://" in remote or "ssh+git://" in remote or "://" not in remote) 

70 and "@" in remote 

71 and remote.endswith(".git") 

72 ): 

73 _, remote = remote.split("@") 

74 netloc, path = remote.split(":") 

75 owner, repo = path.split("/") 

76 return RemoteInfo( 

77 netloc=netloc, 

78 owner=owner, 

79 repo=repo.replace(".git", ""), 

80 original=original, 

81 ) 

82 url = urlparse(remote) 

83 return RemoteInfo( 

84 netloc=url.netloc, 

85 owner=Path(url.path).parts[1], 

86 repo=Path(url.path).parts[2].replace(".git", ""), 

87 original=original, 

88 ) 

89 

90 

91class Repo: 

92 """ 

93 Contains information about the current VCS repo. 

94 """ 

95 

96 def __init__(self, sha: str, branch: str, remote: str, commit_message: str): 

97 self.sha: str = sha 

98 self.branch: str = branch 

99 self.remote: str = remote 

100 self.commit_message: str = commit_message 

101 

102 def files_changed(self, target: str) -> List[str]: 

103 """ 

104 Returns list of file paths that have changed between current sha and 

105 target. 

106 """ 

107 raise NotImplementedError() 

108 

109 @classmethod 

110 def from_env(cls) -> "Repo": 

111 """ 

112 Creates a :class:`~jaypore_ci.interfaces.Repo` instance 

113 from the environment and git repo on disk. 

114 """ 

115 raise NotImplementedError() 

116 

117 

118class Executor: 

119 """ 

120 An executor is something used to run a job. 

121 It could be docker / podman / shell etc. 

122 """ 

123 

124 def run(self, job: "Job") -> str: 

125 "Run a job and return it's ID" 

126 raise NotImplementedError() 

127 

128 def __init__(self): 

129 self.pipe_id = None 

130 self.pipeline = None 

131 

132 def set_pipeline(self, pipeline: "Pipeline") -> None: 

133 """Set the current pipeline to the given one.""" 

134 self.pipe_id = id(pipeline) 

135 self.pipeline = pipeline 

136 

137 def setup(self) -> None: 

138 """ 

139 This function is meant to perform any work that should be done before 

140 running any jobs. 

141 """ 

142 

143 def teardown(self) -> None: 

144 """ 

145 On exit the executor must clean up any pending / stuck / zombie jobs that are still there. 

146 """ 

147 

148 def get_status(self, run_id: str) -> JobStatus: 

149 """ 

150 Returns the status of a given run. 

151 """ 

152 raise NotImplementedError() 

153 

154 

155class Remote: 

156 """ 

157 Something that allows us to show other people the status of the CI job. 

158 It could be gitea / github / gitlab / email system. 

159 """ 

160 

161 def __init__(self, *, sha, branch): 

162 self.sha = sha 

163 self.branch = branch 

164 

165 def publish(self, report: str, status: str): 

166 """ 

167 Publish this report somewhere. 

168 """ 

169 raise NotImplementedError() 

170 

171 def setup(self) -> None: 

172 """ 

173 This function is meant to perform any work that should be done before 

174 running any jobs. 

175 """ 

176 

177 def teardown(self) -> None: 

178 """ 

179 This function will be called once the pipeline is finished. 

180 """ 

181 

182 @classmethod 

183 def from_env(cls, *, repo: "Repo"): 

184 """ 

185 This function should create a Remote instance from the given environment. 

186 It can read git information / look at environment variables etc. 

187 """ 

188 raise NotImplementedError() 

189 

190 

191class Reporter: 

192 """ 

193 Something that generates the status of a pipeline. 

194 

195 It can be used to generate reports in markdown, plaintext, html, pdf etc. 

196 """ 

197 

198 def render(self, pipeline: "Pipeline") -> str: 

199 """ 

200 Render a report for the pipeline. 

201 """ 

202 raise NotImplementedError()