Coverage for slidge/util/test.py: 84%
233 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-10 12:26 +0000
1# type:ignore
2import os
3import sys
4import tempfile
5import types
6import typing
7from pathlib import Path
8from types import ModuleType, NoneType
9from typing import Any, ClassVar, ForwardRef, NotRequired, get_args, get_origin
11from slixmpp import JID, Iq, Message, Presence
12from slixmpp.stanza.error import Error
13from slixmpp.test import SlixTest, TestTransport
14from sqlalchemy import create_engine, delete
16from slidge import (
17 BaseGateway,
18 BaseSession,
19 LegacyContact,
20 LegacyParticipant,
21 LegacyRoom,
22)
23from slidge.util.types import UserPreferences
25from ..command import FormField
26from ..core import config
27from ..core.session import _sessions
28from ..db import SlidgeStore
29from ..db.avatar import avatar_cache
30from ..db.meta import Base
31from ..db.models import Contact, GatewayUser
34class SlixTestPlus(SlixTest):
35 def setUp(self) -> None:
36 super().setUp()
37 Error.namespace = "jabber:component:accept"
39 def next_sent(self, timeout: float = 0.05) -> Message | Iq | Presence | None:
40 self.wait_for_send_queue()
41 sent = self.xmpp.socket.next_sent(timeout=timeout)
42 if sent is None:
43 return None
44 xml = self.parse_xml(sent)
45 self.fix_namespaces(xml, "jabber:component:accept")
46 sent = self.xmpp._build_stanza(xml, "jabber:component:accept")
47 return sent
50class SlidgeTest(SlixTestPlus):
51 plugin: types.ModuleType | dict
53 class Config:
54 jid = "aim.shakespeare.lit"
55 secret = "test"
56 server = "shakespeare.lit"
57 port = 5222
58 upload_service = "upload.test"
59 home_dir = Path(tempfile.mkdtemp())
60 user_jid_validator = ".*"
61 admins: ClassVar[list[str]] = []
62 upload_requester = None
63 ignore_delay_threshold = 300
65 gateway_cls: type[BaseGateway]
67 @classmethod
68 def setUpClass(cls) -> None:
69 for k, v in vars(cls.Config).items():
70 setattr(config, k.upper(), v)
71 if not hasattr(cls, "plugin"):
72 raise RuntimeError(
73 f"{cls.__name__} must set the 'plugin' attribute to the"
74 " module (or dict) containing the gateway class, e.g."
75 " 'plugin = globals()' or 'plugin = my_legacy_module'."
76 )
77 cls.gateway_cls = find_subclass(cls.plugin, BaseGateway)
78 session_cls = cls.gateway_cls._session_cls
79 wired = [
80 cls.gateway_cls,
81 session_cls,
82 session_cls._roster_cls,
83 session_cls._roster_cls._contact_cls,
84 session_cls._bookmarks_cls,
85 session_cls._bookmarks_cls._muc_cls,
86 session_cls._bookmarks_cls._muc_cls._participant_cls,
87 session_cls._space_nodes_cls,
88 session_cls._space_nodes_cls._space_cls,
89 ]
90 for wired_cls in wired:
91 wired_cls.__abstractmethods__ = frozenset()
93 def setUp(self) -> None:
94 # workaround for duplicate output of sql alchemy's log, cf
95 # https://stackoverflow.com/a/76498428/5902284
96 from sqlalchemy import log as sqlalchemy_log
97 from sqlalchemy.pool import StaticPool
99 sqlalchemy_log._add_default_handler = lambda x: None
100 db_url = os.getenv("SLIDGETEST_DB_URL", "sqlite+pysqlite:///:memory:")
101 engine = self.db_engine = create_engine(
102 db_url, poolclass=StaticPool if db_url.startswith("postgresql+") else None
103 )
104 Base.metadata.create_all(engine)
105 BaseGateway.store = SlidgeStore(engine)
106 BaseGateway._test_mode = True
108 # Clean up commands potentially left behind by a previous test.
109 LegacyContact._commands.clear()
110 LegacyContact._commands_chat.clear()
111 LegacyRoom._commands.clear()
112 LegacyRoom._commands_chat.clear()
114 self.xmpp = self.gateway_cls()
116 avatar_cache.store = self.xmpp.store.avatars
117 avatar_cache.set_dir(Path(tempfile.mkdtemp()))
118 self.xmpp._always_send_everything = True
119 engine.echo = True
121 self.xmpp.connection_made(TestTransport(self.xmpp))
122 self.xmpp.session_bind_event.set()
123 # Remove unique ID prefix to make it easier to test
124 self.xmpp._id_prefix = ""
125 self.xmpp.default_lang = None
126 self.xmpp.peer_default_lang = None
128 def new_id() -> str:
129 self.xmpp._id += 1
130 return str(self.xmpp._id)
132 self.xmpp._id = 0
133 self.xmpp.new_id = new_id
135 # Must have the stream header ready for xmpp.process() to work.
136 header = self.xmpp.stream_header
138 self.xmpp.data_received(header)
139 self.wait_for_send_queue()
141 self.xmpp.socket.next_sent()
142 self.xmpp.socket.next_sent()
144 # Some plugins require messages to have ID values. Set
145 # this to True in tests related to those plugins.
146 self.xmpp.use_message_ids = False
147 self.xmpp.use_presence_ids = False
148 Error.namespace = "jabber:component:accept"
150 def _add_slidge_user(
151 self,
152 jid: str | JID = "romeo@shakespeare.lit",
153 legacy_module_data: dict | None = None,
154 preferences: dict | None = None,
155 ) -> None:
156 with self.xmpp.store.session() as orm:
157 user = GatewayUser(
158 jid=JID(jid),
159 legacy_module_data=legacy_module_data or {},
160 preferences=preferences
161 or {"sync_avatar": False, "sync_presence": False},
162 )
163 orm.add(user)
164 orm.commit()
165 self.run_coro(
166 self.xmpp._BaseGateway__dispatcher._on_user_register(Iq(sfrom=JID(jid)))
167 )
168 welcome = self.next_sent()
169 assert welcome["body"]
170 stanza = self.next_sent()
171 assert "logging in" in stanza["status"].lower(), stanza
172 stanza = self.next_sent()
173 assert "syncing contacts" in stanza["status"].lower(), stanza
174 if self.xmpp.ROOMS:
175 stanza = self.next_sent()
176 assert "syncing rooms" in stanza["status"].lower(), stanza
177 probe = self.next_sent()
178 assert probe.get_type() == "probe"
179 stanza = self.next_sent()
180 assert stanza["status"].lower()
182 def user_session(self, jid: str | JID = "romeo@shakespeare.lit") -> BaseSession:
183 return self.xmpp._session_cls.from_jid(JID(jid))
185 def get_joined_muc(
186 self,
187 legacy_id: str | int,
188 user_jid: str | JID = "romeo@shakespeare.lit",
189 resource: str = "gajim",
190 ) -> LegacyRoom:
191 muc: LegacyRoom = self.run_coro(
192 self.user_session(user_jid).rooms.get_or_create(legacy_id)
193 )
194 muc.add_user_resource(resource)
195 muc._update_stored_attribute(participants_filled=True)
196 return muc
198 def tearDown(self) -> None:
199 self.db_engine.echo = False
200 self.xmpp.plugin["xep_0356"].granted_privileges.clear()
201 self.xmpp.plugin["xep_0356"].granted_privileges.default_factory.iq.clear()
202 super().tearDown()
203 Base.metadata.drop_all(self.xmpp.store._engine)
204 self.db_engine.dispose()
205 _sessions.clear()
207 def setup_logged_session(self, n_contacts: int = 0) -> None:
208 with self.xmpp.store.session() as orm:
209 user = GatewayUser(
210 jid=JID("romeo@montague.lit/gajim").bare,
211 legacy_module_data={"username": "romeo", "city": ""},
212 preferences={"sync_avatar": True, "sync_presence": True},
213 )
214 orm.add(user)
215 orm.commit()
217 with self.xmpp.store.session() as session:
218 session.execute(delete(Contact))
219 session.commit()
221 self.run_coro(
222 self.xmpp._BaseGateway__dispatcher._on_user_register(
223 Iq(sfrom="romeo@montague.lit/gajim")
224 )
225 )
226 welcome = self.next_sent()
227 assert welcome["body"], welcome
228 stanza = self.next_sent()
229 assert "logging in" in stanza["status"].lower(), stanza
230 stanza = self.next_sent()
231 assert "syncing contacts" in stanza["status"].lower(), stanza
232 if self.xmpp.ROOMS:
233 stanza = self.next_sent()
234 assert "syncing rooms" in stanza["status"].lower(), stanza
235 probe = self.next_sent()
236 assert probe.get_type() == "probe"
237 stanza = self.next_sent()
238 assert "yup" in stanza["status"].lower(), stanza
239 self.romeo: BaseSession = self.xmpp._session_cls.from_jid(
240 JID("romeo@montague.lit")
241 )
243 self.juliet: LegacyContact = self.run_coro(
244 self.romeo.contacts.get_or_create("juliet")
245 )
246 self.room: LegacyRoom = self.run_coro(self.romeo.rooms.get_or_create("room"))
247 self.first_witch: LegacyParticipant = self.room.participants.get_or_create(
248 nick="firstwitch"
249 )
251 self.send( # language=XML
252 """
253 <iq type="get"
254 to="romeo@montague.lit"
255 id="1"
256 from="aim.shakespeare.lit">
257 <pubsub xmlns="http://jabber.org/protocol/pubsub">
258 <items node="urn:xmpp:avatar:metadata" />
259 </pubsub>
260 </iq>
261 """
262 )
265def find_subclass(o, parent, base_ok: bool = False): # noqa
266 try:
267 vals = vars(o).values()
268 except TypeError:
269 vals = o.values()
270 for x in vals:
271 try:
272 if issubclass(x, parent) and x is not parent:
273 return x
274 except TypeError:
275 pass
276 if base_ok:
277 return parent
278 else:
279 raise RuntimeError(f"Could not find a subclass of {parent} in {o}")
282def typed_dict_consistency_tester(module: ModuleType) -> None:
283 """
284 Ensure consistency between the Preferences TypedDict and the fields defined
285 in Gateway.PREFERENCES.
286 """
287 session_cls = _get_type_arg(module.Gateway, "SessionT")
288 prefs = _get_type_arg(session_cls, "PrefsT")
289 if prefs is Any:
290 prefs = UserPreferences
291 _test_typed_dict_consistency(prefs, module.Gateway.PREFERENCES)
292 registration_dict = _get_type_arg(module.Gateway, "RegistrationFormT")
293 _test_typed_dict_consistency(registration_dict, module.Gateway.REGISTRATION_FIELDS)
296def _test_typed_dict_consistency(obj: object, form_fields: list[FormField]) -> None:
297 annotations = obj.__annotations__
298 total = getattr(obj, "__total__", True)
299 fields_dict = {}
300 for field in form_fields:
301 if field.type == "fixed":
302 continue
303 assert field.var in annotations, (
304 f"{field} is missing in the TypedDict: {annotations}"
305 )
306 fields_dict[field.var] = field
307 name = f"{obj.__name__}.{field.var}"
308 if isinstance(annotations[field.var], ForwardRef):
309 type_ = _evaluate_forward_ref(annotations[field.var])
310 else:
311 type_ = annotations[field.var]
312 if not field.required and total:
313 assert get_origin(type_) is NotRequired, (
314 f"{name} is not 'required', its type should be `NotRequired`, or the typed dict should not be `total`."
315 )
316 types = set(get_args(type_))
317 types.discard(NoneType)
318 type_ = types.pop()
319 assert not types
321 match field.type:
322 case "boolean":
323 assert type_ is bool, (
324 f"{name} is 'boolean', it should be 'bool', instead it is {type_}"
325 )
326 case "fixed":
327 assert type_ is str, (
328 f"{name} is 'fixed', it should be 'str', instead it is {type_}"
329 )
330 case "text-single":
331 assert type_ is str, (
332 f"{name} is 'text-single', it should be 'str', instead it is {type_}"
333 )
334 case "text-multi":
335 assert type_ is str, (
336 f"{name} is 'text-multi', it should be 'str', instead it is {type_}"
337 )
338 case "jid-single":
339 assert type_ is JID, (
340 f"{name} is 'jid-single', it should be 'JID', instead it is {type_}"
341 )
342 case "jid-multi":
343 assert type_ is list[JID], (
344 f"{name} is 'jid-multi', it should be 'list', instead it is {type_}"
345 )
346 case "list-single":
347 assert type_ is str, (
348 f"{name} is 'list-single', it should be 'str', instead it is {type_}"
349 )
350 case "list-multi":
351 assert type_ is list[str], (
352 f"{name} is 'list-multi', it should be 'list', instead it is {type_}"
353 )
354 case "text-private":
355 assert type_ is str, (
356 f"{name} is 'text-private', it should be 'str', instead it is {type_}"
357 )
359 for attr, typ_ in annotations.items():
360 if get_origin(typ_) is NotRequired:
361 continue
362 assert attr in fields_dict, f"{attr!r} is missing in {fields_dict} {typ_}"
365def _evaluate_forward_ref(ref: ForwardRef) -> object:
366 if sys.version_info < (3, 14):
367 return ref._evaluate(
368 globals(), locals(), recursive_guard=frozenset(), type_params=None
369 )
370 else:
371 return typing.evaluate_forward_ref(ref)
374def _get_type_arg(cls: type, name: str) -> object:
375 for base in getattr(cls, "__orig_bases__", ()):
376 origin = get_origin(base)
377 if origin is None:
378 continue
380 params = getattr(origin, "__type_params__", ())
381 args = get_args(base)
383 for param, arg in zip(params, args):
384 if param.__name__ == name:
385 return arg
387 raise KeyError(f"No type argument named {name!r} on {cls.__name__}")