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 

24from typing import Type 

25 

26import configargparse 

27 

28import slidge 

29from slidge import BaseGateway 

30from slidge.core import config 

31from slidge.db import SlidgeStore 

32from slidge.db.avatar import avatar_cache 

33from slidge.db.meta import get_engine 

34from slidge.migration import migrate 

35from slidge.util.conf import ConfigModule 

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=os.getenv( 

66 "SLIDGE_CONF_DIR", "/etc/slidge/conf.d/*.conf" 

67 ).split(":"), 

68 description=__doc__, 

69 ) 

70 p.add_argument( 

71 "-c", 

72 "--config", 

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

74 env_var="SLIDGE_CONFIG", 

75 is_config_file=True, 

76 ) 

77 p.add_argument( 

78 "--log-config", 

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

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

81 "for details.", 

82 ) 

83 p.add_argument( 

84 "-q", 

85 "--quiet", 

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

87 action="store_const", 

88 dest="loglevel", 

89 const=logging.WARNING, 

90 default=logging.INFO, 

91 env_var="SLIDGE_QUIET", 

92 ) 

93 p.add_argument( 

94 "-d", 

95 "--debug", 

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

97 action="store_const", 

98 dest="loglevel", 

99 const=logging.DEBUG, 

100 env_var="SLIDGE_DEBUG", 

101 ) 

102 p.add_argument( 

103 "--version", 

104 action="version", 

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

106 ) 

107 configurator = MainConfig( 

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

109 ) 

110 return configurator 

111 

112 

113def get_parser() -> configargparse.ArgumentParser: 

114 return get_configurator().parser 

115 

116 

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

118 configurator = get_configurator(from_entrypoint) 

119 args, unknown_argv = configurator.set_conf() 

120 

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

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

123 os.makedirs(h) 

124 

125 config.UPLOAD_REQUESTER = config.UPLOAD_REQUESTER or config.JID.bare 

126 

127 return unknown_argv 

128 

129 

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

131 logging.info("Caught SIGTERM") 

132 raise SigTermInterrupt 

133 

134 

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

136 from_entrypoint = module_name is not None 

137 signal.signal(signal.SIGTERM, handle_sigterm) 

138 

139 unknown_argv = configure(from_entrypoint) 

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

141 

142 if module_name is not None: 

143 config.LEGACY_MODULE = module_name 

144 

145 legacy_module = importlib.import_module(config.LEGACY_MODULE) 

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

147 logging.info( 

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

149 config.LEGACY_MODULE, 

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

151 ) 

152 

153 if plugin_config_obj := getattr( 

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

155 ): 

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

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

158 # now the dynamic defaults are set. 

159 if inspect.ismodule(plugin_config_obj): 

160 importlib.reload(plugin_config_obj) 

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

162 ConfigModule.ENV_VAR_PREFIX += ( 

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

164 ) 

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

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

167 

168 if unknown_argv: 

169 logging.error( 

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

171 ) 

172 

173 migrate() 

174 

175 gw_cls: Type[BaseGateway] = BaseGateway.get_unique_subclass() # type:ignore[assignment] 

176 store = SlidgeStore( 

177 get_engine( 

178 config.DB_URL, 

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

180 pool_size=gw_cls.DB_POOL_SIZE, 

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("disconnected", _on_disconnected) 

189 gateway.add_event_handler("stream_error", _on_stream_error) 

190 gateway.connect() 

191 return_code = 0 

192 try: 

193 gateway.loop.run_forever() 

194 except KeyboardInterrupt: 

195 logging.debug("Received SIGINT") 

196 except SigTermInterrupt: 

197 logging.debug("Received SIGTERM") 

198 except SystemExit as e: 

199 return_code = e.code # type: ignore 

200 logging.debug("Exit called") 

201 except Exception as e: 

202 return_code = 2 

203 logging.exception("Exception in __main__") 

204 logging.exception(e) 

205 finally: 

206 if gateway.has_crashed: 

207 if return_code != 0: 

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

209 return_code = 3 

210 if gateway.is_connected(): 

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

212 gateway.del_event_handler("disconnected", _on_disconnected) 

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

214 gateway.disconnect() 

215 gateway.loop.run_until_complete(gateway.disconnected) 

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

217 else: 

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

219 avatar_cache.close() 

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

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

222 exit(return_code) 

223 

224 

225def _on_disconnected(e): 

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

227 exit(10) 

228 

229 

230def _on_stream_error(e): 

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