Coverage for /builds/svesch/cikit/src/cikit/logging_utils.py: 81%

55 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-10-09 14:17 +0000

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

2"""Logging setup and CI-oriented output helpers. 

3 

4This module provides a consistent logging configuration for local execution 

5and CI environments. It configures console logging, integrates provider- 

6specific formatting, and contains utilities used by CI commands to structure 

7their output. 

8 

9The central component is :class:`CiSection`, a context manager that emits 

10clearly formatted start and end markers around logical build steps. CI 

11providers may inject additional markers to support log folding or grouping 

12features offered by the hosting platform. 

13 

14Author: 

15 Sven Schlender 

16License: 

17 SPDX-License-Identifier: GPL-3.0-or-later 

18Date: 

19 2026-03-05 

20""" 

21 

22import sys 

23import time 

24import logging 

25 

26from cikit.errors import CiRecoverableError 

27from cikit.env import detect_provider 

28from cikit.status import CiRunStatus 

29 

30 

31LOGGER = logging.getLogger() 

32 

33 

34class CiSection: # pylint: disable=too-many-instance-attributes 

35 """Context manager for provider-specific CI log sections. 

36 

37 The context manager emits start and end markers for the active CI 

38 provider. Inside the section, an optional provider-defined message 

39 prefix can be injected into all active logging handlers. 

40 

41 Args: 

42 title: Section title shown in logs. 

43 collapsed: Whether the CI section should start collapsed. 

44 """ 

45 def __init__(self, title: str, *, collapsed: bool = False, status: CiRunStatus): 

46 """Initialize the CI section helper. 

47 

48 Args: 

49 title: Human-readable section title. 

50 collapsed: Whether the section should be collapsed initially. 

51 status: Run status collector used to aggregate exit codes. 

52 

53 """ 

54 self.title = title 

55 self.collapsed = collapsed 

56 self.provider = detect_provider() 

57 self._filter = None 

58 self._handlers = [] 

59 self._status = status 

60 self._start_time = None 

61 self._duration = None 

62 

63 def __enter__(self): 

64 """Open the CI section and attach provider-specific filters. 

65 

66 Returns: 

67 The active ``CiSection`` instance. 

68 """ 

69 self._start_time = time.monotonic() 

70 marker = self.provider.section_start(self.title, self.collapsed) 

71 if marker: 

72 LOGGER.info(marker) 

73 else: 

74 LOGGER.info("") 

75 LOGGER.info("=== %s ===", self.title) 

76 

77 return self 

78 

79 def __exit__(self, _exc_type, exc, _tb): 

80 """Close the CI section and remove temporary filters. 

81 

82 Args: 

83 _exc_type: Exception type passed by the context protocol. 

84 exc: Exception instance passed by the context protocol. 

85 _tb: Traceback object passed by the context protocol. 

86 """ 

87 self._duration = time.monotonic() - self._start_time 

88 

89 if isinstance(exc, CiRecoverableError): 

90 LOGGER.warning("%s", exc) 

91 self._status.set_exit_code(getattr(exc, "exit_code", 1)) 

92 

93 if self._filter: 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true

94 for handle in self._handlers: 

95 handle.removeFilter(self._filter) 

96 self._handlers = [] 

97 self._filter = None 

98 

99 end = self.provider.section_end(self.title) 

100 if end: 

101 LOGGER.info(end) 

102 

103 return isinstance(exc, CiRecoverableError) 

104 

105 @property 

106 def duration(self) -> float: 

107 """Return the measured section duration in seconds. 

108 

109 Returns: 

110 Section duration in seconds. 

111 """ 

112 return self._duration 

113 

114 

115class CiOptionalSection(CiSection): # pylint: disable=too-few-public-methods 

116 """CI section that ignores recoverable errors. 

117 

118 This section behaves like a regular CI section for successful execution. 

119 Recoverable errors are handled locally and do not change the global 

120 CI exit code. This is useful for optional report generation steps. 

121 """ 

122 

123 def __exit__(self, exc_type, exc_value, traceback): 

124 """Handle section exit. 

125 

126 Args: 

127 exc_type: Exception type or None. 

128 exc_value: Exception instance or None. 

129 traceback: Exception traceback or None. 

130 

131 Returns: 

132 True if a recoverable error was handled. 

133 """ 

134 if exc_type is not None and issubclass(exc_type, CiRecoverableError): 

135 LOGGER.warning("%s", exc_value) 

136 return True 

137 

138 return super().__exit__(exc_type, exc_value, traceback) 

139 

140 

141def setup_logging(verbose: bool = False): 

142 """Configure root logging for CI-friendly console output. 

143 

144 The setup installs a single stream handler writing to stdout and 

145 uses a formatter that prepends fixed-width level tags such as 

146 ``[INF]``, ``[WRN]`` and ``[ERR]``. 

147 

148 Args: 

149 verbose: If True, enable debug logging. 

150 

151 Returns: 

152 None. 

153 

154 Raises: 

155 None. 

156 """ 

157 logging.basicConfig( 

158 level=logging.DEBUG if verbose else logging.INFO, 

159 format="%(message)s", 

160 stream=sys.stdout, 

161 ) 

162 

163 

164def log_dict_compact(data: dict, *, level: int = logging.INFO, sort_keys: bool = True): 

165 """Pretty-print a dictionary for verbose output. 

166 

167 The dictionary is formatted as indented JSON and emitted line by line 

168 using :meth:`out`, ensuring consistent indentation and verbosity handling. 

169 

170 Args: 

171 data: Dictionary to print. 

172 level: Minimum verbosity level required for the output to be shown. 

173 sort_keys: If True, dictionary keys are sorted alphabetically. 

174 """ 

175 items = sorted(data.items()) if sort_keys else data.items() 

176 width = max((len(str(k)) for k, _ in items), default=0) 

177 for key, value in items: 

178 LOGGER.log(level, "%-*s : %s", width, key, value)