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 ) 

182 ) 

183 BaseGateway.store = store 

184 gateway = gw_cls() 

185 avatar_cache.store = gateway.store.avatars 

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

187 

188 gateway.add_event_handler("connection_lost", _on_connection_lost) 

189 gateway.add_event_handler("disconnected", _on_disconnected) 

190 gateway.add_event_handler("stream_error", _on_stream_error) 

191 gateway.connect() 

192 return_code = 0 

193 try: 

194 gateway.loop.run_forever() 

195 except KeyboardInterrupt: 

196 logging.debug("Received SIGINT") 

197 except SigTermInterrupt: 

198 logging.debug("Received SIGTERM") 

199 except SystemExit as e: 

200 return_code = e.code # type: ignore 

201 logging.debug("Exit called") 

202 except Exception as e: 

203 return_code = 2 

204 logging.exception("Exception in __main__") 

205 logging.exception(e) 

206 finally: 

207 if gateway.has_crashed: 

208 if return_code != 0: 

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

210 return_code = 3 

211 if gateway.is_connected(): 

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

213 gateway.del_event_handler("disconnected", _on_disconnected) 

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

215 gateway.disconnect() 

216 gateway.loop.run_until_complete(gateway.disconnected) 

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

218 else: 

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

220 avatar_cache.close() 

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

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

223 exit(return_code) 

224 

225 

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

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

228 exit(10) 

229 

230 

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

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

233 

234 

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

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

237 exit(15)