Implemented dummy socket server interface and added one test

dummyserver/server.py

@@ -0,0 +1,38 @@
+#!/usr/bin/python
+
+import threading, socket
+
+
+
+"""
+Dummy server using for unit testing
+"""
+
+class Server(threading.Thread):
+    def __init__(self, handler, host='localhost', port=8021):
+        threading.Thread.__init__(self)
+        self.handler = handler
+        self.host = host
+        self.port = port
+        self.ready_event = threading.Event()
+        self.stop_event = threading.Event()
+
+    def run(self):
+        sock = socket.socket()
+        sock.bind((self.host, self.port))
+        sock.listen(0)
+        self.ready_event.set()
+        self.handler(sock)
+        self.stop_event.set()
+        sock.close()
+
+    def __enter__(self):
+       self.start()
+       self.ready_event.wait()
+       return self.host, self.port
+      
+    def __exit__(self, exc_type, exc_value, traceback):
+        if exc_type is None:
+            self.stop_event.wait()
+        return False # allow exceptions to propagate 
+    

test_requests.py

@@ -29,6 +29,8 @@ from requests.structures import CaseInsensitiveDict
 from requests.sessions import SessionRedirectMixin
 from requests.models import urlencode
 from requests.hooks import default_hooks
+from dummyserver.server import Server
+import socket
 
 try:
     import StringIO
@@ -1736,6 +1738,23 @@ def test_vendor_aliases():
     with pytest.raises(ImportError):
         from requests.packages import webbrowser
 
+class TestDummyServer(unittest.TestCase):
+    def test_basic(self):
+        question = "sucess?"
+        answer = "yeah, success"
+        def handler(server_sock):
+            sock, _ = server_sock.accept()
+            text = sock.recv(1000)
+            assert text == question 
+            sock.send(answer)
+        
+        with Server(handler) as (host, port):
+            sock = socket.socket()
+            sock.connect((host, port))
+            sock.send(question)
+            text = sock.recv(1000)
+            assert text == answer
+            sock.close()
 
 if __name__ == '__main__':
     unittest.main()