[2855] Add builder module tests

src/lib/python/isc/memmgr/tests/Makefile.am

@@ -1,5 +1,5 @@
 PYCOVERAGE_RUN = @PYCOVERAGE_RUN@
-PYTESTS = datasrc_info_tests.py
+PYTESTS = builder_tests.py datasrc_info_tests.py
 EXTRA_DIST = $(PYTESTS)
 
 # If necessary (rare cases), explicitly specify paths to dynamic libraries

src/lib/python/isc/memmgr/tests/builder_tests.py

@@ -0,0 +1,80 @@
+# Copyright (C) 2013  Internet Systems Consortium.
+#
+# Permission to use, copy, modify, and distribute this software for any
+# purpose with or without fee is hereby granted, provided that the above
+# copyright notice and this permission notice appear in all copies.
+#
+# THE SOFTWARE IS PROVIDED "AS IS" AND INTERNET SYSTEMS CONSORTIUM
+# DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
+# INTERNET SYSTEMS CONSORTIUM BE LIABLE FOR ANY SPECIAL, DIRECT,
+# INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING
+# FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
+# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION
+# WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+import unittest
+import socket
+import threading
+
+import isc.log
+from isc.memmgr.builder import *
+
+class TestMemorySegmentBuilder(unittest.TestCase):
+    def _create_builder_thread(self):
+        (self._master_sock, self._builder_sock) = \
+            socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
+
+        self._builder_command_queue = []
+        self._builder_response_queue = []
+
+        self._builder_cv = threading.Condition()
+        self._builder_lock = threading.Lock()
+
+        self._builder = MemorySegmentBuilder(self._builder_sock,
+                                             self._builder_cv,
+                                             self._builder_lock,
+                                             self._builder_command_queue,
+                                             self._builder_response_queue)
+        self._builder_thread = threading.Thread(target=self._builder.run)
+
+    def setUp(self):
+        self._create_builder_thread()
+
+    def tearDown(self):
+        # It's the tests' responsibility to stop and join the builder
+        # thread if they start it.
+        self.assertFalse(self._builder_thread.isAlive())
+
+        self._master_sock.close()
+        self._builder_sock.close()
+
+    def test_shutdown(self):
+        """Tests that shutdown command exits the MemorySegmentBuilder
+        loop.
+        """
+
+        self._builder_thread.start()
+
+        # Now that the builder thread is running, send it the shutdown
+        # command. The thread should exit its main loop and be joinable.
+        with self._builder_cv:
+            with self._builder_lock:
+                self._builder_command_queue.append('shutdown')
+            self._builder_cv.notify_all()
+
+        # Wait 5 seconds at most for the main loop of the builder to
+        # exit.
+        self._builder_thread.join(5)
+        self.assertFalse(self._builder_thread.isAlive())
+
+        # The command queue must be cleared, and the response queue must
+        # be untouched (we don't use it in this test).
+        with self._builder_lock:
+            self.assertEqual(len(self._builder_command_queue), 0)
+            self.assertEqual(len(self._builder_response_queue), 0)
+
+if __name__ == "__main__":
+    isc.log.init("bind10-test")
+    isc.log.resetUnitTestRootLogger()
+    unittest.main()