nixos/tests/xmpp/xmpp-sendmessage.nix

   1 { writeScriptBin, writeText, python3, connectTo ? "localhost" }:
   2 let
   3   dummyFile = writeText "dummy-file" ''
   4     Dear dog,
   5
   6     Please find this *really* important attachment.
   7
   8     Yours truly,
   9     Bob
  10   '';
  11 in writeScriptBin "send-message" ''
  12 #!${(python3.withPackages (ps: [ ps.slixmpp ])).interpreter}
  13 import logging
  14 import sys
  15 import signal
  16 from types import MethodType
  17
  18 from slixmpp import ClientXMPP
  19 from slixmpp.exceptions import IqError, IqTimeout
  20
  21
  22 class 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
  68 def timeout_handler(signalnum, stackframe):
  69     print('ERROR: xmpp-sendmessage timed out')
  70     sys.exit(1)
  71
  72 if __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 ''