Coverage for slidge/util/test.py: 97%
176 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-09 03:59 +0000
1# type:ignore
2import os
3import tempfile
4import types
5from pathlib import Path
6from typing import ClassVar
7from xml.dom.minidom import parseString
9from slixmpp import JID, ElementBase, Iq, Message, Presence
10from slixmpp.stanza.error import Error
11from slixmpp.test import SlixTest, TestTransport
12from slixmpp.xmlstream import highlight, tostring
13from sqlalchemy import create_engine, delete
15from slidge import (
16 BaseGateway,
17 BaseSession,
18 LegacyBookmarks,
19 LegacyContact,
20 LegacyMUC,
21 LegacyParticipant,
22 LegacyRoster,
23)
25from ..command import Command
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
32from ..util import SubclassableOnce
35class SlixTestPlus(SlixTest):
36 def setUp(self) -> None:
37 super().setUp()
38 Error.namespace = "jabber:component:accept"
40 def next_sent(self, timeout: float = 0.05) -> Message | Iq | Presence | None:
41 self.wait_for_send_queue()
42 sent = self.xmpp.socket.next_sent(timeout=timeout)
43 if sent is None:
44 return None
45 xml = self.parse_xml(sent)
46 self.fix_namespaces(xml, "jabber:component:accept")
47 sent = self.xmpp._build_stanza(xml, "jabber:component:accept")
48 return sent
51class SlidgeTest(SlixTestPlus):
52 plugin: types.ModuleType | dict
54 class Config:
55 jid = "aim.shakespeare.lit"
56 secret = "test"
57 server = "shakespeare.lit"
58 port = 5222
59 upload_service = "upload.test"
60 home_dir = Path(tempfile.mkdtemp())
61 user_jid_validator = ".*"
62 admins: ClassVar[list[str]] = []
63 upload_requester = None
64 ignore_delay_threshold = 300
66 @classmethod
67 def setUpClass(cls) -> None:
68 for k, v in vars(cls.Config).items():
69 setattr(config, k.upper(), v)
70 if hasattr(cls, "plugin"):
71 subclasses = SubclassableOnce._SubclassableOnce__subclasses
72 subclasses.update(
73 {
74 BaseGateway: find_subclass(cls.plugin, BaseGateway),
75 BaseSession: find_subclass(cls.plugin, BaseSession),
76 LegacyRoster: find_subclass(cls.plugin, LegacyRoster, base_ok=True),
77 LegacyContact: find_subclass(
78 cls.plugin, LegacyContact, base_ok=True
79 ),
80 LegacyMUC: find_subclass(cls.plugin, LegacyMUC, base_ok=True),
81 LegacyBookmarks: find_subclass(
82 cls.plugin, LegacyBookmarks, base_ok=True
83 ),
84 }
85 )
86 for subclass in subclasses.values():
87 if subclass is not None:
88 subclass.__abstractmethods__ = set()
90 def setUp(self) -> None:
91 # workaround for duplicate output of sql alchemy's log, cf
92 # https://stackoverflow.com/a/76498428/5902284
93 from sqlalchemy import log as sqlalchemy_log
94 from sqlalchemy.pool import StaticPool
96 sqlalchemy_log._add_default_handler = lambda x: None
97 db_url = os.getenv("SLIDGETEST_DB_URL", "sqlite+pysqlite:///:memory:")
98 engine = self.db_engine = create_engine(
99 db_url, poolclass=StaticPool if db_url.startswith("postgresql+") else None
100 )
101 Base.metadata.create_all(engine)
102 BaseGateway.store = SlidgeStore(engine)
103 BaseGateway._test_mode = True
105 self.xmpp = BaseGateway.get_self_or_unique_subclass()()
107 self.xmpp.TEST_MODE = True
108 avatar_cache.store = self.xmpp.store.avatars
109 avatar_cache.set_dir(Path(tempfile.mkdtemp()))
110 self.xmpp._always_send_everything = True
111 engine.echo = True
113 self.xmpp.connection_made(TestTransport(self.xmpp))
114 self.xmpp.session_bind_event.set()
115 # Remove unique ID prefix to make it easier to test
116 self.xmpp._id_prefix = ""
117 self.xmpp.default_lang = None
118 self.xmpp.peer_default_lang = None
120 def new_id() -> str:
121 self.xmpp._id += 1
122 return str(self.xmpp._id)
124 self.xmpp._id = 0
125 self.xmpp.new_id = new_id
127 # Must have the stream header ready for xmpp.process() to work.
128 header = self.xmpp.stream_header
130 self.xmpp.data_received(header)
131 self.wait_for_send_queue()
133 self.xmpp.socket.next_sent()
134 self.xmpp.socket.next_sent()
136 # Some plugins require messages to have ID values. Set
137 # this to True in tests related to those plugins.
138 self.xmpp.use_message_ids = False
139 self.xmpp.use_presence_ids = False
140 Error.namespace = "jabber:component:accept"
142 def _add_slidge_user(
143 self,
144 jid: str | JID = "romeo@shakespeare.lit",
145 legacy_module_data: dict | None = None,
146 preferences: dict | None = None,
147 ) -> None:
148 with self.xmpp.store.session() as orm:
149 user = GatewayUser(
150 jid=JID(jid),
151 legacy_module_data=legacy_module_data or {},
152 preferences=preferences
153 or {"sync_avatar": False, "sync_presence": False},
154 )
155 orm.add(user)
156 orm.commit()
157 self.run_coro(
158 self.xmpp._BaseGateway__dispatcher._on_user_register(Iq(sfrom=JID(jid)))
159 )
160 welcome = self.next_sent()
161 assert welcome["body"]
162 stanza = self.next_sent()
163 assert "logging in" in stanza["status"].lower(), stanza
164 stanza = self.next_sent()
165 assert "syncing contacts" in stanza["status"].lower(), stanza
166 if self.xmpp.GROUPS:
167 stanza = self.next_sent()
168 assert "syncing groups" in stanza["status"].lower(), stanza
169 probe = self.next_sent()
170 assert probe.get_type() == "probe"
171 stanza = self.next_sent()
172 assert stanza["status"].lower()
174 def user_session(self, jid: str | JID = "romeo@shakespeare.lit") -> BaseSession:
175 return BaseSession.get_self_or_unique_subclass().from_jid(JID(jid))
177 def get_joined_muc(
178 self,
179 legacy_id: str | int,
180 user_jid: str | JID = "romeo@shakespeare.lit",
181 resource: str = "gajim",
182 ) -> LegacyMUC:
183 muc: LegacyMUC = self.run_coro(
184 self.user_session(user_jid).bookmarks.by_legacy_id(legacy_id)
185 )
186 muc.add_user_resource(resource)
187 return muc
189 def tearDown(self) -> None:
190 self.db_engine.echo = False
191 super().tearDown()
192 Base.metadata.drop_all(self.xmpp.store._engine)
193 self.db_engine.dispose()
194 _sessions.clear()
196 def setup_logged_session(self, n_contacts: int = 0) -> None:
197 with self.xmpp.store.session() as orm:
198 user = GatewayUser(
199 jid=JID("romeo@montague.lit/gajim").bare,
200 legacy_module_data={"username": "romeo", "city": ""},
201 preferences={"sync_avatar": True, "sync_presence": True},
202 )
203 orm.add(user)
204 orm.commit()
206 with self.xmpp.store.session() as session:
207 session.execute(delete(Contact))
208 session.commit()
210 self.run_coro(
211 self.xmpp._BaseGateway__dispatcher._on_user_register(
212 Iq(sfrom="romeo@montague.lit/gajim")
213 )
214 )
215 welcome = self.next_sent()
216 assert welcome["body"], welcome
217 stanza = self.next_sent()
218 assert "logging in" in stanza["status"].lower(), stanza
219 stanza = self.next_sent()
220 assert "syncing contacts" in stanza["status"].lower(), stanza
221 if BaseGateway.get_self_or_unique_subclass().GROUPS:
222 stanza = self.next_sent()
223 assert "syncing groups" in stanza["status"].lower(), stanza
224 probe = self.next_sent()
225 assert probe.get_type() == "probe"
226 stanza = self.next_sent()
227 assert "yup" in stanza["status"].lower(), stanza
228 self.romeo: BaseSession = BaseSession.get_self_or_unique_subclass().from_jid(
229 JID("romeo@montague.lit")
230 )
232 self.juliet: LegacyContact = self.run_coro(
233 self.romeo.contacts.by_legacy_id("juliet")
234 )
235 self.room: LegacyMUC = self.run_coro(self.romeo.bookmarks.by_legacy_id("room"))
236 self.first_witch: LegacyParticipant = self.run_coro(
237 self.room.get_participant("firstwitch")
238 )
239 self.send( # language=XML
240 """
241 <iq type="get"
242 to="romeo@montague.lit"
243 id="1"
244 from="aim.shakespeare.lit">
245 <pubsub xmlns="http://jabber.org/protocol/pubsub">
246 <items node="urn:xmpp:avatar:metadata" />
247 </pubsub>
248 </iq>
249 """
250 )
252 @classmethod
253 def tearDownClass(cls) -> None:
254 reset_subclasses()
257def format_stanza(stanza: ElementBase) -> str:
258 return highlight(
259 "\n".join(parseString(tostring(stanza.xml)).toprettyxml().split("\n")[1:])
260 )
263def find_subclass(o, parent, base_ok: bool = False): # noqa
264 try:
265 vals = vars(o).values()
266 except TypeError:
267 vals = o.values()
268 for x in vals:
269 try:
270 if issubclass(x, parent) and x is not parent:
271 return x
272 except TypeError:
273 pass
274 else:
275 if base_ok:
276 return parent
277 else:
278 raise RuntimeError
281def reset_subclasses() -> None:
282 """
283 Reset registered subclasses between test classes.
285 Needed because these classes are meant to only be subclassed once and raise
286 exceptions otherwise.
287 """
288 BaseSession.reset_subclass()
289 BaseGateway.reset_subclass()
290 LegacyRoster.reset_subclass()
291 LegacyContact.reset_subclass()
292 LegacyMUC.reset_subclass()
293 LegacyBookmarks.reset_subclass()
294 LegacyParticipant.reset_subclass()
295 # reset_commands()
298def reset_commands() -> None:
299 Command.subclasses = [
300 c for c in Command.subclasses if str(c).startswith("<class 'slidge.core")
301 ]