Create a mock comment service that will listen for HTTP requests on a port

lms/djangoapps/django_comment_client/tests/mock_cs_server/mock_cs_server.py

+from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
+import json
+import urlparse
+from logging import getLogger
+logger = getLogger(__name__)
+
+
+class MockCommentServiceRequestHandler(BaseHTTPRequestHandler):
+    '''
+    A handler for Comment Service POST requests.
+    '''
+    protocol = "HTTP/1.0"
+
+    def do_POST(self):
+        '''
+        Handle a POST request from the client
+        Used by the APIs for comment threads, commentables, comments,
+        subscriptions, commentables, users
+        '''
+        # Retrieve the POST data into a dict.
+        # It should have been sent in json format
+        length = int(self.headers.getheader('content-length'))
+        data_string = self.rfile.read(length)
+        post_dict = json.loads(data_string)
+
+        # Log the request
+        logger.debug("Comment Service received POST request %s to path %s" %
+                    (json.dumps(post_dict), self.path))
+
+        # Every good post has at least an API key
+        if 'api_key' in post_dict:
+            response = self.server._response_str
+            # Log the response
+            logger.debug("Comment Service: sending response %s" % json.dumps(response))
+
+            # Send a response back to the client
+            self.send_response(200)
+            self.send_header('Content-type', 'application/json')
+            self.end_headers()
+            self.wfile.write(response)
+
+        else:
+            # Respond with failure
+            self.send_response(500, 'Bad Request: does not contain API key')
+            self.send_header('Content-type', 'text/plain')
+            self.end_headers()
+            return False
+
+
+class MockCommentServiceServer(HTTPServer):
+    '''
+    A mock Comment Service server that responds
+    to POST requests to localhost.
+    '''
+    def __init__(self, port_num,
+                 response={'username': 'new', 'external_id': 1}):
+        '''
+        Initialize the mock Comment Service server instance.
+        *port_num* is the localhost port to listen to
+        *response* is a dictionary that will be JSON-serialized
+            and sent in response to comment service requests.
+        '''
+        self._response_str = json.dumps(response)
+
+        handler = MockCommentServiceRequestHandler
+        address = ('', port_num)
+        HTTPServer.__init__(self, address, handler)
+
+    def shutdown(self):
+        '''
+        Stop the server and free up the port
+        '''
+        # First call superclass shutdown()
+        HTTPServer.shutdown(self)
+
+        # We also need to manually close the socket
+        self.socket.close()

lms/djangoapps/django_comment_client/tests/mock_cs_server/test_mock_cs_server.py

+import mock
+import unittest
+import threading
+import json
+import urllib
+import urllib2
+from mock_cs_server import MockCommentServiceServer, MockCommentServiceRequestHandler
+from nose.plugins.skip import SkipTest
+
+
+class MockCommentServiceServerTest(unittest.TestCase):
+    '''
+    A mock version of the Comment Service server that listens on a local
+    port and responds with pre-defined grade messages.
+    '''
+
+    def setUp(self):
+        # This is a test of the test setup,
+        # so it does not need to run as part of the unit test suite
+        # You can re-enable it by commenting out the line below
+        # raise SkipTest
+
+        # Create the server
+        server_port = 4567
+        self.server_url = 'http://127.0.0.1:%d' % server_port
+
+        # Start up the server and tell it that by default it should
+        # return this as its json response
+        self.expected_response = {'username': 'user100', 'external_id': '4'}
+        self.server = MockCommentServiceServer(port_num=server_port,
+                                               response=self.expected_response)
+
+        # Start the server in a separate daemon thread
+        server_thread = threading.Thread(target=self.server.serve_forever)
+        server_thread.daemon = True
+        server_thread.start()
+
+    def tearDown(self):
+        # Stop the server, freeing up the port
+        self.server.shutdown()
+
+    def test_new_user_request(self):
+        """
+        Test the mock comment service using an example
+        of how you would create a new user
+        """
+        # Send a request
+        values = {'username': u'user100', 'api_key': 'TEST_API_KEY',
+                  'external_id': '4', 'email': u'user100@edx.org'}
+        data = json.dumps(values)
+        headers = {'Content-Type': 'application/json', 'Content-Length': len(data)}
+        req = urllib2.Request(self.server_url + '/api/v1/users/4', data, headers)
+
+        # Send the request to the mock cs server
+        response = urllib2.urlopen(req)
+
+        # Receive the reply from the mock cs server
+        response_dict = json.loads(response.read())
+
+        # You should have received the response specified in the setup above
+        self.assertEqual(response_dict, self.expected_response)