summaryrefslogtreecommitdiff
path: root/tests/test_handlers.py
blob: 70eb024814a56d40b370c26b619cabfad7ca3dbf (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
from . sleektest import *
import sleekxmpp
from sleekxmpp.xmlstream.handler import *
from sleekxmpp.xmlstream.matcher import *

class TestHandlers(SleekTest):
    """
    Test using handlers and waiters.
    """

    def setUp(self):
        self.stream_start()

    def tearDown(self):
        self.stream_close()

    def testCallback(self):
        """Test using stream callback handlers."""

        def callback_handler(stanza):
            self.xmpp.sendRaw("""
              <message>
                <body>Success!</body>
              </message>
            """)

        callback = Callback('Test Callback',
                            MatchXPath('{test}tester'),
                            callback_handler)

        self.xmpp.registerHandler(callback)

        self.stream_recv("""<tester xmlns="test" />""")

        msg = self.Message()
        msg['body'] = 'Success!'
        self.stream_send_message(msg)

    def testWaiter(self):
        """Test using stream waiter handler."""

        def waiter_handler(stanza):
            iq = self.xmpp.Iq()
            iq['id'] = 'test'
            iq['type'] = 'set'
            iq['query'] = 'test'
            reply = iq.send(block=True)
            if reply:
                self.xmpp.sendRaw("""
                  <message>
                    <body>Successful: %s</body>
                  </message>
                """ % reply['query'])

        self.xmpp.add_event_handler('message', waiter_handler, threaded=True)

        # Send message to trigger waiter_handler
        self.stream_recv("""
          <message>
            <body>Testing</body>
          </message>
        """)

        # Check that Iq was sent by waiter_handler
        iq = self.Iq()
        iq['id'] = 'test'
        iq['type'] = 'set'
        iq['query'] = 'test'
        self.stream_send_iq(iq)

        # Send the reply Iq
        self.stream_recv("""
          <iq id="test" type="result">
            <query xmlns="test" />
          </iq>
        """)

        # Check that waiter_handler received the reply
        msg = self.Message()
        msg['body'] = 'Successful: test'
        self.stream_send_message(msg)

    def testWaiterTimeout(self):
        """Test that waiter handler is removed after timeout."""

        def waiter_handler(stanza):
            iq = self.xmpp.Iq()
            iq['id'] = 'test2'
            iq['type'] = 'set'
            iq['query'] = 'test2'
            reply = iq.send(block=True, timeout=0)

        self.xmpp.add_event_handler('message', waiter_handler, threaded=True)

        # Start test by triggerig waiter_handler
        self.stream_recv("""<message><body>Start Test</body></message>""")

        # Check that Iq was sent to trigger start of timeout period
        iq = self.Iq()
        iq['id'] = 'test2'
        iq['type'] = 'set'
        iq['query'] = 'test2'
        self.stream_send_iq(iq)

        # Check that the waiter is no longer registered
        waiter_exists = self.xmpp.removeHandler('IqWait_test2')

        self.failUnless(waiter_exists == False,
            "Waiter handler was not removed.")


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