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

1import os 

2import typing 

3from dataclasses import dataclass, field 

4from enum import StrEnum 

5from pathlib import Path 

6 

7import structlog 

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

9 

10log = structlog.get_logger() 

11 

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

13UNKNOWN_VERSION = "UNKNOWN" 

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

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

16SEMVER_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 

17 

18 

19class UpdatePolicy(StrEnum): 

20 AUTO = "Auto" 

21 PASSIVE = "Passive" 

22 

23 

24class PublishPolicy(StrEnum): 

25 HOMEASSISTANT = "HomeAssistant" 

26 MQTT = "MQTT" 

27 SILENT = "Silent" 

28 

29 

30class LogLevel(StrEnum): 

31 DEBUG = "DEBUG" 

32 INFO = "INFO" 

33 WARNING = "WARNING" 

34 ERROR = "ERROR" 

35 CRITICAL = "CRITICAL" 

36 

37 

38class RegistryAPI(StrEnum): 

39 OCI_V2 = "OCI_V2" 

40 OCI_V2_MINIMAL = "OCI_V2" 

41 DOCKER_CLIENT = "DOCKER_CLIENT" 

42 DISABLED = "DISABLED" 

43 

44 

45class VersionType: 

46 SHORT_SHA = "short_sha" 

47 FULL_SHA = "full_sha" 

48 VERSION_REVISION = "version_revision" 

49 VERSION = "version" 

50 

51 

52@dataclass 

53class RegistryConfig: 

54 api: RegistryAPI = RegistryAPI.OCI_V2 

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

56 immutable_cache_ttl: int | None = 7776000 # 90 days 

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

58 

59 

60@dataclass 

61class MqttConfig: 

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

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

64 password: str = f"${{oc.env:MQTT_PASS,{MISSING}}}" 

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

66 topic_root: str = "updates2mqtt" 

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

68 

69 

70@dataclass 

71class GitHubConfig: 

72 access_token: str | None = None 

73 

74 

75@dataclass 

76class MetadataSourceConfig: 

77 enabled: bool = True 

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

79 

80 

81@dataclass 

82class Selector: 

83 include: list[str] | None = None 

84 exclude: list[str] | None = None 

85 

86 

87class VersionPolicy(StrEnum): 

88 AUTO = "AUTO" 

89 VERSION = "VERSION" 

90 DIGEST = "DIGEST" 

91 VERSION_DIGEST = "VERSION_DIGEST" 

92 TIMESTAMP = "TIMESTAMP" 

93 

94 

95@dataclass 

96class DockerPackageUpdateInfo: 

97 image_name: str = MISSING # untagged image ref 

98 version_policy: VersionPolicy = VersionPolicy.AUTO 

99 

100 

101@dataclass 

102class PackageUpdateInfo: 

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

104 logo_url: str | None = None 

105 release_notes_url: str | None = None 

106 source_repo_url: str | None = None 

107 

108 

109@dataclass 

110class DockerConfig: 

111 enabled: bool = True 

112 allow_pull: bool = True 

113 allow_restart: bool = True 

114 allow_build: bool = True 

115 compose_version: str = "v2" 

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

117 # Icon to show when browsing entities in Home Assistant 

118 device_icon: str = "mdi:docker" 

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

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

121 ) 

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

123 default_api_backoff: int = 60 * 15 

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

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

126 version_policy: VersionPolicy = VersionPolicy.AUTO 

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

128 

129 

130@dataclass 

131class HomeAssistantDiscoveryConfig: 

132 prefix: str = "homeassistant" 

133 enabled: bool = True 

134 

135 

136@dataclass 

137class HomeAssistantConfig: 

138 discovery: HomeAssistantDiscoveryConfig = field(default_factory=HomeAssistantDiscoveryConfig) 

139 state_topic_suffix: str = "state" 

140 device_creation: bool = True 

141 force_command_topic: bool = False 

142 extra_attributes: bool = True 

143 area: str | None = None 

144 release_summary_max_size: int = 6144 

145 

146 

147@dataclass 

148class HealthCheckConfig: 

149 enabled: bool = True 

150 interval: int = 300 # Interval in seconds to publish healthcheck message, 0 to disable 

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

152 

153 

154@dataclass 

155class NodeConfig: 

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

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

158 healthcheck: HealthCheckConfig = field(default_factory=HealthCheckConfig) 

159 

160 

161@dataclass 

162class LogConfig: 

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

164 

165 

166@dataclass 

167class Config: 

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

169 node: NodeConfig = field(default_factory=NodeConfig) 

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

171 homeassistant: HomeAssistantConfig = field(default_factory=HomeAssistantConfig) 

172 docker: DockerConfig = field(default_factory=DockerConfig) 

173 github: GitHubConfig = field(default_factory=GitHubConfig) 

174 scan_interval: int = 60 * 60 * 3 

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

176 

177 

178@dataclass 

179class CommonPackages: 

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

181 

182 

183class IncompleteConfigException(BaseException): 

184 pass 

185 

186 

187def is_autogen_config() -> bool: 

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

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

190 

191 

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

193 base_cfg: DictConfig = OmegaConf.structured(Config) 

194 if conf_file_path.exists(): 

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

196 elif is_autogen_config(): 

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

198 try: 

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

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

201 except Exception as e: 

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

203 try: 

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

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

206 except Exception as e: 

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

208 cfg = base_cfg 

209 else: 

210 cfg = base_cfg 

211 

212 try: 

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

214 OmegaConf.to_container(cfg, throw_on_missing=True) 

215 OmegaConf.set_readonly(cfg, True) 

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

217 

218 if config.mqtt.user in ("", MISSING) or config.mqtt.password in ("", MISSING): 

219 log.info("The config has place holders for MQTT user and/or password") 

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

221 return None 

222 return config 

223 except (MissingMandatoryValue, ValidationError) as e: 

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

225 if return_invalid and cfg is not None: 

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

227 raise