slixmpp/tests/test_stream_xep_0092.py

import threading

from sleekxmpp.test import *


class TestStreamSet(SleekTest):

    def tearDown(self):
        self.stream_close()

    def testHandleSoftwareVersionRequest(self):
        self.stream_start(mode='client', plugins=['xep_0030', 'xep_0092'])

        self.xmpp['xep_0092'].name = 'SleekXMPP'
        self.xmpp['xep_0092'].version = 'dev'
        self.xmpp['xep_0092'].os = 'Linux'

        self.recv("""
          <iq type="get" id="1">
            <query xmlns="jabber:iq:version" />
          </iq>
        """)

        self.send("""
          <iq type="result" id="1">
            <query xmlns="jabber:iq:version">
              <name>SleekXMPP</name>
              <version>dev</version>
              <os>Linux</os>
            </query>
          </iq>
        """)

    def testMakeSoftwareVersionRequest(self):
        results = []

        def query():
            r = self.xmpp['xep_0092'].get_version('foo@bar')
            results.append((r['software_version']['name'],
                            r['software_version']['version'],
                            r['software_version']['os']))

        self.stream_start(mode='client', plugins=['xep_0030', 'xep_0092'])

        t = threading.Thread(target=query)
        t.start()

        self.send("""
          <iq type="get" id="1" to="foo@bar">
            <query xmlns="jabber:iq:version" />
          </iq>
        """)

        self.recv("""
          <iq type="result" id="1" from="foo@bar" to="tester@localhost">
            <query xmlns="jabber:iq:version">
              <name>Foo</name>
              <version>1.0</version>
              <os>Linux</os>
            </query>
          </iq>
        """)

        t.join()

        expected = [('Foo', '1.0', 'Linux')]
        self.assertEqual(results, expected,
                "Did not receive expected results: %s" % results)


suite = unittest.TestLoader().loadTestsFromTestCase(TestStreamSet)