Implemented a dummy http server to run unit-test on a offline computer.

Reviewed: Marcelo Lira <marcelo.lira@openbossa.org>
          Luciano Wolf <luciano.wolf@openbossa.org>

1 files changed, 71 insertions, 0 deletions

diff --git a/tests/util/httpd.py b/tests/util/httpd.py
new file mode 100644
index 000000000..96750f01c
--- /dev/null
+++ b/tests/util/httpd.py
@@ -0,0 +1,71 @@
+import SocketServer
+import BaseHTTPServer
+import random
+
+from threading import Thread
+
+
+class TestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
+    DATA = "PySide Server"
+
+    def do_GET(self):
+        self.send_head()
+        self.wfile.write(TestHandler.DATA)
+
+    def do_HEAD(self):
+        self.send_head()
+
+    def send_head(self):
+        self.send_response(200)
+        self.send_header("Content-type", "text/plain")
+        self.send_header("Content-Length", str(len(TestHandler.DATA)))
+        self.end_headers()
+
+class TestSecureHandler(BaseHTTPServer.BaseHTTPRequestHandler):
+    DATA = "PySide"
+
+    def do_GET(self):
+        self.send_head()
+        self.wfile.write(TestHandler.DATA)
+
+    def do_HEAD(self):
+        self.send_head()
+
+    def send_head(self):
+        try:
+            handler = self.marshall_handler()
+            handler.do_request(self)
+        except:
+            self.send_response(401)
+            self.send_header("WWW-Authenticate", "Basic realm='Secure Area'")
+            self.send_header("Content-type", "text/plain")
+            self.send_header("Content-Length", str(len(TestHandler.DATA)))
+            self.end_headers()
+
+
+class TestServer(Thread):
+
+    def __init__(self, secure=False):
+        Thread.__init__(self)
+
+        self._port = 8000 + random.randint(0, 100)
+        self.keep_running = True
+        server = SocketServer.TCPServer
+
+        if secure:
+            handle = TestSecureHandler
+        else:
+            handle = TestHandler
+
+        self.httpd = SocketServer.TCPServer((''  , self._port), handle)
+
+    def port(self):
+        return self._port
+
+    def run(self):
+        self.httpd.serve_forever()
+
+    def shutdown(self):
+        self.httpd.shutdown()
+        self.join()
+