Coverage for src/updates2mqtt/config.py: 91%
180 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-14 15:07 +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}"
77 connect_timeout: float = 20
78 keepalive: int = 30
81@dataclass
82class GitHubConfig:
83 access_token: str | None = None
84 mutable_cache_ttl: int = 60 * 60 * 15
87@dataclass
88class MetadataSourceConfig:
89 enabled: bool = True
90 cache_ttl: int = 60 * 60 * 24 * 7 # 1 week
93@dataclass
94class Selector:
95 include: list[str] | None = None
96 exclude: list[str] | None = None
99class VersionPolicy(StrEnum):
100 AUTO = "AUTO"
101 VERSION = "VERSION"
102 DIGEST = "DIGEST"
103 VERSION_DIGEST = "VERSION_DIGEST"
104 TIMESTAMP = "TIMESTAMP"
107@dataclass
108class DockerPackageUpdateInfo:
109 image_name: typing.Any = MISSING # untagged image ref, either a single string or a list of strings
110 version_policy: VersionPolicy = VersionPolicy.AUTO
113def docker_image_names(docker_info: DockerPackageUpdateInfo) -> list[str]:
114 if isinstance(docker_info.image_name, str):
115 return [docker_info.image_name]
116 return list(docker_info.image_name)
119@dataclass
120class PackageUpdateInfo:
121 docker: DockerPackageUpdateInfo | None = field(default_factory=DockerPackageUpdateInfo)
122 logo_url: str | None = None
123 release_notes_url: str | None = None
124 source_repo_url: str | None = None
127@dataclass
128class DockerConfig:
129 enabled: bool = True
130 allow_pull: bool = True
131 allow_restart: bool = True
132 allow_build: bool = True
133 compose_version: str = "v2"
134 default_entity_picture_url: str = "https://www.docker.com/wp-content/uploads/2022/03/Moby-logo.png"
135 # Icon to show when browsing entities in Home Assistant
136 device_icon: str = "mdi:docker"
137 discover_metadata: dict[str, MetadataSourceConfig] = field(
138 default_factory=lambda: {"linuxserver.io": MetadataSourceConfig(enabled=True)}
139 )
140 registry: RegistryConfig = field(default_factory=lambda: RegistryConfig())
141 default_api_backoff: int = 60 * 15
142 image_ref_select: Selector = field(default_factory=lambda: Selector())
143 version_select: Selector = field(default_factory=lambda: Selector())
144 version_policy: VersionPolicy = VersionPolicy.AUTO
145 registry_select: Selector = field(default_factory=lambda: Selector())
148@dataclass
149class HomeAssistantDiscoveryConfig:
150 prefix: str = "homeassistant"
151 enabled: bool = True
154@dataclass
155class HomeAssistantConfig:
156 discovery: HomeAssistantDiscoveryConfig = field(default_factory=HomeAssistantDiscoveryConfig)
157 state_topic_suffix: str = "state"
158 device_creation: bool = True
159 force_command_topic: bool = False
160 extra_attributes: bool = True
161 area: str | None = None
162 release_summary_max_size: int = 6144
165@dataclass
166class HealthCheckConfig:
167 enabled: bool = True
168 interval: int = 300 # Interval in seconds to publish heartbeat message, 0 to disable
169 topic_template: str = "healthcheck/{node_name}/updates2mqtt"
172@dataclass
173class NodeConfig:
174 name: str = field(default_factory=lambda: os.uname().nodename.replace(".local", ""))
175 git_path: str = "/usr/bin/git"
176 healthcheck: HealthCheckConfig = field(default_factory=HealthCheckConfig)
179@dataclass
180class LogConfig:
181 level: LogLevel = "${oc.decode:${oc.env:U2M_LOG_LEVEL,INFO}}" # type: ignore[assignment] # pyright: ignore[reportAssignmentType]
182 json: bool = field(default=False, doc="Use JSON structured logging for non-interactive running")
185@dataclass
186class Config:
187 log: LogConfig = field(default_factory=LogConfig) # pyright: ignore[reportArgumentType, reportCallIssue]
188 node: NodeConfig = field(default_factory=NodeConfig)
189 mqtt: MqttConfig = field(default_factory=MqttConfig) # pyright: ignore[reportArgumentType, reportCallIssue]
190 homeassistant: HomeAssistantConfig = field(default_factory=HomeAssistantConfig)
191 docker: DockerConfig = field(default_factory=DockerConfig)
192 github: GitHubConfig = field(default_factory=GitHubConfig)
193 scan_interval: int = 60 * 60 * 3
194 packages: dict[str, PackageUpdateInfo] = field(default_factory=dict)
197@dataclass
198class CommonPackages:
199 common_packages: dict[str, PackageUpdateInfo] = field(default_factory=dict)
202class IncompleteConfigException(BaseException):
203 pass
206def is_autogen_config() -> bool:
207 env_var: str | None = os.environ.get("U2M_AUTOGEN_CONFIG")
208 return not (env_var and env_var.lower() in ("no", "0", "false"))
211def load_app_config(conf_file_path: Path, return_invalid: bool = False) -> Config | None:
212 base_cfg: DictConfig = OmegaConf.structured(Config)
213 if conf_file_path.exists():
214 cfg: DictConfig = typing.cast("DictConfig", OmegaConf.merge(base_cfg, OmegaConf.load(conf_file_path)))
215 elif is_autogen_config():
216 if not conf_file_path.parent.exists(): 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 try:
218 log.debug(f"Creating config directory {conf_file_path.parent} if not already present")
219 conf_file_path.parent.mkdir(parents=True, exist_ok=True)
220 except Exception as e:
221 log.warning("Unable to create config directory: %s", e, path=conf_file_path.parent)
222 try:
223 conf_file_path.write_text(OmegaConf.to_yaml(base_cfg))
224 log.info(f"Auto-generated a new config file at {conf_file_path}")
225 except Exception as e:
226 log.warning("Unable to write config file: %s", e, path=conf_file_path)
227 cfg = base_cfg
228 else:
229 cfg = base_cfg
231 try:
232 # Validate that all required fields are present, throw exception now rather than when config first used
233 OmegaConf.to_container(cfg, throw_on_missing=True)
234 OmegaConf.set_readonly(cfg, True)
235 config: Config = typing.cast("Config", cfg)
237 if config.mqtt.user in ("", MISSING) or config.mqtt.password in ("", MISSING):
238 log.info("The config has place holders for MQTT user and/or password")
239 if not return_invalid: 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true
240 return None
241 return config
242 except (MissingMandatoryValue, ValidationError) as e:
243 log.error("Configuration error %s", e, path=conf_file_path.as_posix())
244 if return_invalid and cfg is not None:
245 return typing.cast("Config", cfg)
246 raise