Coverage for src/updates2mqtt/config.py: 92%

1import os 

2import ssl 

3import typing 

4from dataclasses import dataclass, field 

5from enum import StrEnum 

6from pathlib import Path 

7 

8import structlog 

9from omegaconf import MISSING, DictConfig, MissingMandatoryValue, OmegaConf, ValidationError 

10 

11log = structlog.get_logger() 

12 

13PKG_INFO_FILE = Path("./common_packages.yaml") 

14UNKNOWN_VERSION = "UNKNOWN" 

15VERSION_RE = r"[vVr]?[0-9]+(\.[0-9]+)*" 

16# source: https://semver.org/#is-there-a-suggested-regular-expression-regex-to-check-a-semver-string 

17SEMVER_RE = r"^(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]\d*)(?:-(?P<prerelease>(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+(?P<buildmetadata>[0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$" # noqa: E501 

18 

19SOURCE_PLATFORM_GITHUB = "GitHub" 

20SOURCE_PLATFORM_CODEBERG = "CodeBerg" 

21SOURCE_PLATFORM_GITLAB = "GitLab" 

22SOURCE_PLATFORM_SOURCEHUT = "SourceHut" 

23SOURCE_PLATFORMS: dict[str, str] = { 

24 SOURCE_PLATFORM_GITHUB: r"https://github.com/.*", 

25 SOURCE_PLATFORM_GITLAB: r"https://gitlab.com/.*", 

26 SOURCE_PLATFORM_CODEBERG: r"https://codeberg.org/.*", 

27 SOURCE_PLATFORM_SOURCEHUT: r"https://sr.ht/.*", 

28} 

29 

30 

31class UpdatePolicy(StrEnum): 

32 AUTO = "Auto" 

33 PASSIVE = "Passive" 

34 

35 

36class PublishPolicy(StrEnum): 

37 HOMEASSISTANT = "HomeAssistant" 

38 MQTT = "MQTT" 

39 SILENT = "Silent" 

40 

41 

42class LogLevel(StrEnum): 

43 DEBUG = "DEBUG" 

44 INFO = "INFO" 

45 WARNING = "WARNING" 

46 ERROR = "ERROR" 

47 CRITICAL = "CRITICAL" 

48 

49 

50class RegistryAPI(StrEnum): 

51 OCI_V2 = "OCI_V2" 

52 OCI_V2_MINIMAL = "OCI_V2" 

53 DOCKER_CLIENT = "DOCKER_CLIENT" 

54 DISABLED = "DISABLED" 

55 

56 

57class VersionType: 

58 SHORT_SHA = "short_sha" 

59 FULL_SHA = "full_sha" 

60 VERSION_REVISION = "version_revision" 

61 VERSION = "version" 

62 

63 

64@dataclass 

65class RegistryConfig: 

66 api: RegistryAPI = RegistryAPI.OCI_V2 

67 mutable_cache_ttl: int | None = None # default to server cache hint 

68 immutable_cache_ttl: int | None = 7776000 # 90 days 

69 token_cache_ttl: int | None = None # default to server cache hint 

70 

71 

72@dataclass 

73class MqttConfig: 

74 host: str = "${oc.env:MQTT_HOST,localhost}" 

75 user: str = f"${{oc.env:MQTT_USER,{MISSING}}}" 

76 password: str | None = "${oc.env:MQTT_PASS,''}" # noqa: S105 

77 port: int = "${oc.decode:${oc.env:MQTT_PORT,1883}}" # type: ignore[assignment] 

78 topic_root: str = "updates2mqtt" 

79 protocol: str = "${oc.env:MQTT_VERSION,3.11}" 

80 connect_timeout: float = 20 

81 keepalive: int = 30 

82 ca_certs: str | None = "${oc.env:MQTT_CA_CERTS,''}" 

83 client_cert: str | None = "${oc.env:MQTT_CLIENT_CERT,''}" 

84 client_key: str | None = "${oc.env:MQTT_CLIENT_KEY,''}" 

85 cert_reqs: ssl.VerifyMode = ssl.CERT_REQUIRED 

86 

87 

88@dataclass 

89class GitHubConfig: 

90 access_token: str | None = None 

91 mutable_cache_ttl: int = 60 * 60 * 15 

92 

93 

94@dataclass 

95class MetadataSourceConfig: 

96 enabled: bool = True 

97 cache_ttl: int = 60 * 60 * 24 * 7 # 1 week 

98 

99 

100@dataclass 

101class Selector: 

102 include: list[str] | None = None 

103 exclude: list[str] | None = None 

104 

105 

106class VersionPolicy(StrEnum): 

107 AUTO = "AUTO" 

108 VERSION = "VERSION" 

109 DIGEST = "DIGEST" 

110 VERSION_DIGEST = "VERSION_DIGEST" 

111 TIMESTAMP = "TIMESTAMP" 

112 

113 

114@dataclass 

115class DockerPackageUpdateInfo: 

116 image_name: typing.Any = MISSING # untagged image ref, either a single string or a list of strings 

117 version_policy: VersionPolicy = VersionPolicy.AUTO 

118 

119 

120def docker_image_names(docker_info: DockerPackageUpdateInfo) -> list[str]: 

121 if isinstance(docker_info.image_name, str): 

122 return [docker_info.image_name] 

123 return list(docker_info.image_name) 

124 

125 

126@dataclass 

127class PackageUpdateInfo: 

128 docker: DockerPackageUpdateInfo | None = field(default_factory=DockerPackageUpdateInfo) 

129 logo_url: str | None = None 

130 release_notes_url: str | None = None 

131 source_repo_url: str | None = None 

132 

133 

134@dataclass 

135class DockerConfig: 

136 enabled: bool = True 

137 allow_pull: bool = True 

138 allow_restart: bool = True 

139 allow_build: bool = True 

140 compose_version: str = "v2" 

141 default_entity_picture_url: str = "https://www.docker.com/wp-content/uploads/2022/03/Moby-logo.png" 

142 # Icon to show when browsing entities in Home Assistant 

143 device_icon: str = "mdi:docker" 

144 discover_metadata: dict[str, MetadataSourceConfig] = field( 

145 default_factory=lambda: {"linuxserver.io": MetadataSourceConfig(enabled=True)} 

146 ) 

147 registry: RegistryConfig = field(default_factory=lambda: RegistryConfig()) 

148 default_api_backoff: int = 60 * 15 

149 image_ref_select: Selector = field(default_factory=lambda: Selector()) 

150 version_select: Selector = field(default_factory=lambda: Selector()) 

151 version_policy: VersionPolicy = VersionPolicy.AUTO 

152 registry_select: Selector = field(default_factory=lambda: Selector()) 

153 

154 

155@dataclass 

156class HomeAssistantDiscoveryConfig: 

157 prefix: str = "homeassistant" 

158 enabled: bool = True 

159 

160 

161@dataclass 

162class HomeAssistantConfig: 

163 discovery: HomeAssistantDiscoveryConfig = field(default_factory=HomeAssistantDiscoveryConfig) 

164 state_topic_suffix: str = "state" 

165 device_creation: bool = True 

166 force_command_topic: bool = False 

167 extra_attributes: bool = True 

168 area: str | None = None 

169 release_summary_max_size: int = 6144 

170 

171 

172@dataclass 

173class HealthCheckConfig: 

174 enabled: bool = True 

175 interval: int = 300 # Interval in seconds to publish heartbeat message, 0 to disable 

176 topic_template: str = "healthcheck/{node_name}/updates2mqtt" 

177 

178 

179@dataclass 

180class NodeConfig: 

181 name: str = field(default_factory=lambda: os.getenv("NODE_NAME", os.uname().nodename.replace(".local", ""))) 

182 git_path: str = "/usr/bin/git" 

183 healthcheck: HealthCheckConfig = field(default_factory=HealthCheckConfig) 

184 

185 

186@dataclass 

187class LogConfig: 

188 level: LogLevel = "${oc.decode:${oc.env:U2M_LOG_LEVEL,INFO}}" # type: ignore[assignment] # pyright: ignore[reportAssignmentType] 

189 json: bool = field(default=False, doc="Use JSON structured logging for non-interactive running") 

190 

191 

192@dataclass 

193class Config: 

194 log: LogConfig = field(default_factory=LogConfig) # pyright: ignore[reportArgumentType, reportCallIssue] 

195 node: NodeConfig = field(default_factory=NodeConfig) 

196 mqtt: MqttConfig = field(default_factory=MqttConfig) # pyright: ignore[reportArgumentType, reportCallIssue] 

197 homeassistant: HomeAssistantConfig = field(default_factory=HomeAssistantConfig) 

198 docker: DockerConfig = field(default_factory=DockerConfig) 

199 github: GitHubConfig = field(default_factory=GitHubConfig) 

200 scan_interval: int = 60 * 60 * 3 

201 packages: dict[str, PackageUpdateInfo] = field(default_factory=dict) 

202 

203 

204@dataclass 

205class CommonPackages: 

206 common_packages: dict[str, PackageUpdateInfo] = field(default_factory=dict) 

207 

208 

209class IncompleteConfigException(BaseException): 

210 pass 

211 

212 

213def is_autogen_config() -> bool: 

214 env_var: str | None = os.environ.get("U2M_AUTOGEN_CONFIG") 

215 return not (env_var and env_var.lower() in ("no", "0", "false")) 

216 

217 

218def load_app_config(conf_file_path: Path, return_invalid: bool = False) -> Config | None: 

219 base_cfg: DictConfig = OmegaConf.structured(Config) 

220 if conf_file_path.exists(): 

221 cfg: DictConfig = typing.cast("DictConfig", OmegaConf.merge(base_cfg, OmegaConf.load(conf_file_path))) 

222 elif is_autogen_config(): 

223 if not conf_file_path.parent.exists(): 223 ↛ 224line 223 didn't jump to line 224 because the condition on line 223 was never true

224 try: 

225 log.debug(f"Creating config directory {conf_file_path.parent} if not already present") 

226 conf_file_path.parent.mkdir(parents=True, exist_ok=True) 

227 except Exception as e: 

228 log.warning("Unable to create config directory: %s", e, path=conf_file_path.parent) 

229 try: 

230 conf_file_path.write_text(OmegaConf.to_yaml(base_cfg)) 

231 log.info(f"Auto-generated a new config file at {conf_file_path}") 

232 except Exception as e: 

233 log.warning("Unable to write config file: %s", e, path=conf_file_path) 

234 cfg = base_cfg 

235 else: 

236 cfg = base_cfg 

237 

238 try: 

239 # Validate that all required fields are present, throw exception now rather than when config first used 

240 OmegaConf.to_container(cfg, throw_on_missing=True) 

241 OmegaConf.set_readonly(cfg, True) 

242 config: Config = typing.cast("Config", cfg) 

243 

244 if config.mqtt.user in ("", MISSING): 

245 log.info("The config has place holders for MQTT user") 

246 if not return_invalid: 246 ↛ 247line 246 didn't jump to line 247 because the condition on line 246 was never true

247 return None 

248 return config 

249 except (MissingMandatoryValue, ValidationError) as e: 

250 log.error("Configuration error %s", e, path=conf_file_path.as_posix()) 

251 if return_invalid and cfg is not None: 

252 return typing.cast("Config", cfg) 

253 raise