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
25 class MessagingTests(TestCase):
27 def get_context(self, *args, **kwargs):
28 return Messaging(*args, **kwargs)
30 def test_register(self):
31 x = self.get_context()
34 msg_type = x.register(callback)
35 x.deregister(callback, msg_type)
37 def test_assign_server_id(self):
38 x = self.get_context()
39 self.assertTrue(isinstance(x.server_id, tuple))
40 self.assertEquals(3, len(x.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)
48 msg_ping = server_ctx.register(ping_callback)
49 msg_exit = server_ctx.register(exit_callback)
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, "")