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