Coverage for slidge/main.py: 39%

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 re 

23import signal 

24from pathlib import Path 

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.user_jid_validator is None: 

56 args.user_jid_validator = ".*@" + re.escape(args.server) 

57 

58 if args.db_url is None: 

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

60 

61 

62class SigTermInterrupt(Exception): 

63 pass 

64 

65 

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

67 p = configargparse.ArgumentParser( 

68 default_config_files=os.getenv( 

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

70 ).split(":"), 

71 description=__doc__, 

72 ) 

73 p.add_argument( 

74 "-c", 

75 "--config", 

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

77 env_var="SLIDGE_CONFIG", 

78 is_config_file=True, 

79 ) 

80 p.add_argument( 

81 "--log-config", 

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

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

84 "for details.", 

85 ) 

86 p.add_argument( 

87 "-q", 

88 "--quiet", 

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

90 action="store_const", 

91 dest="loglevel", 

92 const=logging.WARNING, 

93 default=logging.INFO, 

94 env_var="SLIDGE_QUIET", 

95 ) 

96 p.add_argument( 

97 "-d", 

98 "--debug", 

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

100 action="store_const", 

101 dest="loglevel", 

102 const=logging.DEBUG, 

103 env_var="SLIDGE_DEBUG", 

104 ) 

105 p.add_argument( 

106 "--version", 

107 action="version", 

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

109 ) 

110 configurator = MainConfig( 

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

112 ) 

113 return configurator 

114 

115 

116def get_parser() -> configargparse.ArgumentParser: 

117 return get_configurator().parser 

118 

119 

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

121 configurator = get_configurator(from_entrypoint) 

122 args, unknown_argv = configurator.set_conf() 

123 

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

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

126 os.makedirs(h) 

127 

128 config.UPLOAD_REQUESTER = config.UPLOAD_REQUESTER or config.JID.bare 

129 

130 return unknown_argv 

131 

132 

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

134 logging.info("Caught SIGTERM") 

135 raise SigTermInterrupt 

136 

137 

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

139 from_entrypoint = module_name is not None 

140 signal.signal(signal.SIGTERM, handle_sigterm) 

141 

142 unknown_argv = configure(from_entrypoint) 

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

144 

145 if module_name is not None: 

146 config.LEGACY_MODULE = module_name 

147 

148 legacy_module = importlib.import_module(config.LEGACY_MODULE) 

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

150 logging.info( 

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

152 config.LEGACY_MODULE, 

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

154 ) 

155 

156 if plugin_config_obj := getattr( 

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

158 ): 

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

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

161 # now the dynamic defaults are set. 

162 if inspect.ismodule(plugin_config_obj): 

163 importlib.reload(plugin_config_obj) 

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

165 ConfigModule.ENV_VAR_PREFIX += ( 

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

167 ) 

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

169 ConfigModule(plugin_config_obj).set_conf(unknown_argv) 

170 else: 

171 if unknown_argv: 

172 raise RuntimeError("Some arguments have not been recognized", unknown_argv) 

173 

174 migrate() 

175 

176 store = SlidgeStore( 

177 get_engine( 

178 config.DB_URL, echo=logging.getLogger().isEnabledFor(level=logging.DEBUG) 

179 ) 

180 ) 

181 BaseGateway.store = store 

182 gateway: BaseGateway = BaseGateway.get_unique_subclass()() 

183 avatar_cache.store = gateway.store.avatars 

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

185 

186 gateway.connect() 

187 

188 return_code = 0 

189 try: 

190 gateway.loop.run_forever() 

191 except KeyboardInterrupt: 

192 logging.debug("Received SIGINT") 

193 except SigTermInterrupt: 

194 logging.debug("Received SIGTERM") 

195 except SystemExit as e: 

196 return_code = e.code # type: ignore 

197 logging.debug("Exit called") 

198 except Exception as e: 

199 return_code = 2 

200 logging.exception("Exception in __main__") 

201 logging.exception(e) 

202 finally: 

203 if gateway.has_crashed: 

204 if return_code != 0: 

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

206 return_code = 3 

207 if gateway.is_connected(): 

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

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

210 gateway.disconnect() 

211 gateway.loop.run_until_complete(gateway.disconnected) 

212 else: 

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

214 avatar_cache.close() 

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

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

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

218 exit(return_code)