Coverage for slidge/util/test.py: 96%
161 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-18 04:30 +0000
1# type:ignore
2import os
3import tempfile
4import types
5from pathlib import Path
6from typing import ClassVar
8from slixmpp import JID, Iq, Message, Presence
9from slixmpp.stanza.error import Error
10from slixmpp.test import SlixTest, TestTransport
11from sqlalchemy import create_engine, delete
13from slidge import (
14 BaseGateway,
15 BaseSession,
16 LegacyContact,
17 LegacyMUC,
18 LegacyParticipant,
19)
21from ..core import config
22from ..core.session import _sessions
23from ..db import SlidgeStore
24from ..db.avatar import avatar_cache
25from ..db.meta import Base
26from ..db.models import Contact, GatewayUser
29class SlixTestPlus(SlixTest):
30 def setUp(self) -> None:
31 super().setUp()
32 Error.namespace = "jabber:component:accept"
34 def next_sent(self, timeout: float = 0.05) -> Message | Iq | Presence | None:
35 self.wait_for_send_queue()
36 sent = self.xmpp.socket.next_sent(timeout=timeout)
37 if sent is None:
38 return None
39 xml = self.parse_xml(sent)
40 self.fix_namespaces(xml, "jabber:component:accept")
41 sent = self.xmpp._build_stanza(xml, "jabber:component:accept")
42 return sent
45class SlidgeTest(SlixTestPlus):
46 plugin: types.ModuleType | dict
48 class Config:
49 jid = "aim.shakespeare.lit"
50 secret = "test"
51 server = "shakespeare.lit"
52 port = 5222
53 upload_service = "upload.test"
54 home_dir = Path(tempfile.mkdtemp())
55 user_jid_validator = ".*"
56 admins: ClassVar[list[str]] = []
57 upload_requester = None
58 ignore_delay_threshold = 300
60 gateway_cls: type[BaseGateway]
62 @classmethod
63 def setUpClass(cls) -> None:
64 for k, v in vars(cls.Config).items():
65 setattr(config, k.upper(), v)
66 if not hasattr(cls, "plugin"):
67 raise RuntimeError(
68 f"{cls.__name__} must set the 'plugin' attribute to the"
69 " module (or dict) containing the gateway class, e.g."
70 " 'plugin = globals()' or 'plugin = my_legacy_module'."
71 )
72 cls.gateway_cls = find_subclass(cls.plugin, BaseGateway)
73 session_cls = cls.gateway_cls.session_cls
74 wired = [
75 cls.gateway_cls,
76 session_cls,
77 session_cls.roster_cls,
78 session_cls.roster_cls.contact_cls,
79 session_cls.bookmarks_cls,
80 session_cls.bookmarks_cls.muc_cls,
81 session_cls.bookmarks_cls.muc_cls.participant_cls,
82 ]
83 for wired_cls in wired:
84 wired_cls.__abstractmethods__ = frozenset()
86 def setUp(self) -> None:
87 # workaround for duplicate output of sql alchemy's log, cf
88 # https://stackoverflow.com/a/76498428/5902284
89 from sqlalchemy import log as sqlalchemy_log
90 from sqlalchemy.pool import StaticPool
92 sqlalchemy_log._add_default_handler = lambda x: None
93 db_url = os.getenv("SLIDGETEST_DB_URL", "sqlite+pysqlite:///:memory:")
94 engine = self.db_engine = create_engine(
95 db_url, poolclass=StaticPool if db_url.startswith("postgresql+") else None
96 )
97 Base.metadata.create_all(engine)
98 BaseGateway.store = SlidgeStore(engine)
99 BaseGateway._test_mode = True
101 # Clean up commands potentially left behind by a previous test.
102 LegacyContact.commands.clear()
103 LegacyContact.commands_chat.clear()
104 LegacyMUC.commands.clear()
105 LegacyMUC.commands_chat.clear()
107 self.xmpp = self.gateway_cls()
109 self.xmpp.TEST_MODE = True
110 avatar_cache.store = self.xmpp.store.avatars
111 avatar_cache.set_dir(Path(tempfile.mkdtemp()))
112 self.xmpp._always_send_everything = True
113 engine.echo = True
115 self.xmpp.connection_made(TestTransport(self.xmpp))
116 self.xmpp.session_bind_event.set()
117 # Remove unique ID prefix to make it easier to test
118 self.xmpp._id_prefix = ""
119 self.xmpp.default_lang = None
120 self.xmpp.peer_default_lang = None
122 def new_id() -> str:
123 self.xmpp._id += 1
124 return str(self.xmpp._id)
126 self.xmpp._id = 0
127 self.xmpp.new_id = new_id
129 # Must have the stream header ready for xmpp.process() to work.
130 header = self.xmpp.stream_header
132 self.xmpp.data_received(header)
133 self.wait_for_send_queue()
135 self.xmpp.socket.next_sent()
136 self.xmpp.socket.next_sent()
138 # Some plugins require messages to have ID values. Set
139 # this to True in tests related to those plugins.
140 self.xmpp.use_message_ids = False
141 self.xmpp.use_presence_ids = False
142 Error.namespace = "jabber:component:accept"
144 def _add_slidge_user(
145 self,
146 jid: str | JID = "romeo@shakespeare.lit",
147 legacy_module_data: dict | None = None,
148 preferences: dict | None = None,
149 ) -> None:
150 with self.xmpp.store.session() as orm:
151 user = GatewayUser(
152 jid=JID(jid),
153 legacy_module_data=legacy_module_data or {},
154 preferences=preferences
155 or {"sync_avatar": False, "sync_presence": False},
156 )
157 orm.add(user)
158 orm.commit()
159 self.run_coro(
160 self.xmpp._BaseGateway__dispatcher._on_user_register(Iq(sfrom=JID(jid)))
161 )
162 welcome = self.next_sent()
163 assert welcome["body"]
164 stanza = self.next_sent()
165 assert "logging in" in stanza["status"].lower(), stanza
166 stanza = self.next_sent()
167 assert "syncing contacts" in stanza["status"].lower(), stanza
168 if self.xmpp.GROUPS:
169 stanza = self.next_sent()
170 assert "syncing groups" in stanza["status"].lower(), stanza
171 probe = self.next_sent()
172 assert probe.get_type() == "probe"
173 stanza = self.next_sent()
174 assert stanza["status"].lower()
176 def user_session(self, jid: str | JID = "romeo@shakespeare.lit") -> BaseSession:
177 return self.xmpp.session_cls.from_jid(JID(jid))
179 def get_joined_muc(
180 self,
181 legacy_id: str | int,
182 user_jid: str | JID = "romeo@shakespeare.lit",
183 resource: str = "gajim",
184 ) -> LegacyMUC:
185 muc: LegacyMUC = self.run_coro(
186 self.user_session(user_jid).bookmarks.by_legacy_id(legacy_id)
187 )
188 muc.add_user_resource(resource)
189 return muc
191 def tearDown(self) -> None:
192 self.db_engine.echo = False
193 super().tearDown()
194 Base.metadata.drop_all(self.xmpp.store._engine)
195 self.db_engine.dispose()
196 _sessions.clear()
198 def setup_logged_session(self, n_contacts: int = 0) -> None:
199 with self.xmpp.store.session() as orm:
200 user = GatewayUser(
201 jid=JID("romeo@montague.lit/gajim").bare,
202 legacy_module_data={"username": "romeo", "city": ""},
203 preferences={"sync_avatar": True, "sync_presence": True},
204 )
205 orm.add(user)
206 orm.commit()
208 with self.xmpp.store.session() as session:
209 session.execute(delete(Contact))
210 session.commit()
212 self.run_coro(
213 self.xmpp._BaseGateway__dispatcher._on_user_register(
214 Iq(sfrom="romeo@montague.lit/gajim")
215 )
216 )
217 welcome = self.next_sent()
218 assert welcome["body"], welcome
219 stanza = self.next_sent()
220 assert "logging in" in stanza["status"].lower(), stanza
221 stanza = self.next_sent()
222 assert "syncing contacts" in stanza["status"].lower(), stanza
223 if self.xmpp.GROUPS:
224 stanza = self.next_sent()
225 assert "syncing groups" in stanza["status"].lower(), stanza
226 probe = self.next_sent()
227 assert probe.get_type() == "probe"
228 stanza = self.next_sent()
229 assert "yup" in stanza["status"].lower(), stanza
230 self.romeo: BaseSession = self.xmpp.session_cls.from_jid(
231 JID("romeo@montague.lit")
232 )
234 self.juliet: LegacyContact = self.run_coro(
235 self.romeo.contacts.by_legacy_id("juliet")
236 )
237 self.room: LegacyMUC = self.run_coro(self.romeo.bookmarks.by_legacy_id("room"))
238 self.first_witch: LegacyParticipant = self.run_coro(
239 self.room.get_participant("firstwitch")
240 )
241 self.send( # language=XML
242 """
243 <iq type="get"
244 to="romeo@montague.lit"
245 id="1"
246 from="aim.shakespeare.lit">
247 <pubsub xmlns="http://jabber.org/protocol/pubsub">
248 <items node="urn:xmpp:avatar:metadata" />
249 </pubsub>
250 </iq>
251 """
252 )
255def find_subclass(o, parent, base_ok: bool = False): # noqa
256 try:
257 vals = vars(o).values()
258 except TypeError:
259 vals = o.values()
260 for x in vals:
261 try:
262 if issubclass(x, parent) and x is not parent:
263 return x
264 except TypeError:
265 pass
266 if base_ok:
267 return parent
268 else:
269 raise RuntimeError(f"Could not find a subclass of {parent} in {o}")