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()
35 msg_type
= x
.register(callback
)
36 x
.deregister(callback
, msg_type
)
38 def test_all_servers(self
):
39 x
= self
.get_context()
40 self
.assertTrue(isinstance(x
.irpc_all_servers(), list))
42 def test_by_name(self
):
43 x
= self
.get_context()
44 for name
in x
.irpc_all_servers():
45 self
.assertTrue(isinstance(x
.irpc_servers_byname(name
.name
), list))
47 def test_assign_server_id(self
):
48 x
= self
.get_context()
49 self
.assertTrue(isinstance(x
.server_id
, server_id
))
51 def test_ping_speed(self
):
52 server_ctx
= self
.get_context((0, 1))
53 def ping_callback(src
, data
):
54 server_ctx
.send(src
, data
)
57 msg_ping
= server_ctx
.register(ping_callback
)
58 msg_exit
= server_ctx
.register(exit_callback
)
62 client_ctx
= self
.get_context((0, 2))
63 msg_pong
= client_ctx
.register(pong_callback
)
65 client_ctx
.send((0, 1), msg_ping
, "testing")
66 client_ctx
.send((0, 1), msg_ping
, "")