Coverage for /builds/svesch/cikit/src/cikit/env/provider.py: 99%

1# SPDX-License-Identifier: GPL-3.0-or-later 

2"""CI provider abstraction for cikit. 

3 

4This module provides provider-specific CI and VCS metadata access. 

5It separates runtime configuration from provider-derived context such as 

6pipeline identifiers, branch names and commit metadata. 

7 

8Author: 

9 Sven Schlender 

10Date: 

11 2026-05-29 

12""" 

13 

14import os 

15import re 

16import time 

17import uuid 

18from abc import ABC 

19from enum import Enum 

20from pathlib import Path 

21 

22from cikit.command import run_command 

23from cikit.errors import CiToolsError 

24 

25 

26class CIProvider(Enum): 

27 """Supported CI providers. 

28 

29 The enum values identify the currently active CI or local provider. 

30 They are used for provider comparisons and provider-specific behavior 

31 in macros and logging helpers. 

32 """ 

33 

34 GITLAB = "gitlab" 

35 LOCALGIT = "localgit" 

36 AZURE_DEVOPS = "azure_devops" 

37 

38 

39def _clean_section_name(title: str) -> str: 

40 """Return a CI-safe section name. 

41 

42 The title is normalized to lowercase and all characters not accepted by 

43 common CI section markers are replaced by underscores. Leading and trailing 

44 underscores are removed to keep generated section identifiers compact. 

45 

46 Args: 

47 title: Human-readable section title. 

48 

49 Returns: 

50 Normalized section name. 

51 """ 

52 name = title.lower() 

53 name = re.sub(r"[^a-z0-9_.-]+", "_", name) 

54 

55 return name.strip("_") 

56 

57 

58def _run_git(repo_root: Path, args: list) -> str: 

59 """Run one Git command in the given repository. 

60 

61 The command is executed through the shared command runner so that process 

62 handling and error reporting stay consistent with the rest of cikit. 

63 

64 Args: 

65 repo_root: Repository directory used as command working directory. 

66 args: Git arguments without the leading ``git`` executable. 

67 

68 Returns: 

69 Stripped stdout returned by the Git command. 

70 

71 Raises: 

72 CiToolsError: If the Git command cannot be started or exits with an 

73 error code. 

74 """ 

75 result = run_command(["git", *args], cwd=repo_root, live_output=False) 

76 

77 return result.stdout.strip() 

78 

79 

80class BaseProvider(ABC): 

81 """Base class for CI providers. 

82 

83 Providers expose CI and VCS context through explicit properties. The base 

84 implementation returns empty or neutral fallback values so callers can use 

85 the provider interface without checking for optional capabilities first. 

86 """ 

87 

88 name: CIProvider = CIProvider.LOCALGIT 

89 

90 @property 

91 def provider_scope(self) -> str: 

92 """Return the provider-specific project scope. 

93 

94 Returns: 

95 Provider-specific project scope or an empty string if unavailable. 

96 """ 

97 return "" 

98 

99 @property 

100 def provider_project(self) -> str: 

101 """Return the provider-specific project name. 

102 

103 Returns: 

104 Provider-specific project name or an empty string if unavailable. 

105 """ 

106 return "" 

107 

108 @property 

109 def pipeline_id(self) -> str: 

110 """Return the provider-specific pipeline identifier. 

111 

112 Returns: 

113 Pipeline identifier or an empty string if unavailable. 

114 """ 

115 return "" 

116 

117 @property 

118 def pipeline_alias(self) -> str: 

119 """Return the provider-specific human-readable pipeline identifier. 

120 

121 Returns: 

122 Human-readable pipeline identifier or ``pipeline_id`` if no separate 

123 alias is available. 

124 """ 

125 return self.pipeline_id 

126 

127 @property 

128 def branch_name(self) -> str: 

129 """Return the active branch name. 

130 

131 Returns: 

132 Branch name or ``DETACHED`` if no branch can be resolved. 

133 """ 

134 return "DETACHED" 

135 

136 @property 

137 def commit_sha(self) -> str: 

138 """Return the full commit SHA. 

139 

140 Returns: 

141 Full commit SHA or an empty string if unavailable. 

142 """ 

143 return "" 

144 

145 @property 

146 def commit_short_sha(self) -> str: 

147 """Return the short commit SHA. 

148 

149 Returns: 

150 Short commit SHA or an empty string if unavailable. 

151 """ 

152 return "" 

153 

154 @property 

155 def commit_title(self) -> str: 

156 """Return the commit title. 

157 

158 Returns: 

159 First line of the commit message or an empty string if unavailable. 

160 """ 

161 return "" 

162 

163 @property 

164 def commit_message(self) -> str: 

165 """Return the full commit message. 

166 

167 Returns: 

168 Full commit message or an empty string if unavailable. 

169 """ 

170 return "" 

171 

172 @property 

173 def commit_author(self) -> str: 

174 """Return the commit author. 

175 

176 Returns: 

177 Commit author name or an empty string if unavailable. 

178 """ 

179 return "" 

180 

181 @property 

182 def commit_timestamp(self) -> str: 

183 """Return the commit timestamp. 

184 

185 Returns: 

186 Commit timestamp or an empty string if unavailable. 

187 """ 

188 return "" 

189 

190 def section_start(self, _title: str, _collapsed: bool = False) -> str: 

191 """Return a provider-specific section start marker. 

192 

193 Providers that support collapsible CI log sections can override this 

194 method. The default implementation returns an empty string and therefore 

195 disables provider-specific section markup. 

196 

197 Args: 

198 _title: Human-readable section title. 

199 _collapsed: Whether the section should start collapsed. 

200 

201 Returns: 

202 Provider-specific section start marker or an empty string. 

203 """ 

204 return "" 

205 

206 def section_end(self, _title: str) -> str: 

207 """Return a provider-specific section end marker. 

208 

209 Providers that support CI log sections can override this method. The 

210 default implementation returns an empty string and therefore disables 

211 provider-specific section markup. 

212 

213 Args: 

214 _title: Human-readable section title. 

215 

216 Returns: 

217 Provider-specific section end marker or an empty string. 

218 """ 

219 return "" 

220 

221 def log_prefix(self) -> str: 

222 """Return a provider-specific log prefix. 

223 

224 The prefix can be used by logging helpers to indent or mark output 

225 inside provider-specific sections. The default implementation returns 

226 an empty string and therefore does not modify log output. 

227 

228 Returns: 

229 Provider-specific log prefix or an empty string. 

230 """ 

231 return "" 

232 

233 def __str__(self) -> str: 

234 """Return the provider name. 

235 

236 Returns: 

237 Uppercase enum name of the provider. 

238 """ 

239 return self.name.name 

240 

241 def __eq__(self, other): 

242 """Compare this provider with a CIProvider enum value. 

243 

244 Args: 

245 other: Object to compare with this provider. 

246 

247 Returns: 

248 True if ``other`` matches this provider's ``CIProvider`` value. 

249 NotImplemented if the comparison type is unsupported. 

250 """ 

251 if isinstance(other, CIProvider): 

252 return self.name == other 

253 

254 return NotImplemented 

255 

256 

257class LocalGitProvider(BaseProvider): 

258 """Local Git provider with metadata fallback from the current repository. 

259 

260 This provider is used outside a supported CI environment. It derives 

261 pipeline, branch and commit context from the local Git repository when 

262 possible. Missing Git metadata is treated as unavailable instead of failing 

263 local CI helper execution. 

264 """ 

265 

266 name = CIProvider.LOCALGIT 

267 

268 def __init__(self, repo_root: Path = None): 

269 """Initialize the local Git provider. 

270 

271 Args: 

272 repo_root: Optional repository root. If omitted, the current working 

273 directory is used as Git command working directory. 

274 """ 

275 self.repo_root = Path(repo_root or os.getcwd()).resolve() 

276 self._git_metadata = None 

277 

278 @property 

279 def pipeline_id(self) -> str: 

280 """Return a generated local pipeline id. 

281 

282 The id is generated lazily and cached in the process environment for 

283 the lifetime of the current local execution. 

284 

285 Returns: 

286 Generated local pipeline identifier. 

287 """ 

288 pipeline_id = os.getenv("CIKIT_PIPELINE_ID") 

289 if pipeline_id: 

290 return pipeline_id 

291 

292 timestamp = int(time.time()) 

293 short = uuid.uuid4().hex[:8] 

294 pipeline_id = f"auto-{timestamp}-{short}" 

295 os.environ["CIKIT_PIPELINE_ID"] = pipeline_id 

296 

297 return pipeline_id 

298 

299 @property 

300 def branch_name(self) -> str: 

301 """Return the local Git branch name. 

302 

303 Returns: 

304 Current Git branch name or ``DETACHED`` if unavailable. 

305 """ 

306 return self._git_value("branch_name", "DETACHED") 

307 

308 @property 

309 def commit_sha(self) -> str: 

310 """Return the local Git commit SHA. 

311 

312 Returns: 

313 Full commit SHA or an empty string if unavailable. 

314 """ 

315 return self._git_value("commit_sha") 

316 

317 @property 

318 def commit_short_sha(self) -> str: 

319 """Return the local Git short commit SHA. 

320 

321 Returns: 

322 Short commit SHA or an empty string if unavailable. 

323 """ 

324 return self._git_value("commit_short_sha") 

325 

326 @property 

327 def commit_title(self) -> str: 

328 """Return the local Git commit title. 

329 

330 Returns: 

331 First line of the current commit message or an empty string if 

332 unavailable. 

333 """ 

334 return self._git_value("commit_title") 

335 

336 @property 

337 def commit_message(self) -> str: 

338 """Return the local Git commit message. 

339 

340 Returns: 

341 Full current commit message or an empty string if unavailable. 

342 """ 

343 return self._git_value("commit_message") 

344 

345 @property 

346 def commit_author(self) -> str: 

347 """Return the local Git commit author. 

348 

349 Returns: 

350 Current commit author or an empty string if unavailable. 

351 """ 

352 return self._git_value("commit_author") 

353 

354 @property 

355 def commit_timestamp(self) -> str: 

356 """Return the local Git commit timestamp. 

357 

358 Returns: 

359 Current commit timestamp or an empty string if unavailable. 

360 """ 

361 return self._git_value("commit_timestamp") 

362 

363 def log_prefix(self) -> str: 

364 """Return the local log prefix. 

365 

366 Returns: 

367 Prefix used to indent local section output. 

368 """ 

369 return " " 

370 

371 def _git_value(self, key: str, default: str = "") -> str: 

372 """Return one cached Git metadata value. 

373 

374 Args: 

375 key: Metadata field name. 

376 default: Value returned if the field is unavailable. 

377 

378 Returns: 

379 Cached metadata value or default. 

380 """ 

381 metadata = self._load_git_metadata() 

382 

383 return metadata.get(key, default) 

384 

385 def _load_git_metadata(self) -> dict: 

386 """Load and cache Git metadata from the repository. 

387 

388 Metadata is loaded only once. Subsequent calls return the cached 

389 dictionary. If Git metadata cannot be obtained, an empty dictionary is 

390 cached and returned. 

391 

392 Returns: 

393 Dictionary containing cached Git metadata. 

394 """ 

395 if self._git_metadata is not None: 

396 return self._git_metadata 

397 

398 try: 

399 self._git_metadata = { 

400 "branch_name": _run_git(self.repo_root, ["rev-parse", "--abbrev-ref", "HEAD"]), 

401 "commit_sha": _run_git(self.repo_root, ["log", "-1", "--pretty=format:%H"]), 

402 "commit_short_sha": _run_git(self.repo_root, ["log", "-1", "--pretty=format:%h"]), 

403 "commit_title": _run_git(self.repo_root, ["log", "-1", "--pretty=format:%s"]), 

404 "commit_message": _run_git(self.repo_root, ["log", "-1", "--pretty=format:%B"]), 

405 "commit_author": _run_git(self.repo_root, ["log", "-1", "--pretty=format:%an"]), 

406 "commit_timestamp": _run_git(self.repo_root, ["log", "-1", "--pretty=format:%aI"]), 

407 } 

408 except CiToolsError: 

409 self._git_metadata = {} 

410 

411 return self._git_metadata 

412 

413 

414class GitLabProvider(BaseProvider): 

415 """GitLab CI provider. 

416 

417 This provider reads CI and commit context from GitLab CI environment 

418 variables. It also provides GitLab section markers for collapsible CI log 

419 sections. 

420 """ 

421 

422 name = CIProvider.GITLAB 

423 

424 @property 

425 def provider_scope(self) -> str: 

426 """Return the GitLab project namespace. 

427 

428 Returns: 

429 GitLab project namespace or an empty string if unavailable. 

430 """ 

431 return os.getenv("CI_PROJECT_NAMESPACE", "") 

432 

433 @property 

434 def provider_project(self) -> str: 

435 """Return the GitLab project name. 

436 

437 Returns: 

438 GitLab project name or an empty string if unavailable. 

439 """ 

440 return os.getenv("CI_PROJECT_NAME", "") 

441 

442 @property 

443 def pipeline_id(self) -> str: 

444 """Return the GitLab pipeline id. 

445 

446 Returns: 

447 GitLab pipeline id or an empty string if unavailable. 

448 """ 

449 return os.getenv("CI_PIPELINE_ID", "") 

450 

451 @property 

452 def branch_name(self) -> str: 

453 """Return the GitLab branch name. 

454 

455 Returns: 

456 GitLab commit ref name or ``DETACHED`` if unavailable. 

457 """ 

458 return os.getenv("CI_COMMIT_REF_NAME", "DETACHED") 

459 

460 @property 

461 def commit_sha(self) -> str: 

462 """Return the GitLab commit SHA. 

463 

464 Returns: 

465 Full GitLab commit SHA or an empty string if unavailable. 

466 """ 

467 return os.getenv("CI_COMMIT_SHA", "") 

468 

469 @property 

470 def commit_short_sha(self) -> str: 

471 """Return the GitLab short commit SHA. 

472 

473 Returns: 

474 Short GitLab commit SHA or an empty string if unavailable. 

475 """ 

476 return os.getenv("CI_COMMIT_SHORT_SHA", "") 

477 

478 @property 

479 def commit_title(self) -> str: 

480 """Return the GitLab commit title. 

481 

482 Returns: 

483 GitLab commit title or an empty string if unavailable. 

484 """ 

485 return os.getenv("CI_COMMIT_TITLE", "") 

486 

487 @property 

488 def commit_message(self) -> str: 

489 """Return the GitLab commit message. 

490 

491 Returns: 

492 GitLab commit message or an empty string if unavailable. 

493 """ 

494 return os.getenv("CI_COMMIT_MESSAGE", "") 

495 

496 @property 

497 def commit_author(self) -> str: 

498 """Return the GitLab commit author. 

499 

500 Returns: 

501 GitLab commit author or an empty string if unavailable. 

502 """ 

503 return os.getenv("CI_COMMIT_AUTHOR", "") 

504 

505 @property 

506 def commit_timestamp(self) -> str: 

507 """Return the GitLab commit timestamp. 

508 

509 Returns: 

510 GitLab commit timestamp or an empty string if unavailable. 

511 """ 

512 return os.getenv("CI_COMMIT_TIMESTAMP", "") 

513 

514 def section_start(self, title: str, collapsed: bool = False) -> str: 

515 """Return the GitLab section start marker. 

516 

517 Args: 

518 title: Human-readable section title. 

519 collapsed: Whether the section should start collapsed. 

520 

521 Returns: 

522 GitLab-compatible section start marker. 

523 """ 

524 timestamp = int(time.time()) 

525 flag = "[collapsed=true]" if collapsed else "" 

526 name = _clean_section_name(title) 

527 

528 return f"section_start:{timestamp}:{name}{flag}\r\033[0K{title}" 

529 

530 def section_end(self, title: str) -> str: 

531 """Return the GitLab section end marker. 

532 

533 Args: 

534 title: Human-readable section title. 

535 

536 Returns: 

537 GitLab-compatible section end marker. 

538 """ 

539 timestamp = int(time.time()) 

540 name = _clean_section_name(title) 

541 

542 return f"section_end:{timestamp}:{name}\r\033[0K" 

543 

544 

545class AzureDevOpsProvider(LocalGitProvider): 

546 """Azure DevOps provider with Git metadata from the checked-out repository. 

547 

548 Azure DevOps provides pipeline metadata through environment variables. Git 

549 metadata is inherited from ``LocalGitProvider`` and resolved from the local 

550 checkout instead of Azure-specific environment variables. 

551 """ 

552 

553 name = CIProvider.AZURE_DEVOPS 

554 

555 @property 

556 def provider_scope(self) -> str: 

557 """Return the Azure DevOps project name. 

558 

559 Returns: 

560 Azure DevOps project name from ``SYSTEM_TEAMPROJECT`` or an empty 

561 string if unavailable. 

562 """ 

563 return os.getenv("SYSTEM_TEAMPROJECT", "") 

564 

565 @property 

566 def provider_project(self) -> str: 

567 """Return the Azure DevOps repository name. 

568 

569 Returns: 

570 Azure DevOps repository name from ``BUILD_REPOSITORY_NAME`` or an 

571 empty string if unavailable. 

572 """ 

573 return os.getenv("BUILD_REPOSITORY_NAME", "") 

574 

575 @property 

576 def pipeline_id(self) -> str: 

577 """Return the Azure DevOps technical pipeline run id. 

578 

579 Returns: 

580 Pipeline id from ``BUILD_BUILDID``. 

581 """ 

582 return os.getenv("BUILD_BUILDID", "") 

583 

584 @property 

585 def pipeline_alias(self) -> str: 

586 """Return the Azure DevOps human-readable pipeline number. 

587 

588 Returns: 

589 Pipeline alias ``BUILD_BUILDNUMBER``. 

590 If unavailable, ``pipeline_id`` is used. 

591 """ 

592 return os.getenv("BUILD_BUILDNUMBER", "") or self.pipeline_id 

593 

594 @property 

595 def branch_name(self) -> str: 

596 """Return the Azure DevOps source branch name.""" 

597 source_branch = os.getenv("BUILD_SOURCEBRANCH", "") 

598 

599 if source_branch.startswith("refs/heads/"): 

600 return source_branch.removeprefix("refs/heads/") 

601 

602 if source_branch.startswith("refs/tags/"): 

603 return source_branch.removeprefix("refs/tags/") 

604 

605 return os.getenv("BUILD_SOURCEBRANCHNAME") or super().branch_name 

606 

607 def section_start(self, title: str, collapsed: bool = False) -> str: 

608 """Return the Azure DevOps log group start marker. 

609 

610 Azure DevOps supports collapsible log groups through logging commands. 

611 The provider interface exposes the ``collapsed`` argument for compatibility 

612 with other CI providers, but Azure DevOps does not support controlling the 

613 initial collapsed state through the group marker. 

614 

615 Args: 

616 title: Human-readable section title. 

617 collapsed: Ignored for Azure DevOps. 

618 

619 Returns: 

620 Azure DevOps-compatible log group start marker. 

621 """ 

622 del collapsed 

623 return f"##[group]{title}" 

624 

625 

626 def section_end(self, _title: str) -> str: 

627 """Return the Azure DevOps log group end marker. 

628 

629 Args: 

630 _title: Human-readable section title. The value is unused because 

631 Azure DevOps group end markers do not include the title. 

632 

633 Returns: 

634 Azure DevOps-compatible log group end marker. 

635 """ 

636 return "##[endgroup]" 

637 

638 

639def detect_provider(repo_root: Path = None) -> BaseProvider: 

640 """Detect and return the active CI provider. 

641 

642 GitLab CI is detected through the ``GITLAB_CI`` environment variable. If no 

643 supported CI environment is detected, a local Git provider is returned. 

644 

645 Args: 

646 repo_root: Optional repository root used by ``LocalGitProvider``. 

647 

648 Returns: 

649 Detected provider instance. 

650 """ 

651 if os.getenv("GITLAB_CI", "").lower() == "true": 

652 return GitLabProvider() 

653 

654 if os.getenv("TF_BUILD", "").lower() == "true": 

655 return AzureDevOpsProvider(repo_root) 

656 

657 return LocalGitProvider(repo_root)