source4/scripting/python/samba/tests/messaging.py

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