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 kwargs
['lp_ctx'] = samba
.tests
.env_loadparm()
30 return Messaging(*args
, **kwargs
)
32 def test_register(self
):
33 x
= self
.get_context()
36 msg_type
= x
.register(callback
)
37 x
.deregister(callback
, msg_type
)
39 def test_all_servers(self
):
40 x
= self
.get_context()
41 self
.assertTrue(isinstance(x
.irpc_all_servers(), list))
43 def test_by_name(self
):
44 x
= self
.get_context()
45 for name
in x
.irpc_all_servers():
46 self
.assertTrue(isinstance(x
.irpc_servers_byname(name
.name
), list))
48 def test_assign_server_id(self
):
49 x
= self
.get_context()
50 self
.assertTrue(isinstance(x
.server_id
, server_id
))
52 def test_ping_speed(self
):
53 server_ctx
= self
.get_context((0, 1))
54 def ping_callback(src
, data
):
55 server_ctx
.send(src
, data
)
58 msg_ping
= server_ctx
.register(ping_callback
)
59 msg_exit
= server_ctx
.register(exit_callback
)
63 client_ctx
= self
.get_context((0, 2))
64 msg_pong
= client_ctx
.register(pong_callback
)
66 client_ctx
.send((0, 1), msg_ping
, "testing")
67 client_ctx
.send((0, 1), msg_ping
, "")