pymessaging: Use the server_id IDL structure rather than a tuple
[Samba/gebeck_regimport.git] / source4 / scripting / python / samba / tests / messaging.py
blob3348ff8deb21c453e81654f8f223f484921e530f
1 # -*- coding: utf-8 -*-
3 # Unix SMB/CIFS implementation.
4 # Copyright © Jelmer Vernooij <jelmer@samba.org> 2008
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.
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.
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/>.
20 """Tests for samba.messaging."""
22 from samba.messaging import Messaging
23 from samba.tests import TestCase
24 from samba.dcerpc.server_id import server_id
26 class MessagingTests(TestCase):
28 def get_context(self, *args, **kwargs):
29 return Messaging(*args, **kwargs)
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)
38 def test_assign_server_id(self):
39 x = self.get_context()
40 self.assertTrue(isinstance(x.server_id, server_id))
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)
51 def pong_callback():
52 print "received pong"
53 client_ctx = self.get_context((0, 2))
54 msg_pong = client_ctx.register(pong_callback)
56 client_ctx.send((0, 1), msg_ping, "testing")
57 client_ctx.send((0, 1), msg_ping, "")