Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/infra/telemetry.py: 100%

1# SPDX-License-Identifier: MIT 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Provides concrete telemetry service implementations for event tracking. 

5 

6This module defines concrete classes that implement the `TelemetryProtocol`. 

7It offers different strategies for handling telemetry events, allowing the 

8application's analytics behavior to be configured easily. 

9 

10Key components include: 

11 * `TelemetryEvent`: An enumeration of all standardized event names, providing 

12 a single source of truth for telemetry event types. 

13 * `NullTelemetry`: A no-op implementation that silently discards all events, 

14 useful for disabling telemetry entirely. 

15 * `LoggingTelemetry`: An implementation that forwards all telemetry events to 

16 the application's structured logging service. 

17""" 

18 

19from __future__ import annotations 

20 

21from enum import Enum 

22from typing import Any 

23 

24from injector import inject 

25 

26from bijux_cli.contracts import ObservabilityProtocol, TelemetryProtocol 

27 

28 

29class TelemetryEvent(str, Enum): 

30 """Defines standardized telemetry event names for tracking CLI activities.""" 

31 

32 CLI_STARTED = "cli_started" 

33 CLI_ERROR = "cli_error" 

34 CLI_INTERRUPTED = "cli_interrupted" 

35 CLI_SYSTEM_EXIT = "cli_system_exit" 

36 CLI_UNEXPECTED_ERROR = "cli_unexpected_error" 

37 CLI_SHUTDOWN_FAILED = "cli_shutdown_failed" 

38 ENGINE_INITIALIZED = "engine_initialized" 

39 ENGINE_SHUTDOWN = "engine_shutdown" 

40 PLUGINS_LIST_COMMAND = "cmd/plugins/list" 

41 PLUGINS_LIST_COMMAND_FAILED = "cmd/err/plugins/list" 

42 PLUGINS_INFO_COMMAND = "cmd/plugins/info" 

43 PLUGINS_INFO_COMMAND_FAILED = "cmd/err/plugins/info" 

44 PLUGINS_INFO_NOT_FOUND = "cmd/err/plugins/info/not_found" 

45 PLUGINS_INSTALL_COMMAND = "cmd/plugins/install" 

46 PLUGINS_INSTALL_COMMAND_FAILED = "cmd/err/plugins/install" 

47 PLUGINS_UNINSTALL_COMMAND = "cmd/plugins/uninstall" 

48 PLUGINS_UNINSTALL_COMMAND_FAILED = "cmd/err/plugins/uninstall" 

49 PLUGINS_UNINSTALL_NOT_FOUND = "cmd/err/plugins/uninstall/not_found" 

50 PLUGINS_CHECK_COMMAND = "cmd/plugins/check" 

51 PLUGINS_CHECK_COMMAND_FAILED = "cmd/err/plugins/check" 

52 PLUGINS_CHECK_NOT_FOUND = "cmd/err/plugins/check/not_found" 

53 PLUGINS_SCAFFOLD_COMMAND = "cmd/plugins/scaffold" 

54 PLUGINS_SCAFFOLD_COMMAND_FAILED = "cmd/err/plugins/scaffold" 

55 PLUGINS_SCAFFOLD_DIR_EXISTS = "cmd/err/plugins/scaffold/dir_exists" 

56 CONFIG_COMMAND = "cmd/config" 

57 CONFIG_COMMAND_FAILED = "cmd/err/config" 

58 AUDIT_COMMAND = "cmd/audit" 

59 AUDIT_COMMAND_FAILED = "cmd/err/audit" 

60 DOCTOR_COMMAND = "cmd/doctor" 

61 DOCTOR_COMMAND_FAILED = "cmd/err/doctor" 

62 VERSION_COMMAND = "cmd/version" 

63 VERSION_COMMAND_FAILED = "cmd/err/version" 

64 STATUS_COMMAND = "cmd/status" 

65 STATUS_COMMAND_FAILED = "cmd/err/status" 

66 SLEEP_COMMAND = "cmd/test/sleep" 

67 SLEEP_COMMAND_FAILED = "cmd/err/test/sleep" 

68 HISTORY_COMMAND = "cmd/history" 

69 HISTORY_COMMAND_FAILED = "cmd/err/history" 

70 REPL_COMMAND = "cmd/repl" 

71 REPL_EXIT = "cmd/repl/exit" 

72 REPL_COMMAND_NOT_FOUND = "cmd/err/repl/not_found" 

73 DEV_COMMAND_EXECUTED = "cmd/dev" 

74 DEV_COMMAND_FAILED = "cmd/err/dev" 

75 MEMORY_COMMAND_EXECUTED = "cmd/memory" 

76 MEMORY_COMMAND_FAILED = "cmd/err/memory" 

77 HELP_COMMAND = "cmd/help" 

78 HELP_COMMAND_FAILED = "cmd/err/help" 

79 PLUGIN_STARTED = "plugin_started" 

80 PLUGIN_SHUTDOWN = "plugin_shutdown" 

81 PLUGIN_INSTALLED = "plugin_installed" 

82 PLUGIN_LOADED = "plugin_loaded" 

83 PLUGIN_CLI_REGISTERED = "plugin_cli_registered" 

84 PLUGIN_LOAD_FAILED = "plugin_load_failed" 

85 

86 

87class NullTelemetry(TelemetryProtocol): 

88 """A no-op telemetry service that discards all events. 

89 

90 This implementation of `TelemetryProtocol` can be used to effectively 

91 disable analytics and event tracking. 

92 """ 

93 

94 def event(self, name: str | TelemetryEvent, payload: dict[str, Any]) -> None: 

95 """Discards the telemetry event. 

96 

97 Args: 

98 name (str | TelemetryEvent): The event name (ignored). 

99 payload (dict[str, Any]): The event data (ignored). 

100 

101 Returns: 

102 None: 

103 """ 

104 return 

105 

106 def flush(self) -> None: 

107 """Performs a no-op flush operation.""" 

108 return 

109 

110 def enable(self) -> None: 

111 """Performs a no-op enable operation.""" 

112 return 

113 

114 

115class LoggingTelemetry(TelemetryProtocol): 

116 """A telemetry service that logs events via the `Observability` service. 

117 

118 This implementation of `TelemetryProtocol` forwards all telemetry events 

119 to the structured logger as debug-level messages. 

120 

121 Attributes: 

122 _obs (ObservabilityProtocol): The logging service instance. 

123 _buffer (list): A buffer to store events (currently only cleared on flush). 

124 """ 

125 

126 @inject 

127 def __init__(self, observability: ObservabilityProtocol): 

128 """Initializes the `LoggingTelemetry` service. 

129 

130 Args: 

131 observability (ObservabilityProtocol): The service for logging events. 

132 """ 

133 self._obs = observability 

134 self._buffer: list[tuple[str, dict[str, Any]]] = [] 

135 

136 def event(self, name: str | TelemetryEvent, payload: dict[str, Any]) -> None: 

137 """Logs a telemetry event at the 'debug' level. 

138 

139 Args: 

140 name (str | TelemetryEvent): The event name or enum member. 

141 payload (dict[str, Any]): The event data dictionary. 

142 

143 Returns: 

144 None: 

145 """ 

146 event_name = name.value if isinstance(name, TelemetryEvent) else name 

147 self._obs.log("debug", f"Telemetry event: {event_name}", extra=payload) 

148 self._buffer.append((event_name, payload)) 

149 

150 def flush(self) -> None: 

151 """Clears the internal buffer of telemetry events.""" 

152 self._buffer.clear() 

153 

154 def enable(self) -> None: 

155 """Performs a no-op enable operation.""" 

156 return 

157 

158 

159__all__ = ["TelemetryEvent", "NullTelemetry", "LoggingTelemetry"]