Coverage for /builds/svesch/cikit/src/cikit/macros/common.py: 100%

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

2"""Common macro helpers for cikit. 

3 

4This module provides shared CI macro functions for environment creation, 

5basic badge generation, cikit data initialization, timeline chart creation, 

6directory comparison, and name/label resolution. 

7 

8Author: 

9 Sven Schlender 

10Date: 

11 2026-05-28 

12""" 

13 

14import logging 

15from pathlib import Path 

16 

17from cikit.artifacts import Artifact 

18from cikit.badge import write_badge 

19from cikit.command import run_command 

20from cikit.data import RefKey 

21from cikit.env import CiEnvironment, CiRunEnvironment, get_env_vars 

22from cikit.factory import build_cikit_data_from_env 

23from cikit.timeline import TimelineChart 

24 

25 

26LOGGER = logging.getLogger(__name__) 

27 

28 

29def init_env(run_name: str = None, env_vars=None): 

30 """Create a resolved CI environment. 

31 

32 If a run name is provided, the returned environment includes a run-bound 

33 cikit data context. Without a run name, a generic CI environment is 

34 created. 

35 

36 Args: 

37 run_name: Optional logical cikit run name. 

38 env_vars: Optional environment variable definitions. 

39 

40 Returns: 

41 Resolved CI environment instance. 

42 """ 

43 if env_vars is None: 

44 env_vars = get_env_vars() 

45 

46 if run_name is not None: 

47 LOGGER.info("Preparing '%s' run...", run_name) 

48 return CiRunEnvironment(env_vars, run_name=run_name) 

49 

50 return CiEnvironment(env_vars) 

51 

52 

53def python_badge(env) -> Artifact: 

54 """Generate a Python version badge. 

55 

56 Args: 

57 env: Resolved CI environment. 

58 """ 

59 return write_badge(env, "python", label="python", value=env["PYTHON_VERSION"], color="green") 

60 

61 

62def ai_assisted_badge(env, value) -> Artifact: 

63 """Generate an AI assistance badge. 

64 

65 Args: 

66 env: Resolved CI environment. 

67 value: Badge value text. 

68 """ 

69 return write_badge(env, "ai_assisted", label="AI", value=value, color="blueviolet") 

70 

71 

72def data_prepare(env): 

73 """Initialize cikit data storage for the current pipeline. 

74 

75 This macro prepares cikit data for the current pipeline by creating the 

76 pipeline entry in the configured cikit data backend. It should typically 

77 be called once at the beginning of a CI run sequence. 

78 

79 Args: 

80 env: Resolved CI environment. 

81 """ 

82 ci_provider = str(env.provider) 

83 pipeline_id, cikit_data = build_cikit_data_from_env(env) 

84 cikit_data.create(pipeline_id, ci_provider, env.commit_sha) 

85 

86 

87def chart_timeline(env, *, keys, title: str, key_ref: RefKey = RefKey.INDEX_DESC, 

88 limit: int = None) -> TimelineChart: 

89 """Create a timeline chart object from cikit data. 

90 

91 This macro fetches timeline data from the CI environment and returns a 

92 :class:`TimelineChart` instance. The returned object can be further 

93 customized before rendering. 

94 

95 Args: 

96 env: Active CI run environment. 

97 keys: Cikit data keys to query for the timeline series. 

98 title: Title of the generated chart. 

99 key_ref: Reference key used for the X-axis. 

100 limit: Optional limit for the count of timeline entries. 

101 

102 Returns: 

103 Configurable timeline chart object. 

104 

105 Raises: 

106 CiRecoverableError: If the timeline data cannot be retrieved. 

107 CiToolsError: If the returned timeline data is invalid. 

108 """ 

109 timeline_data = env.timeline(keys, key_ref, limit) 

110 

111 return TimelineChart(timeline_data, title) 

112 

113 

114def compare_dirs(expected_dir: Path, actual_dir: Path) -> str: 

115 """Compare two directories using rsync dry-run mode. 

116 

117 Args: 

118 expected_dir: Directory containing the expected files. 

119 actual_dir: Directory containing the generated files. 

120 

121 Returns: 

122 Rsync itemized output. An empty string means both directories are 

123 identical. 

124 """ 

125 args = ["rsync", "-rcn", "--delete", "--itemize-changes", f"{expected_dir}/", f"{actual_dir}/"] 

126 result = run_command(args, live_output=False) 

127 

128 return result.stdout 

129 

130 

131def resolve_name_label(env, name: str = None, label: str = None) -> tuple: 

132 """Resolve default name and label from the run name. 

133 

134 Args: 

135 env: Resolved CI run environment. 

136 name: Optional output or badge name. 

137 label: Optional display label. 

138 

139 Returns: 

140 Resolved name and label. 

141 

142 Raises: 

143 AttributeError: If neither name nor label is provided and the 

144 environment has no run name. 

145 """ 

146 if name is None: 

147 name = env.name 

148 if label is None: 

149 label = env.name 

150 

151 return name, label