pymessaging: Use the server_id IDL structure rather than a tuple
[kai/samba.git] / source4 / scripting / python / samba / tests / messaging.py
1 # -*- coding: utf-8 -*-
2 #
3 # Unix SMB/CIFS implementation.
4 # Copyright © Jelmer Vernooij <jelmer@samba.org> 2008
5 #
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 3 of the License, or
9 # (at your option) any later version.
10 #
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14 # GNU General Public License for more details.
15 #
16 # You should have received a copy of the GNU General Public License
17 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
18 #
19
20 """Tests for samba.messaging."""
21
22 from samba.messaging import Messaging
23 from samba.tests import TestCase
24 from samba.dcerpc.server_id import server_id
25
26 class MessagingTests(TestCase):
27
28     def get_context(self, *args, **kwargs):
29         return Messaging(*args, **kwargs)
30
31     def test_register(self):
32         x = self.get_context()
33         def callback():
34             pass
35         msg_type = x.register(callback)
36         x.deregister(callback, msg_type)
37
38     def test_assign_server_id(self):
39         x = self.get_context()
40         self.assertTrue(isinstance(x.server_id, server_id))
41
42     def test_ping_speed(self):
43         server_ctx = self.get_context((0, 1))
44         def ping_callback(src, data):
45                 server_ctx.send(src, data)
46         def exit_callback():
47                 print "received exit"
48         msg_ping = server_ctx.register(ping_callback)
49         msg_exit = server_ctx.register(exit_callback)
50
51         def pong_callback():
52                 print "received pong"
53         client_ctx = self.get_context((0, 2))
54         msg_pong = client_ctx.register(pong_callback)
55
56         client_ctx.send((0, 1), msg_ping, "testing")
57         client_ctx.send((0, 1), msg_ping, "")
58