projects / ganeti-local / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | tree
raw | inline | side by side
Merge branch 'devel-2.4'
[ganeti-local] / test / ganeti.daemon_unittest.py
diff --git a/test/ganeti.daemon_unittest.py b/test/ganeti.daemon_unittest.py
index 1c9160e..5be22dd 100755 (executable)
--- a/test/ganeti.daemon_unittest.py
+++ b/test/ganeti.daemon_unittest.py
@@ -149,8 +149,8 @@ class TestMainloop(testutils.GanetiTestCase):
 
 class _MyAsyncUDPSocket(daemon.AsyncUDPSocket):
 
-  def __init__(self):
-    daemon.AsyncUDPSocket.__init__(self)
+  def __init__(self, family):
+    daemon.AsyncUDPSocket.__init__(self, family)
     self.received = []
     self.error_count = 0
 
@@ -166,15 +166,17 @@ class _MyAsyncUDPSocket(daemon.AsyncUDPSocket):
     raise
 
 
-class TestAsyncUDPSocket(testutils.GanetiTestCase):
-  """Test daemon.AsyncUDPSocket"""
+class _BaseAsyncUDPSocketTest:
+  """Base class for  AsyncUDPSocket tests"""
+
+  family = None
+  address = None
 
   def setUp(self):
-    testutils.GanetiTestCase.setUp(self)
     self.mainloop = daemon.Mainloop()
-    self.server = _MyAsyncUDPSocket()
-    self.client = _MyAsyncUDPSocket()
-    self.server.bind(("127.0.0.1", 0))
+    self.server = _MyAsyncUDPSocket(self.family)
+    self.client = _MyAsyncUDPSocket(self.family)
+    self.server.bind((self.address, 0))
     self.port = self.server.getsockname()[1]
     # Save utils.IgnoreSignals so we can do evil things to it...
     self.saved_utils_ignoresignals = utils.IgnoreSignals
@@ -187,38 +189,38 @@ class TestAsyncUDPSocket(testutils.GanetiTestCase):
     testutils.GanetiTestCase.tearDown(self)
 
   def testNoDoubleBind(self):
-    self.assertRaises(socket.error, self.client.bind, ("127.0.0.1", self.port))
+    self.assertRaises(socket.error, self.client.bind, (self.address, self.port))
 
   def testAsyncClientServer(self):
-    self.client.enqueue_send("127.0.0.1", self.port, "p1")
-    self.client.enqueue_send("127.0.0.1", self.port, "p2")
-    self.client.enqueue_send("127.0.0.1", self.port, "terminate")
+    self.client.enqueue_send(self.address, self.port, "p1")
+    self.client.enqueue_send(self.address, self.port, "p2")
+    self.client.enqueue_send(self.address, self.port, "terminate")
     self.mainloop.Run()
     self.assertEquals(self.server.received, ["p1", "p2", "terminate"])
 
   def testSyncClientServer(self):
     self.client.handle_write()
-    self.client.enqueue_send("127.0.0.1", self.port, "p1")
-    self.client.enqueue_send("127.0.0.1", self.port, "p2")
+    self.client.enqueue_send(self.address, self.port, "p1")
+    self.client.enqueue_send(self.address, self.port, "p2")
     while self.client.writable():
       self.client.handle_write()
     self.server.process_next_packet()
     self.assertEquals(self.server.received, ["p1"])
     self.server.process_next_packet()
     self.assertEquals(self.server.received, ["p1", "p2"])
-    self.client.enqueue_send("127.0.0.1", self.port, "p3")
+    self.client.enqueue_send(self.address, self.port, "p3")
     while self.client.writable():
       self.client.handle_write()
     self.server.process_next_packet()
     self.assertEquals(self.server.received, ["p1", "p2", "p3"])
 
   def testErrorHandling(self):
-    self.client.enqueue_send("127.0.0.1", self.port, "p1")
-    self.client.enqueue_send("127.0.0.1", self.port, "p2")
-    self.client.enqueue_send("127.0.0.1", self.port, "error")
-    self.client.enqueue_send("127.0.0.1", self.port, "p3")
-    self.client.enqueue_send("127.0.0.1", self.port, "error")
-    self.client.enqueue_send("127.0.0.1", self.port, "terminate")
+    self.client.enqueue_send(self.address, self.port, "p1")
+    self.client.enqueue_send(self.address, self.port, "p2")
+    self.client.enqueue_send(self.address, self.port, "error")
+    self.client.enqueue_send(self.address, self.port, "p3")
+    self.client.enqueue_send(self.address, self.port, "error")
+    self.client.enqueue_send(self.address, self.port, "terminate")
     self.assertRaises(errors.GenericError, self.mainloop.Run)
     self.assertEquals(self.server.received,
                       ["p1", "p2", "error"])
@@ -234,11 +236,11 @@ class TestAsyncUDPSocket(testutils.GanetiTestCase):
 
   def testSignaledWhileReceiving(self):
     utils.IgnoreSignals = lambda fn, *args, **kwargs: None
-    self.client.enqueue_send("127.0.0.1", self.port, "p1")
-    self.client.enqueue_send("127.0.0.1", self.port, "p2")
+    self.client.enqueue_send(self.address, self.port, "p1")
+    self.client.enqueue_send(self.address, self.port, "p2")
     self.server.handle_read()
     self.assertEquals(self.server.received, [])
-    self.client.enqueue_send("127.0.0.1", self.port, "terminate")
+    self.client.enqueue_send(self.address, self.port, "terminate")
     utils.IgnoreSignals = self.saved_utils_ignoresignals
     self.mainloop.Run()
     self.assertEquals(self.server.received, ["p1", "p2", "terminate"])
@@ -246,7 +248,37 @@ class TestAsyncUDPSocket(testutils.GanetiTestCase):
   def testOversizedDatagram(self):
     oversized_data = (constants.MAX_UDP_DATA_SIZE + 1) * "a"
     self.assertRaises(errors.UdpDataSizeError, self.client.enqueue_send,
-                      "127.0.0.1", self.port, oversized_data)
+                      self.address, self.port, oversized_data)
+
+
+class TestAsyncIP4UDPSocket(testutils.GanetiTestCase, _BaseAsyncUDPSocketTest):
+  """Test IP4 daemon.AsyncUDPSocket"""
+
+  family = socket.AF_INET
+  address = "127.0.0.1"
+
+  def setUp(self):
+    testutils.GanetiTestCase.setUp(self)
+    _BaseAsyncUDPSocketTest.setUp(self)
+
+  def tearDown(self):
+    testutils.GanetiTestCase.tearDown(self)
+    _BaseAsyncUDPSocketTest.tearDown(self)
+
+
+class TestAsyncIP6UDPSocket(testutils.GanetiTestCase, _BaseAsyncUDPSocketTest):
+  """Test IP6 daemon.AsyncUDPSocket"""
+
+  family = socket.AF_INET6
+  address = "::1"
+
+  def setUp(self):
+    testutils.GanetiTestCase.setUp(self)
+    _BaseAsyncUDPSocketTest.setUp(self)
+
+  def tearDown(self):
+    testutils.GanetiTestCase.tearDown(self)
+    _BaseAsyncUDPSocketTest.tearDown(self)
 
 
 class _MyAsyncStreamServer(daemon.AsyncStreamServer):
@@ -273,10 +305,11 @@ class _MyAsyncStreamServer(daemon.AsyncStreamServer):
 class _MyMessageStreamHandler(daemon.AsyncTerminatedMessageStream):
 
   def __init__(self, connected_socket, client_address, terminator, family,
-               message_fn, client_id):
+               message_fn, client_id, unhandled_limit):
     daemon.AsyncTerminatedMessageStream.__init__(self, connected_socket,
                                                  client_address,
-                                                 terminator, family)
+                                                 terminator, family,
+                                                 unhandled_limit)
     self.message_fn = message_fn
     self.client_id = client_id
     self.error_count = 0
@@ -301,6 +334,7 @@ class TestAsyncStreamServerTCP(testutils.GanetiTestCase):
     self.server = _MyAsyncStreamServer(self.family, self.address,
                                        self.handle_connection)
     self.client_handler = _MyMessageStreamHandler
+    self.unhandled_limit = None
     self.terminator = "\3"
     self.address = self.server.getsockname()
     self.clients = []
@@ -339,7 +373,7 @@ class TestAsyncStreamServerTCP(testutils.GanetiTestCase):
     client_handler = self.client_handler(connected_socket, client_address,
                                          self.terminator, self.family,
                                          self.handle_message,
-                                         client_id)
+                                         client_id, self.unhandled_limit)
     self.connections.append(client_handler)
     self.countTerminate("connect_terminate_count")
 
@@ -494,6 +528,61 @@ class TestAsyncStreamServerTCP(testutils.GanetiTestCase):
     self.assertEquals(client1.recv(4096), "r0\3r1\3r2\3")
     self.assertRaises(socket.error, client2.recv, 4096)
 
+  def testLimitedUnhandledMessages(self):
+    self.connect_terminate_count = None
+    self.message_terminate_count = 3
+    self.unhandled_limit = 2
+    client1 = self.getClient()
+    client2 = self.getClient()
+    client1.send("one\3composed\3long\3message\3")
+    client2.send("c2one\3")
+    self.mainloop.Run()
+    self.assertEquals(self.messages[0], ["one", "composed"])
+    self.assertEquals(self.messages[1], ["c2one"])
+    self.assertFalse(self.connections[0].readable())
+    self.assert_(self.connections[1].readable())
+    self.connections[0].send_message("r0")
+    self.message_terminate_count = None
+    client1.send("another\3")
+    # when we write replies messages queued also get handled, but not the ones
+    # in the socket.
+    while self.connections[0].writable():
+      self.connections[0].handle_write()
+    self.assertFalse(self.connections[0].readable())
+    self.assertEquals(self.messages[0], ["one", "composed", "long"])
+    self.connections[0].send_message("r1")
+    self.connections[0].send_message("r2")
+    while self.connections[0].writable():
+      self.connections[0].handle_write()
+    self.assertEquals(self.messages[0], ["one", "composed", "long", "message"])
+    self.assert_(self.connections[0].readable())
+
+  def testLimitedUnhandledMessagesOne(self):
+    self.connect_terminate_count = None
+    self.message_terminate_count = 2
+    self.unhandled_limit = 1
+    client1 = self.getClient()
+    client2 = self.getClient()
+    client1.send("one\3composed\3message\3")
+    client2.send("c2one\3")
+    self.mainloop.Run()
+    self.assertEquals(self.messages[0], ["one"])
+    self.assertEquals(self.messages[1], ["c2one"])
+    self.assertFalse(self.connections[0].readable())
+    self.assertFalse(self.connections[1].readable())
+    self.connections[0].send_message("r0")
+    self.message_terminate_count = None
+    while self.connections[0].writable():
+      self.connections[0].handle_write()
+    self.assertFalse(self.connections[0].readable())
+    self.assertEquals(self.messages[0], ["one", "composed"])
+    self.connections[0].send_message("r2")
+    self.connections[0].send_message("r3")
+    while self.connections[0].writable():
+      self.connections[0].handle_write()
+    self.assertEquals(self.messages[0], ["one", "composed", "message"])
+    self.assert_(self.connections[0].readable())
+
 
 class TestAsyncStreamServerUnixPath(TestAsyncStreamServerTCP):
   """Test daemon.AsyncStreamServer with a Unix path connection"""
Unnamed repository; edit this file 'description' to name the repository.