[svn] commit: r3066 - in /branches/trac352/src/lib/python/isc/utils: serve_mixin.py tests/serve_mixin_test.py

[BIND 10 source code commits]

Author: zhanglikun
Date: Wed Sep 29 10:38:58 2010
New Revision: 3066

Log:
Add test code for class ServerMixIn.

Modified:
    branches/trac352/src/lib/python/isc/utils/serve_mixin.py
    branches/trac352/src/lib/python/isc/utils/tests/serve_mixin_test.py

Modified: branches/trac352/src/lib/python/isc/utils/serve_mixin.py
==============================================================================
--- branches/trac352/src/lib/python/isc/utils/serve_mixin.py (original)
+++ branches/trac352/src/lib/python/isc/utils/serve_mixin.py Wed Sep 29 10:38:58 2010
@@ -17,7 +17,7 @@
 import socket
 import select
 
-class ServeMixIn():
+class ServeMixIn:
     '''Mix-In class to override the function serve_forever()
     and shutdown() in class socketserver. 
     '''

Modified: branches/trac352/src/lib/python/isc/utils/tests/serve_mixin_test.py
==============================================================================
--- branches/trac352/src/lib/python/isc/utils/tests/serve_mixin_test.py (original)
+++ branches/trac352/src/lib/python/isc/utils/tests/serve_mixin_test.py Wed Sep 29 10:38:58 2010
@@ -14,13 +14,51 @@
 # WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 
 import unittest
-import isc.utils.serve_mixin
+from isc.utils.serve_mixin import ServeMixIn
+import socketserver
+import threading
+import socket
+import time
+
+class MyHandler(socketserver.BaseRequestHandler):
+    def handle(self):
+        data = self.request.recv(20)
+        self.request.send(data)
+
+class MyServer(ServeMixIn, 
+               socketserver.ThreadingMixIn,
+               socketserver.TCPServer):
+    pass
+
+
+def send_and_get_reply(ip, port, msg):
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.connect((ip, port))
+    sock.send(msg)
+    response = sock.recv(20)
+    sock.close()
+    return response
 
 class TestServeMixIn(unittest.TestCase):
     def test_serve_forever(self):
-        pass
+        # use port 0 to select an arbitrary unused port.
+        server = MyServer(('localhost', 0), MyHandler)
+        ip, port = server.server_address
+        server_thread = threading.Thread(target=server.serve_forever)
+        server_thread.setDaemon(True)
+        server_thread.start()
 
-   if __name__== "__main__":
+        msg = b'senddata'
+        self.assertEqual(msg, send_and_get_reply(ip, port, msg))
+        self.assertTrue(server_thread.is_alive())
+
+        # Now shutdown the server
+        server.shutdown()
+        # Sleep a while, make sure the thread has finished.
+        time.sleep(0.1)
+        self.assertFalse(server_thread.is_alive())
+
+if __name__== "__main__":
     unittest.main()