Coverage for slidge/main.py: 38%

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 

23import sys 

24from pathlib import Path 

25 

26import configargparse 

27 

28import slidge 

29from slidge.core import config 

30from slidge.core.gateway import BaseGateway 

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 

36from slidge.util.types import AnyGateway 

37 

38 

39class MainConfig(ConfigModule): 

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

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

42 # or basicConfig has no effect 

43 if args.log_config: 

44 logging.config.fileConfig(args.log_config) 

45 else: 

46 logging.basicConfig( 

47 level=args.loglevel, 

48 filename=args.log_file, 

49 force=True, 

50 format=args.log_format, 

51 ) 

52 

53 if args.home_dir is None: 

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

55 

56 if args.db_url is None: 

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

58 

59 

60class SigTermInterrupt(Exception): 

61 pass 

62 

63 

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

65 p = configargparse.ArgumentParser( 

66 default_config_files=[ 

67 f"{p}/*" 

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

69 ], 

70 description=__doc__, 

71 ) 

72 p.add_argument( 

73 "-c", 

74 "--config", 

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

76 env_var="SLIDGE_CONFIG", 

77 is_config_file=True, 

78 ) 

79 p.add_argument( 

80 "--log-config", 

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

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

83 "for details.", 

84 ) 

85 p.add_argument( 

86 "-q", 

87 "--quiet", 

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

89 action="store_const", 

90 dest="loglevel", 

91 const=logging.WARNING, 

92 default=logging.INFO, 

93 env_var="SLIDGE_QUIET", 

94 ) 

95 p.add_argument( 

96 "-d", 

97 "--debug", 

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

99 action="store_const", 

100 dest="loglevel", 

101 const=logging.DEBUG, 

102 env_var="SLIDGE_DEBUG", 

103 ) 

104 p.add_argument( 

105 "--version", 

106 action="version", 

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

108 ) 

109 configurator = MainConfig( 

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

111 ) 

112 return configurator 

113 

114 

115def get_parser() -> configargparse.ArgumentParser: 

116 return get_configurator().parser 

117 

118 

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

120 configurator = get_configurator(from_entrypoint) 

121 _args, unknown_argv = configurator.set_conf() 

122 

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

124 logging.info("Creating directory '%s'", h) # noqa: LOG015 

125 os.makedirs(h) 

126 

127 config.UPLOAD_REQUESTER = config.UPLOAD_REQUESTER or config.JID.bare 

128 

129 return unknown_argv 

130 

131 

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

133 logging.info("Caught SIGTERM") # noqa: LOG015 

134 raise SigTermInterrupt 

135 

136 

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

138 from_entrypoint = module_name is not None 

139 signal.signal(signal.SIGTERM, handle_sigterm) 

140 

141 unknown_argv = configure(from_entrypoint) 

142 logging.info("Starting slidge version %s", slidge.__version__) # noqa: LOG015 

143 

144 if module_name is not None: 

145 config.LEGACY_MODULE = module_name 

146 

147 legacy_module = importlib.import_module(config.LEGACY_MODULE) 

148 logging.debug("Legacy module: %s", dir(legacy_module)) # noqa: LOG015 

149 logging.info( # noqa: LOG015 

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

151 config.LEGACY_MODULE, 

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

153 ) 

154 

155 if plugin_config_obj := getattr( 

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

157 ): 

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

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

160 # now the dynamic defaults are set. 

161 if inspect.ismodule(plugin_config_obj): 

162 importlib.reload(plugin_config_obj) 

163 logging.debug("Found a config object in plugin: %r", plugin_config_obj) # noqa: LOG015 

164 ConfigModule.ENV_VAR_PREFIX += ( 

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

166 ) 

167 logging.debug("Env var prefix: %s", ConfigModule.ENV_VAR_PREFIX) # noqa: LOG015 

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

169 

170 if unknown_argv: 

171 logging.error( # noqa: LOG015 

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

173 ) 

174 

175 migrate() 

176 

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

178 store = SlidgeStore( 

179 get_engine( 

180 config.DB_URL, 

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

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") # noqa: LOG015 

198 except SigTermInterrupt: 

199 logging.debug("Received SIGTERM") # noqa: LOG015 

200 except SystemExit as e: 

201 return_code = e.code # type: ignore 

202 logging.debug("Exit called") # noqa: LOG015 

203 except Exception: 

204 return_code = 2 

205 logging.exception("Exception in __main__") # noqa: LOG015 

206 finally: 

207 if gateway.has_crashed: 

208 if return_code != 0: 

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

210 return_code = 3 

211 if gateway.is_connected(): 

212 logging.debug("Gateway is connected, cleaning up") # noqa: LOG015 

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") # noqa: LOG015 

218 else: 

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

220 avatar_cache.close() 

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

222 logging.debug("Exiting with code %s", return_code) # noqa: LOG015 

223 sys.exit(return_code) 

224 

225 

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

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

228 sys.exit(10) 

229 

230 

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

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

233 

234 

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

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

237 sys.exit(15)