Coverage for src/types/gitlab.py: 67.57%

1#!/usr/bin/env python3 

2 

3# Standard libraries 

4from os.path import join 

5from typing import List, NamedTuple 

6 

7# Modules libraries 

8from gitlab.v4.objects import ( 

9 GroupRunner, 

10 ProjectRunner, 

11) 

12 

13# MigrationEntities class 

14class MigrationEntities: 

15 

16 # Members 

17 cache_entities: List[str] = [] 

18 cache_keys: List[str] = [] 

19 cache_removers: List[str] = [] 

20 cache_templates: List[str] = [] 

21 

22 # Entity 

23 class Entity(NamedTuple): 

24 

25 # Variables 

26 name: str 

27 group_globs: List[str] = [] 

28 project_directories: List[str] = [] 

29 project_files: List[str] = [] 

30 

31 # Remover 

32 class Remover(Entity): 

33 pass 

34 

35 # Template 

36 class Template(Entity): 

37 pass 

38 

39 # Defaults 

40 @staticmethod 

41 def defaults() -> List[str]: 

42 

43 # Return default exclusions 

44 return [ 

45 MigrationEntities.MEMBERS.name, 

46 ] 

47 

48 # Get 

49 @staticmethod 

50 def get(key: str) -> Entity: 

51 

52 # Get feature object 

53 feature = getattr(MigrationEntities, key) 

54 assert isinstance(feature, MigrationEntities.Entity) 

55 return feature 

56 

57 # Keys 

58 @staticmethod 

59 def keys() -> List[str]: 

60 

61 # Evaluate keys 

62 if MigrationEntities.cache_keys: 

63 return MigrationEntities.cache_keys 

64 

65 # Evaluate keys 

66 MigrationEntities.cache_keys = [ 

67 key for key in MigrationEntities.__dict__ 

68 if isinstance(getattr(MigrationEntities, key), MigrationEntities.Entity) 

69 ] 

70 return MigrationEntities.cache_keys 

71 

72 # Entities 

73 @staticmethod 

74 def entities() -> List[str]: 

75 

76 # Evaluate entities 

77 if MigrationEntities.cache_entities: 

78 return MigrationEntities.cache_entities 

79 

80 # Evaluate entities 

81 MigrationEntities.cache_entities = [ 

82 MigrationEntities.get(key).name for key in MigrationEntities.__dict__ 

83 if isinstance(getattr(MigrationEntities, key), MigrationEntities.Entity) and 

84 not isinstance(getattr(MigrationEntities, key), MigrationEntities.Remover) and 

85 not isinstance(getattr(MigrationEntities, key), MigrationEntities.Template) 

86 ] 

87 return MigrationEntities.cache_entities 

88 

89 # Removers 

90 @staticmethod 

91 def removers() -> List[str]: 

92 

93 # Evaluate removers 

94 if MigrationEntities.cache_removers: 

95 return MigrationEntities.cache_removers 

96 

97 # Evaluate removers 

98 MigrationEntities.cache_removers = [ 

99 MigrationEntities.get(key).name for key in MigrationEntities.__dict__ 

100 if isinstance(getattr(MigrationEntities, key), MigrationEntities.Remover) 

101 ] 

102 return MigrationEntities.cache_removers 

103 

104 # Templates 

105 @staticmethod 

106 def templates() -> List[str]: 

107 

108 # Evaluate templates 

109 if MigrationEntities.cache_templates: 

110 return MigrationEntities.cache_templates 

111 

112 # Evaluate templates 

113 MigrationEntities.cache_templates = [ 

114 MigrationEntities.get(key).name for key in MigrationEntities.__dict__ 

115 if isinstance(getattr(MigrationEntities, key), MigrationEntities.Template) 

116 ] 

117 return MigrationEntities.cache_templates 

118 

119 # Entity: Members 

120 MEMBERS = Entity( 

121 name='Members', 

122 group_globs=[ 

123 join('tree', 'groups', '*', 'members.ndjson'), 

124 ], 

125 project_directories=[], 

126 project_files=[ 

127 join('tree', 'project', 'project_members.ndjson'), 

128 ], 

129 ) 

130 

131 # Entity: Issues 

132 ISSUES = Entity( 

133 name='Issues', 

134 group_globs=[], 

135 project_directories=[], 

136 project_files=[ 

137 join('tree', 'project', 'issues.ndjson'), 

138 ], 

139 ) 

140 

141 # Entity: Issue boards 

142 BOARDS = Entity( 

143 name='Issue boards', 

144 group_globs=[], 

145 project_directories=[], 

146 project_files=[ 

147 join('tree', 'project', 'boards.ndjson'), 

148 ], 

149 ) 

150 

151 # Entity: Milestones 

152 MILESTONES = Entity( 

153 name='Milestones', 

154 group_globs=[], 

155 project_directories=[], 

156 project_files=[ 

157 join('tree', 'project', 'milestones.ndjson'), 

158 ], 

159 ) 

160 

161 # Entity: Labels 

162 LABELS = Entity( 

163 name='Labels', 

164 group_globs=[], 

165 project_directories=[], 

166 project_files=[ 

167 join('tree', 'project', 'labels.ndjson'), 

168 ], 

169 ) 

170 

171 # Entity: Repository / Merge_requests 

172 MERGE_REQUESTS = Entity( 

173 name='Merge requests', 

174 group_globs=[], 

175 project_directories=[], 

176 project_files=[ 

177 join('tree', 'project', 'merge_requests.ndjson'), 

178 ], 

179 ) 

180 

181 # Entity: Repository / LFS objects 

182 LFS_OBJECTS = Entity( 

183 name='LFS objects', 

184 group_globs=[], 

185 project_directories=[ 

186 join('lfs-objects'), 

187 ], 

188 project_files=[ 

189 join('lfs-objects.json'), 

190 ], 

191 ) 

192 

193 # Entity: Repository / CI/CD Pipelines 

194 PIPELINES = Entity( 

195 name='Pipelines', 

196 group_globs=[], 

197 project_directories=[], 

198 project_files=[ 

199 join('tree', 'project', 'ci_pipelines.ndjson'), 

200 join('tree', 'project', 'pipeline_schedules.ndjson'), 

201 ], 

202 ) 

203 

204 # Entity: Wiki 

205 WIKI = Entity( 

206 name='Wiki', 

207 group_globs=[], 

208 project_directories=[], 

209 project_files=[ 

210 join('project.wiki.bundle'), 

211 ], 

212 ) 

213 

214 # Entity: Snippets 

215 SNIPPETS = Entity( 

216 name='Snippets', 

217 group_globs=[], 

218 project_directories=[ 

219 join('snippets'), 

220 ], 

221 project_files=[ 

222 join('tree', 'project', 'snippets.ndjson'), 

223 ], 

224 ) 

225 

226 # Entity: Releases 

227 RELEASES = Entity( 

228 name='Releases', 

229 group_globs=[], 

230 project_directories=[], 

231 project_files=[ 

232 join('tree', 'project', 'releases.ndjson'), 

233 ], 

234 ) 

235 

236 # Entity: Badges 

237 BADGES = Entity( 

238 name='Badges', 

239 group_globs=[], 

240 project_directories=[], 

241 project_files=[ 

242 join('tree', 'project', 'project_badges.ndjson'), 

243 ], 

244 ) 

245 

246 # Entity: Uploads (wiki / issues) 

247 UPLOADS = Entity( 

248 name='Uploads', 

249 group_globs=[], 

250 project_directories=[ 

251 join('uploads'), 

252 ], 

253 project_files=[], 

254 ) 

255 

256 # Remover: Issues 

257 REMOVER_ISSUES = Remover( 

258 name='Remove/Issues', 

259 group_globs=[ 

260 *MEMBERS.group_globs, 

261 *ISSUES.group_globs, 

262 *BOARDS.group_globs, 

263 *MILESTONES.group_globs, 

264 *LABELS.group_globs, 

265 *UPLOADS.group_globs, 

266 ], 

267 project_directories=[ 

268 *MEMBERS.project_directories, 

269 *ISSUES.project_directories, 

270 *BOARDS.project_directories, 

271 *MILESTONES.project_directories, 

272 *LABELS.project_directories, 

273 *UPLOADS.project_directories, 

274 ], 

275 project_files=[ 

276 *MEMBERS.project_files, 

277 *ISSUES.project_files, 

278 *BOARDS.project_files, 

279 *MILESTONES.project_files, 

280 *LABELS.project_files, 

281 *UPLOADS.project_files, 

282 ], 

283 ) 

284 

285 # Remover: Repository 

286 REMOVER_REPOSITORY = Remover( 

287 name='Remove/Repository', 

288 group_globs=[ 

289 *MEMBERS.group_globs, 

290 *MERGE_REQUESTS.group_globs, 

291 *LFS_OBJECTS.group_globs, 

292 *PIPELINES.group_globs, 

293 *RELEASES.group_globs, 

294 ], 

295 project_directories=[ 

296 *MEMBERS.project_directories, 

297 *MERGE_REQUESTS.project_directories, 

298 *LFS_OBJECTS.project_directories, 

299 *PIPELINES.project_directories, 

300 *RELEASES.project_directories, 

301 ], 

302 project_files=[ 

303 *MEMBERS.project_files, 

304 *MERGE_REQUESTS.project_files, 

305 *LFS_OBJECTS.project_files, 

306 *PIPELINES.project_files, 

307 *RELEASES.project_files, 

308 join('project.bundle'), 

309 join('tree', 'project', 'protected_branches.ndjson'), 

310 join('tree', 'project', 'protected_tags.ndjson'), 

311 ], 

312 ) 

313 

314 # Remover: Wiki 

315 REMOVER_WIKI = Remover( 

316 name='Remove/Wiki', 

317 group_globs=[ 

318 *MEMBERS.group_globs, 

319 *WIKI.group_globs, 

320 ], 

321 project_directories=[ 

322 *MEMBERS.project_directories, 

323 *WIKI.project_directories, 

324 ], 

325 project_files=[ 

326 *MEMBERS.project_files, 

327 *WIKI.project_files, 

328 ], 

329 ) 

330 

331 # Template: Issues 

332 TEMPLATE_ISSUES = Template( 

333 name='Template/Issues', 

334 group_globs=[ 

335 *MEMBERS.group_globs, 

336 *ISSUES.group_globs, 

337 *MILESTONES.group_globs, 

338 *UPLOADS.group_globs, 

339 ], 

340 project_directories=[ 

341 *MEMBERS.project_directories, 

342 *ISSUES.project_directories, 

343 *MILESTONES.project_directories, 

344 *UPLOADS.project_directories, 

345 ], 

346 project_files=[ 

347 *MEMBERS.project_files, 

348 *ISSUES.project_files, 

349 *MILESTONES.project_files, 

350 *UPLOADS.project_files, 

351 ], 

352 ) 

353 

354 # Template: Repository 

355 TEMPLATE_REPOSITORY = Template( 

356 name='Template/Repository', 

357 group_globs=[ 

358 *MEMBERS.group_globs, 

359 *MERGE_REQUESTS.group_globs, 

360 *PIPELINES.group_globs, 

361 *RELEASES.group_globs, 

362 ], 

363 project_directories=[ 

364 *MEMBERS.project_directories, 

365 *MERGE_REQUESTS.project_directories, 

366 *PIPELINES.project_directories, 

367 *RELEASES.project_directories, 

368 ], 

369 project_files=[ 

370 *MEMBERS.project_files, 

371 *MERGE_REQUESTS.project_files, 

372 *PIPELINES.project_files, 

373 *RELEASES.project_files, 

374 join('tree', 'project', 'protected_tags.ndjson'), 

375 ], 

376 ) 

377 

378# RunnerHelpers class, pylint: disable=too-few-public-methods 

379class RunnerHelpers: 

380 

381 # Active 

382 @staticmethod 

383 def active(runner: GroupRunner | ProjectRunner) -> bool: 

384 

385 # Detect inactive runner 

386 if not runner.active: 

387 return False 

388 

389 # Detect paused runner 

390 if (hasattr(runner, 'paused') and runner.paused) or runner.status == 'paused': 

391 return False 

392 

393 # Result 

394 return True