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

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") 

13NO_KNOWN_IMAGE = "UNKNOWN" 

14 

15 

16class UpdatePolicy(StrEnum): 

17 AUTO = "Auto" 

18 PASSIVE = "Passive" 

19 

20 

21class PublishPolicy(StrEnum): 

22 HOMEASSISTANT = "HomeAssistant" 

23 MQTT = "MQTT" 

24 SILENT = "Silent" 

25 

26 

27class LogLevel(StrEnum): 

28 DEBUG = "DEBUG" 

29 INFO = "INFO" 

30 WARNING = "WARNING" 

31 ERROR = "ERROR" 

32 CRITICAL = "CRITICAL" 

33 

34 

35class VersionType: 

36 SHORT_SHA = "short_sha" 

37 FULL_SHA = "full_sha" 

38 VERSION_REVISION = "version_revision" 

39 VERSION = "version" 

40 

41 

42@dataclass 

43class MqttConfig: 

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

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

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

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

48 topic_root: str = "updates2mqtt" 

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

50 

51 

52@dataclass 

53class MetadataSourceConfig: 

54 enabled: bool = True 

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

56 

57 

58@dataclass 

59class Selector: 

60 include: list[str] | None = None 

61 exclude: list[str] | None = None 

62 

63 

64@dataclass 

65class DockerConfig: 

66 enabled: bool = True 

67 allow_pull: bool = True 

68 allow_restart: bool = True 

69 allow_build: bool = True 

70 compose_version: str = "v2" 

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

72 # Icon to show when browsing entities in Home Assistant 

73 device_icon: str = "mdi:docker" 

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

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

76 ) 

77 default_api_backoff: int = 60 * 15 

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

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

80 

81 

82@dataclass 

83class HomeAssistantDiscoveryConfig: 

84 prefix: str = "homeassistant" 

85 enabled: bool = True 

86 

87 

88@dataclass 

89class HomeAssistantConfig: 

90 discovery: HomeAssistantDiscoveryConfig = field(default_factory=HomeAssistantDiscoveryConfig) 

91 state_topic_suffix: str = "state" 

92 device_creation: bool = True 

93 force_command_topic: bool = False 

94 extra_attributes: bool = True 

95 area: str | None = None 

96 

97 

98@dataclass 

99class HealthCheckConfig: 

100 enabled: bool = True 

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

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

103 

104 

105@dataclass 

106class NodeConfig: 

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

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

109 healthcheck: HealthCheckConfig = field(default_factory=HealthCheckConfig) 

110 

111 

112@dataclass 

113class LogConfig: 

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

115 

116 

117@dataclass 

118class Config: 

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

120 node: NodeConfig = field(default_factory=NodeConfig) 

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

122 homeassistant: HomeAssistantConfig = field(default_factory=HomeAssistantConfig) 

123 docker: DockerConfig = field(default_factory=DockerConfig) 

124 scan_interval: int = 60 * 60 * 3 

125 

126 

127@dataclass 

128class DockerPackageUpdateInfo: 

129 image_name: str = MISSING 

130 

131 

132@dataclass 

133class PackageUpdateInfo: 

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

135 logo_url: str | None = None 

136 release_notes_url: str | None = None 

137 source_repo_url: str | None = None 

138 

139 

140@dataclass 

141class UpdateInfoConfig: 

142 common_packages: dict[str, PackageUpdateInfo] = field(default_factory=lambda: {}) 

143 

144 

145class IncompleteConfigException(BaseException): 

146 pass 

147 

148 

149def is_autogen_config() -> bool: 

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

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

152 

153 

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

155 base_cfg: DictConfig = OmegaConf.structured(Config) 

156 if conf_file_path.exists(): 

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

158 elif is_autogen_config(): 

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

160 try: 

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

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

163 except Exception: 

164 log.warning("Unable to create config directory", path=conf_file_path.parent) 

165 try: 

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

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

168 except Exception: 

169 log.warning("Unable to write config file", path=conf_file_path) 

170 cfg = base_cfg 

171 else: 

172 cfg = base_cfg 

173 

174 try: 

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

176 OmegaConf.to_container(cfg, throw_on_missing=True) 

177 OmegaConf.set_readonly(cfg, True) 

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

179 

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

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

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

183 return None 

184 return config 

185 except (MissingMandatoryValue, ValidationError) as e: 

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

187 if return_invalid and cfg is not None: 

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

189 raise