Coverage for src / updates2mqtt / config.py: 91%
173 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-20 23:13 +0000
1import os
2import typing
3from dataclasses import dataclass, field
4from enum import StrEnum
5from pathlib import Path
7import structlog
8from omegaconf import MISSING, DictConfig, MissingMandatoryValue, OmegaConf, ValidationError
10log = structlog.get_logger()
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
18SOURCE_PLATFORM_GITHUB = "GitHub"
19SOURCE_PLATFORM_CODEBERG = "CodeBerg"
20SOURCE_PLATFORM_GITLAB = "GitLab"
21SOURCE_PLATFORMS = {
22 SOURCE_PLATFORM_GITHUB: r"https://github.com/.*",
23 SOURCE_PLATFORM_GITLAB: r"https://gitlab.com/.*",
24 SOURCE_PLATFORM_CODEBERG: r"https://codeberg.org/.*",
25}
28class UpdatePolicy(StrEnum):
29 AUTO = "Auto"
30 PASSIVE = "Passive"
33class PublishPolicy(StrEnum):
34 HOMEASSISTANT = "HomeAssistant"
35 MQTT = "MQTT"
36 SILENT = "Silent"
39class LogLevel(StrEnum):
40 DEBUG = "DEBUG"
41 INFO = "INFO"
42 WARNING = "WARNING"
43 ERROR = "ERROR"
44 CRITICAL = "CRITICAL"
47class RegistryAPI(StrEnum):
48 OCI_V2 = "OCI_V2"
49 OCI_V2_MINIMAL = "OCI_V2"
50 DOCKER_CLIENT = "DOCKER_CLIENT"
51 DISABLED = "DISABLED"
54class VersionType:
55 SHORT_SHA = "short_sha"
56 FULL_SHA = "full_sha"
57 VERSION_REVISION = "version_revision"
58 VERSION = "version"
61@dataclass
62class RegistryConfig:
63 api: RegistryAPI = RegistryAPI.OCI_V2
64 mutable_cache_ttl: int | None = None # default to server cache hint
65 immutable_cache_ttl: int | None = 7776000 # 90 days
66 token_cache_ttl: int | None = None # default to server cache hint
69@dataclass
70class MqttConfig:
71 host: str = "${oc.env:MQTT_HOST,localhost}"
72 user: str = f"${{oc.env:MQTT_USER,{MISSING}}}"
73 password: str = f"${{oc.env:MQTT_PASS,{MISSING}}}"
74 port: int = "${oc.decode:${oc.env:MQTT_PORT,1883}}" # type: ignore[assignment]
75 topic_root: str = "updates2mqtt"
76 protocol: str = "${oc.env:MQTT_VERSION,3.11}"
79@dataclass
80class GitHubConfig:
81 access_token: str | None = None
82 mutable_cache_ttl: int = 60 * 60 * 15
85@dataclass
86class MetadataSourceConfig:
87 enabled: bool = True
88 cache_ttl: int = 60 * 60 * 24 * 7 # 1 week
91@dataclass
92class Selector:
93 include: list[str] | None = None
94 exclude: list[str] | None = None
97class VersionPolicy(StrEnum):
98 AUTO = "AUTO"
99 VERSION = "VERSION"
100 DIGEST = "DIGEST"
101 VERSION_DIGEST = "VERSION_DIGEST"
102 TIMESTAMP = "TIMESTAMP"
105@dataclass
106class DockerPackageUpdateInfo:
107 image_name: str = MISSING # untagged image ref
108 version_policy: VersionPolicy = VersionPolicy.AUTO
111@dataclass
112class PackageUpdateInfo:
113 docker: DockerPackageUpdateInfo | None = field(default_factory=DockerPackageUpdateInfo)
114 logo_url: str | None = None
115 release_notes_url: str | None = None
116 source_repo_url: str | None = None
119@dataclass
120class DockerConfig:
121 enabled: bool = True
122 allow_pull: bool = True
123 allow_restart: bool = True
124 allow_build: bool = True
125 compose_version: str = "v2"
126 default_entity_picture_url: str = "https://www.docker.com/wp-content/uploads/2022/03/Moby-logo.png"
127 # Icon to show when browsing entities in Home Assistant
128 device_icon: str = "mdi:docker"
129 discover_metadata: dict[str, MetadataSourceConfig] = field(
130 default_factory=lambda: {"linuxserver.io": MetadataSourceConfig(enabled=True)}
131 )
132 registry: RegistryConfig = field(default_factory=lambda: RegistryConfig())
133 default_api_backoff: int = 60 * 15
134 image_ref_select: Selector = field(default_factory=lambda: Selector())
135 version_select: Selector = field(default_factory=lambda: Selector())
136 version_policy: VersionPolicy = VersionPolicy.AUTO
137 registry_select: Selector = field(default_factory=lambda: Selector())
140@dataclass
141class HomeAssistantDiscoveryConfig:
142 prefix: str = "homeassistant"
143 enabled: bool = True
146@dataclass
147class HomeAssistantConfig:
148 discovery: HomeAssistantDiscoveryConfig = field(default_factory=HomeAssistantDiscoveryConfig)
149 state_topic_suffix: str = "state"
150 device_creation: bool = True
151 force_command_topic: bool = False
152 extra_attributes: bool = True
153 area: str | None = None
154 release_summary_max_size: int = 6144
157@dataclass
158class HealthCheckConfig:
159 enabled: bool = True
160 interval: int = 300 # Interval in seconds to publish healthcheck message, 0 to disable
161 topic_template: str = "healthcheck/{node_name}/updates2mqtt"
164@dataclass
165class NodeConfig:
166 name: str = field(default_factory=lambda: os.uname().nodename.replace(".local", ""))
167 git_path: str = "/usr/bin/git"
168 healthcheck: HealthCheckConfig = field(default_factory=HealthCheckConfig)
171@dataclass
172class LogConfig:
173 level: LogLevel = "${oc.decode:${oc.env:U2M_LOG_LEVEL,INFO}}" # type: ignore[assignment] # pyright: ignore[reportAssignmentType]
176@dataclass
177class Config:
178 log: LogConfig = field(default_factory=LogConfig) # pyright: ignore[reportArgumentType, reportCallIssue]
179 node: NodeConfig = field(default_factory=NodeConfig)
180 mqtt: MqttConfig = field(default_factory=MqttConfig) # pyright: ignore[reportArgumentType, reportCallIssue]
181 homeassistant: HomeAssistantConfig = field(default_factory=HomeAssistantConfig)
182 docker: DockerConfig = field(default_factory=DockerConfig)
183 github: GitHubConfig = field(default_factory=GitHubConfig)
184 scan_interval: int = 60 * 60 * 3
185 packages: dict[str, PackageUpdateInfo] = field(default_factory=dict)
188@dataclass
189class CommonPackages:
190 common_packages: dict[str, PackageUpdateInfo] = field(default_factory=dict)
193class IncompleteConfigException(BaseException):
194 pass
197def is_autogen_config() -> bool:
198 env_var: str | None = os.environ.get("U2M_AUTOGEN_CONFIG")
199 return not (env_var and env_var.lower() in ("no", "0", "false"))
202def load_app_config(conf_file_path: Path, return_invalid: bool = False) -> Config | None:
203 base_cfg: DictConfig = OmegaConf.structured(Config)
204 if conf_file_path.exists():
205 cfg: DictConfig = typing.cast("DictConfig", OmegaConf.merge(base_cfg, OmegaConf.load(conf_file_path)))
206 elif is_autogen_config():
207 if not conf_file_path.parent.exists(): 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true
208 try:
209 log.debug(f"Creating config directory {conf_file_path.parent} if not already present")
210 conf_file_path.parent.mkdir(parents=True, exist_ok=True)
211 except Exception as e:
212 log.warning("Unable to create config directory: %s", e, path=conf_file_path.parent)
213 try:
214 conf_file_path.write_text(OmegaConf.to_yaml(base_cfg))
215 log.info(f"Auto-generated a new config file at {conf_file_path}")
216 except Exception as e:
217 log.warning("Unable to write config file: %s", e, path=conf_file_path)
218 cfg = base_cfg
219 else:
220 cfg = base_cfg
222 try:
223 # Validate that all required fields are present, throw exception now rather than when config first used
224 OmegaConf.to_container(cfg, throw_on_missing=True)
225 OmegaConf.set_readonly(cfg, True)
226 config: Config = typing.cast("Config", cfg)
228 if config.mqtt.user in ("", MISSING) or config.mqtt.password in ("", MISSING):
229 log.info("The config has place holders for MQTT user and/or password")
230 if not return_invalid: 230 ↛ 231line 230 didn't jump to line 231 because the condition on line 230 was never true
231 return None
232 return config
233 except (MissingMandatoryValue, ValidationError) as e:
234 log.error("Configuration error %s", e, path=conf_file_path.as_posix())
235 if return_invalid and cfg is not None:
236 return typing.cast("Config", cfg)
237 raise