Skip to content

Instantly share code, notes, and snippets.

@kevinlynx
Created March 3, 2015 06:49
Show Gist options
  • Save kevinlynx/c3c75040ac9abe3df0f7 to your computer and use it in GitHub Desktop.
Save kevinlynx/c3c75040ac9abe3df0f7 to your computer and use it in GitHub Desktop.
create a dummy tcp/http server in python to accept connections only, which can be used in unittest, to mock a tcp/http server
# dummy_httpd.py
# based on python 2.7
import BaseHTTPServer
from BaseHTTPServer import BaseHTTPRequestHandler
from SocketServer import ThreadingMixIn
import threading
import time
class ThreadedHTTPServer(ThreadingMixIn, BaseHTTPServer.HTTPServer):
daemon_threads = True
pass
class HTTPServerLayer():
class Handler(BaseHTTPRequestHandler):
def __init__(self, layer, *args):
self.layer = layer
# see StreamRequestHandler, `timeout' will set to `connection', set a timeout value
# to make the HTTP handle exits
self.timeout = 0.5
BaseHTTPRequestHandler.__init__(self, *args)
def do_GET(self):
self.send_response(self.layer.code)
self.end_headers()
self.wfile.write('\n')
def log_message(self, format, *args):
pass
def __init__(self, host = 'localhost', port = 8000):
self.host = host
self.port = port
self.code = 200
def start(self):
def handler(*args):
return HTTPServerLayer.Handler(self, *args)
self.server = ThreadedHTTPServer((self.host, self.port), handler)
self.server_thread = threading.Thread(target = self.server.serve_forever)
self.server_thread.setDaemon(True)
self.server_thread.start()
time.sleep(0.25)
def stop(self):
self.server.shutdown()
self.server.server_close()
self.server_thread.join()
def setcode(self, c = 200):
self.code = c
if __name__ == '__main__':
server = HTTPServerLayer()
server.start()
print('start done')
time.sleep(10)
server.setcode(404)
time.sleep(10)
print('to stop')
server.stop()
# dummy_tcpd.py
# based on python 2.7
import threading
import SocketServer
class TCPServerLayer:
class ThreadedTCPServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
allow_reuse_address = True
def __init__(self, port, host = 'localhost'):
self.port = port
self.host = host
def start(self):
self.server = TCPServerLayer.ThreadedTCPServer((self.host, self.port), SocketServer.BaseRequestHandler)
self.thread = threading.Thread(target = self.server.serve_forever)
self.thread.setDaemon(True)
self.thread.start()
def stop(self):
self.server.shutdown()
self.server.server_close()
self.thread.join()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment