Coverage for src/updates2mqtt/config.py: 92%
196 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-09 06:17 +0000
1import os
2import ssl
3import typing
4from dataclasses import dataclass, field
5from enum import EnumMeta, StrEnum, auto
6from pathlib import Path
8import structlog
9from omegaconf import MISSING, DictConfig, MissingMandatoryValue, OmegaConf, ValidationError
11log = structlog.get_logger()
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-]+)*))?$"
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}
31class UpdatePolicy(StrEnum):
32 AUTO = "Auto"
33 PASSIVE = "Passive"
36class PublishPolicy(StrEnum):
37 HOMEASSISTANT = "HomeAssistant"
38 MQTT = "MQTT"
39 SILENT = "Silent"
42class LogLevel(StrEnum):
43 DEBUG = "DEBUG"
44 INFO = "INFO"
45 WARNING = "WARNING"
46 ERROR = "ERROR"
47 CRITICAL = "CRITICAL"
50class RegistryAPI(StrEnum):
51 OCI_V2 = "OCI_V2"
52 OCI_V2_MINIMAL = "OCI_V2"
53 DOCKER_CLIENT = "DOCKER_CLIENT"
54 DISABLED = "DISABLED"
57class VersionType:
58 SHORT_SHA = "short_sha"
59 FULL_SHA = "full_sha"
60 VERSION_REVISION = "version_revision"
61 VERSION = "version"
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
72class CaseInsensitiveEnumMeta(EnumMeta):
73 def __getitem__(self, name):
74 return self._member_map_[name] if name in self._member_map_ else self._member_map_[name.upper()]
77class TlsMode(StrEnum, metaclass=CaseInsensitiveEnumMeta):
78 ON = auto()
79 OFF = auto()
80 INSECURE = auto()
83@dataclass
84class MqttConfig:
85 host: str = "${oc.env:MQTT_HOST,localhost}"
86 user: str = f"${{oc.env:MQTT_USER,{MISSING}}}"
87 password: str | None = "${oc.env:MQTT_PASS,''}"
88 port: int = "${oc.decode:${oc.env:MQTT_PORT,1883}}" # type: ignore[assignment]
89 topic_root: str = "updates2mqtt"
90 protocol: str = "${oc.env:MQTT_VERSION,3.11}"
91 transport: str = "tcp"
92 connect_timeout: float = 20
93 keepalive: int = 30
94 tls_mode: TlsMode = TlsMode.OFF
95 ca_certs: str | None = "${oc.env:MQTT_CA_CERTS,''}"
96 client_cert: str | None = "${oc.env:MQTT_CLIENT_CERT,''}" # certfile
97 client_key: str | None = "${oc.env:MQTT_CLIENT_KEY,''}" # keyfile
98 client_key_password: str | None = "${oc.env:MQTT_CLIENT_KEY_PASS,''}" # keyfile_password
99 cert_reqs: ssl.VerifyMode = ssl.CERT_REQUIRED
102@dataclass
103class GitHubConfig:
104 access_token: str | None = None
105 mutable_cache_ttl: int = 60 * 60 * 15
108@dataclass
109class MetadataSourceConfig:
110 enabled: bool = True
111 cache_ttl: int = 60 * 60 * 24 * 7 # 1 week
114@dataclass
115class Selector:
116 include: list[str] | None = None
117 exclude: list[str] | None = None
120class VersionPolicy(StrEnum):
121 AUTO = "AUTO"
122 VERSION = "VERSION"
123 DIGEST = "DIGEST"
124 VERSION_DIGEST = "VERSION_DIGEST"
125 TIMESTAMP = "TIMESTAMP"
128@dataclass
129class DockerPackageUpdateInfo:
130 image_name: typing.Any = MISSING # untagged image ref, either a single string or a list of strings
131 version_policy: VersionPolicy = VersionPolicy.AUTO
134def docker_image_names(docker_info: DockerPackageUpdateInfo) -> list[str]:
135 if isinstance(docker_info.image_name, str):
136 return [docker_info.image_name]
137 return list(docker_info.image_name)
140@dataclass
141class PackageUpdateInfo:
142 docker: DockerPackageUpdateInfo | None = field(default_factory=DockerPackageUpdateInfo)
143 logo_url: str | None = None
144 release_notes_url: str | None = None
145 source_repo_url: str | None = None
148@dataclass
149class DockerConfig:
150 enabled: bool = True
151 allow_pull: bool = True
152 allow_restart: bool = True
153 allow_build: bool = True
154 compose_version: str = "v2"
155 default_entity_picture_url: str = "https://www.docker.com/wp-content/uploads/2022/03/Moby-logo.png"
156 # Icon to show when browsing entities in Home Assistant
157 device_icon: str = "mdi:docker"
158 discover_metadata: dict[str, MetadataSourceConfig] = field(
159 default_factory=lambda: {"linuxserver.io": MetadataSourceConfig(enabled=True)}
160 )
161 registry: RegistryConfig = field(default_factory=lambda: RegistryConfig())
162 default_api_backoff: int = 60 * 15
163 image_ref_select: Selector = field(default_factory=lambda: Selector())
164 version_select: Selector = field(default_factory=lambda: Selector())
165 version_policy: VersionPolicy = VersionPolicy.AUTO
166 registry_select: Selector = field(default_factory=lambda: Selector())
169@dataclass
170class HomeAssistantDiscoveryConfig:
171 prefix: str = "homeassistant"
172 enabled: bool = True
175@dataclass
176class HomeAssistantConfig:
177 discovery: HomeAssistantDiscoveryConfig = field(default_factory=HomeAssistantDiscoveryConfig)
178 state_topic_suffix: str = "state"
179 device_creation: bool = True
180 force_command_topic: bool = False
181 extra_attributes: bool = True
182 area: str | None = None
183 release_summary_max_size: int = 6144
186@dataclass
187class HealthCheckConfig:
188 enabled: bool = True
189 interval: int = 300 # Interval in seconds to publish heartbeat message, 0 to disable
190 topic_template: str = "healthcheck/{node_name}/updates2mqtt"
193@dataclass
194class NodeConfig:
195 name: str = field(default_factory=lambda: os.getenv("NODE_NAME", os.uname().nodename.replace(".local", "")))
196 git_path: str = "/usr/bin/git"
197 healthcheck: HealthCheckConfig = field(default_factory=HealthCheckConfig)
200@dataclass
201class LogConfig:
202 level: LogLevel = "${oc.decode:${oc.env:U2M_LOG_LEVEL,INFO}}" # type: ignore[assignment] # pyright: ignore[reportAssignmentType]
203 json: bool = field(default=False, doc="Use JSON structured logging for non-interactive running")
206@dataclass
207class Config:
208 log: LogConfig = field(default_factory=LogConfig) # pyright: ignore[reportArgumentType, reportCallIssue]
209 node: NodeConfig = field(default_factory=NodeConfig)
210 mqtt: MqttConfig = field(default_factory=MqttConfig) # pyright: ignore[reportArgumentType, reportCallIssue]
211 homeassistant: HomeAssistantConfig = field(default_factory=HomeAssistantConfig)
212 docker: DockerConfig = field(default_factory=DockerConfig)
213 github: GitHubConfig = field(default_factory=GitHubConfig)
214 scan_interval: int = 60 * 60 * 3
215 packages: dict[str, PackageUpdateInfo] = field(default_factory=dict)
218@dataclass
219class CommonPackages:
220 common_packages: dict[str, PackageUpdateInfo] = field(default_factory=dict)
223class IncompleteConfigException(BaseException):
224 pass
227def is_autogen_config() -> bool:
228 env_var: str | None = os.environ.get("U2M_AUTOGEN_CONFIG")
229 return not (env_var and env_var.lower() in ("no", "0", "false"))
232def load_app_config(conf_file_path: Path, return_invalid: bool = False) -> Config | None:
233 base_cfg: DictConfig = OmegaConf.structured(Config)
234 if conf_file_path.exists():
235 cfg: DictConfig = typing.cast("DictConfig", OmegaConf.merge(base_cfg, OmegaConf.load(conf_file_path)))
236 elif is_autogen_config():
237 if not conf_file_path.parent.exists(): 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true
238 try:
239 log.debug(f"Creating config directory {conf_file_path.parent} if not already present")
240 conf_file_path.parent.mkdir(parents=True, exist_ok=True)
241 except Exception as e:
242 log.warning("Unable to create config directory: %s", e, path=conf_file_path.parent)
243 try:
244 conf_file_path.write_text(OmegaConf.to_yaml(base_cfg))
245 log.info(f"Auto-generated a new config file at {conf_file_path}")
246 except Exception as e:
247 log.warning("Unable to write config file: %s", e, path=conf_file_path)
248 cfg = base_cfg
249 else:
250 cfg = base_cfg
252 try:
253 # Validate that all required fields are present, throw exception now rather than when config first used
254 OmegaConf.to_container(cfg, throw_on_missing=True)
255 OmegaConf.set_readonly(cfg, True)
256 config: Config = typing.cast("Config", cfg)
258 if config.mqtt.user in ("", MISSING):
259 log.info("The config has place holders for MQTT user")
260 if not return_invalid: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 return None
262 return config
263 except (MissingMandatoryValue, ValidationError) as e:
264 log.error("Configuration error %s", e, path=conf_file_path.as_posix())
265 if return_invalid and cfg is not None:
266 return typing.cast("Config", cfg)
267 raise