Coverage for slidge / main.py: 37%

1""" 

2Slidge can be configured via CLI args, environment variables and/or INI files. 

3 

4To use env vars, use this convention: ``--home-dir`` becomes ``HOME_DIR``. 

5 

6Everything in ``/etc/slidge/conf.d/*`` is automatically used. 

7To use a plugin-specific INI file, put it in another dir, 

8and launch slidge with ``-c /path/to/plugin-specific.conf``. 

9Use the long version of the CLI arg without the double dash prefix inside this 

10INI file, eg ``debug=true``. 

11 

12An example configuration file is available at 

13https://codeberg.org/slidge/slidge/src/branch/main/dev/confs/slidge-example.ini 

14""" 

15 

16import asyncio 

17import importlib 

18import inspect 

19import logging 

20import logging.config 

21import os 

22import signal 

23from pathlib import Path 

24 

25import configargparse 

26 

27import slidge 

28from slidge.core import config 

29from slidge.core.gateway import BaseGateway 

30from slidge.db import SlidgeStore 

31from slidge.db.avatar import avatar_cache 

32from slidge.db.meta import get_engine 

33from slidge.migration import migrate 

34from slidge.util.conf import ConfigModule 

35from slidge.util.types import AnyGateway 

36 

37 

38class MainConfig(ConfigModule): 

39 def update_dynamic_defaults(self, args: configargparse.Namespace) -> None: 

40 # force=True is needed in case we call a logger before this is reached, 

41 # or basicConfig has no effect 

42 if args.log_config: 

43 logging.config.fileConfig(args.log_config) 

44 else: 

45 logging.basicConfig( 

46 level=args.loglevel, 

47 filename=args.log_file, 

48 force=True, 

49 format=args.log_format, 

50 ) 

51 

52 if args.home_dir is None: 

53 args.home_dir = Path("/var/lib/slidge") / str(args.jid) 

54 

55 if args.db_url is None: 

56 args.db_url = f"sqlite:///{args.home_dir}/slidge.sqlite" 

57 

58 

59class SigTermInterrupt(Exception): 

60 pass 

61 

62 

63def get_configurator(from_entrypoint: bool = False) -> MainConfig: 

64 p = configargparse.ArgumentParser( 

65 default_config_files=[ 

66 f"{p}/*" 

67 for p in os.getenv("SLIDGE_CONF_DIR", "/etc/slidge/conf.d/").split(":") 

68 ], 

69 description=__doc__, 

70 ) 

71 p.add_argument( 

72 "-c", 

73 "--config", 

74 help="Path to a INI config file.", 

75 env_var="SLIDGE_CONFIG", 

76 is_config_file=True, 

77 ) 

78 p.add_argument( 

79 "--log-config", 

80 help="Path to a INI config file to personalise logging output. Refer to " 

81 "<https://docs.python.org/3/library/logging.config.html#configuration-file-format> " 

82 "for details.", 

83 ) 

84 p.add_argument( 

85 "-q", 

86 "--quiet", 

87 help="loglevel=WARNING (unused if --log-config is specified)", 

88 action="store_const", 

89 dest="loglevel", 

90 const=logging.WARNING, 

91 default=logging.INFO, 

92 env_var="SLIDGE_QUIET", 

93 ) 

94 p.add_argument( 

95 "-d", 

96 "--debug", 

97 help="loglevel=DEBUG (unused if --log-config is specified)", 

98 action="store_const", 

99 dest="loglevel", 

100 const=logging.DEBUG, 

101 env_var="SLIDGE_DEBUG", 

102 ) 

103 p.add_argument( 

104 "--version", 

105 action="version", 

106 version=f"%(prog)s {slidge.__version__}", 

107 ) 

108 configurator = MainConfig( 

109 config, p, skip_options=("legacy_module",) if from_entrypoint else () 

110 ) 

111 return configurator 

112 

113 

114def get_parser() -> configargparse.ArgumentParser: 

115 return get_configurator().parser 

116 

117 

118def configure(from_entrypoint: bool) -> list[str]: 

119 configurator = get_configurator(from_entrypoint) 

120 _args, unknown_argv = configurator.set_conf() 

121 

122 if not (h := config.HOME_DIR).exists(): 

123 logging.info("Creating directory '%s'", h) 

124 os.makedirs(h) 

125 

126 config.UPLOAD_REQUESTER = config.UPLOAD_REQUESTER or config.JID.bare 

127 

128 return unknown_argv 

129 

130 

131def handle_sigterm(_signum: int, _frame: object) -> None: 

132 logging.info("Caught SIGTERM") 

133 raise SigTermInterrupt 

134 

135 

136def main(module_name: str | None = None) -> None: 

137 from_entrypoint = module_name is not None 

138 signal.signal(signal.SIGTERM, handle_sigterm) 

139 

140 unknown_argv = configure(from_entrypoint) 

141 logging.info("Starting slidge version %s", slidge.__version__) 

142 

143 if module_name is not None: 

144 config.LEGACY_MODULE = module_name 

145 

146 legacy_module = importlib.import_module(config.LEGACY_MODULE) 

147 logging.debug("Legacy module: %s", dir(legacy_module)) 

148 logging.info( 

149 "Starting legacy module: '%s' version %s", 

150 config.LEGACY_MODULE, 

151 getattr(legacy_module, "__version__", "No version"), 

152 ) 

153 

154 if plugin_config_obj := getattr( 

155 legacy_module, "config", getattr(legacy_module, "Config", None) 

156 ): 

157 # If the legacy module has default parameters that depend on dynamic defaults 

158 # of the slidge main config, it needs to be refreshed at this point, because 

159 # now the dynamic defaults are set. 

160 if inspect.ismodule(plugin_config_obj): 

161 importlib.reload(plugin_config_obj) 

162 logging.debug("Found a config object in plugin: %r", plugin_config_obj) 

163 ConfigModule.ENV_VAR_PREFIX += ( 

164 f"_{config.LEGACY_MODULE.split('.')[-1].upper()}_" 

165 ) 

166 logging.debug("Env var prefix: %s", ConfigModule.ENV_VAR_PREFIX) 

167 _, unknown_argv = ConfigModule(plugin_config_obj).set_conf(unknown_argv) 

168 

169 if unknown_argv: 

170 logging.error( 

171 f"These config options have not been recognized and ignored: {unknown_argv}" 

172 ) 

173 

174 migrate() 

175 

176 gw_cls: type[AnyGateway] = BaseGateway.get_unique_subclass() # type:ignore[assignment] 

177 store = SlidgeStore( 

178 get_engine( 

179 config.DB_URL, 

180 echo=logging.getLogger().isEnabledFor(level=logging.DEBUG), 

181 pool_size=gw_cls.DB_POOL_SIZE, 

182 ) 

183 ) 

184 BaseGateway.store = store 

185 gateway = gw_cls() 

186 avatar_cache.store = gateway.store.avatars 

187 avatar_cache.set_dir(config.HOME_DIR / "slidge_avatars_v3") 

188 

189 gateway.add_event_handler("connection_lost", _on_connection_lost) 

190 gateway.add_event_handler("disconnected", _on_disconnected) 

191 gateway.add_event_handler("stream_error", _on_stream_error) 

192 gateway.connect() 

193 return_code = 0 

194 try: 

195 gateway.loop.run_forever() 

196 except KeyboardInterrupt: 

197 logging.debug("Received SIGINT") 

198 except SigTermInterrupt: 

199 logging.debug("Received SIGTERM") 

200 except SystemExit as e: 

201 return_code = e.code # type: ignore 

202 logging.debug("Exit called") 

203 except Exception as e: 

204 return_code = 2 

205 logging.exception("Exception in __main__") 

206 logging.exception(e) 

207 finally: 

208 if gateway.has_crashed: 

209 if return_code != 0: 

210 logging.warning("Return code has been set twice. Please report this.") 

211 return_code = 3 

212 if gateway.is_connected(): 

213 logging.debug("Gateway is connected, cleaning up") 

214 gateway.del_event_handler("disconnected", _on_disconnected) 

215 gateway.loop.run_until_complete(asyncio.gather(*gateway.shutdown())) 

216 gateway.disconnect() 

217 gateway.loop.run_until_complete(gateway.disconnected) 

218 logging.info("Successful clean shut down") 

219 else: 

220 logging.debug("Gateway is not connected, no need to clean up") 

221 avatar_cache.close() 

222 gateway.loop.run_until_complete(gateway.http.close()) 

223 logging.debug("Exiting with code %s", return_code) 

224 exit(return_code) 

225 

226 

227def _on_disconnected(e: BaseException) -> None: 

228 logging.error("Disconnected from the XMPP server: '%s'.", e) 

229 exit(10) 

230 

231 

232def _on_stream_error(e: BaseException) -> None: 

233 logging.error("Stream error: '%s'.", e) 

234 

235 

236def _on_connection_lost(*args: object) -> None: 

237 logging.error("Connection lost: '%s'", args) 

238 exit(15)