pyrox.dev
1{ writeScriptBin, writeText, python3, connectTo ? "localhost" }:
2let
3 dummyFile = writeText "dummy-file" ''
4 Dear dog,
5
6 Please find this *really* important attachment.
7
8 Yours truly,
9 Bob
10 '';
11in writeScriptBin "send-message" ''
12#!${(python3.withPackages (ps: [ ps.slixmpp ])).interpreter}
13import logging
14import sys
15import signal
16from types import MethodType
17
18from slixmpp import ClientXMPP
19from slixmpp.exceptions import IqError, IqTimeout
20
21
22class CthonTest(ClientXMPP):
23
24 def __init__(self, jid, password):
25 ClientXMPP.__init__(self, jid, password)
26 self.add_event_handler("session_start", self.session_start)
27 self.test_succeeded = False
28
29 async def session_start(self, event):
30 try:
31 # Exceptions in event handlers are printed to stderr but not
32 # propagated, they do not make the script terminate with a non-zero
33 # exit code. We use the `test_succeeded` flag as a workaround and
34 # check it later at the end of the script to exit with a proper
35 # exit code.
36 # Additionally, this flag ensures that this event handler has been
37 # actually run by ClientXMPP, which may well not be the case.
38 await self.test_xmpp_server()
39 self.test_succeeded = True
40 finally:
41 # Even if an exception happens in `test_xmpp_server()`, we still
42 # need to disconnect explicitly, otherwise the process will hang
43 # forever.
44 self.disconnect(wait=True)
45
46 async def test_xmpp_server(self):
47 log = logging.getLogger(__name__)
48 self.send_presence()
49 self.get_roster()
50 # Sending a test message
51 self.send_message(mto="azurediamond@example.com", mbody="Hello, this is dog.", mtype="chat")
52 log.info('Message sent')
53
54 # Test http upload (XEP_0363)
55 try:
56 url = await self['xep_0363'].upload_file("${dummyFile}",timeout=10)
57 except:
58 log.error("ERROR: Cannot run upload command. XEP_0363 seems broken")
59 sys.exit(1)
60 log.info('Upload success!')
61
62 # Test MUC
63 # TODO: use join_muc_wait() after slixmpp 1.8.0 is released.
64 self.plugin['xep_0045'].join_muc('testMucRoom', 'cthon98')
65 log.info('MUC join success!')
66 log.info('XMPP SCRIPT TEST SUCCESS')
67
68def timeout_handler(signalnum, stackframe):
69 print('ERROR: xmpp-sendmessage timed out')
70 sys.exit(1)
71
72if __name__ == '__main__':
73 signal.signal(signal.SIGALRM, timeout_handler)
74 signal.alarm(120)
75 logging.basicConfig(level=logging.DEBUG,
76 format='%(levelname)-8s %(message)s')
77
78 ct = CthonTest('cthon98@example.com', 'nothunter2')
79 ct.register_plugin('xep_0071')
80 ct.register_plugin('xep_0128')
81 # HTTP Upload
82 ct.register_plugin('xep_0363')
83 # MUC
84 ct.register_plugin('xep_0045')
85 ct.connect(("${connectTo}", 5222))
86 ct.process(forever=False)
87
88 if not ct.test_succeeded:
89 sys.exit(1)
90''