diff --git a/lms/djangoapps/django_comment_client/tests/mock_cs_server/__init__.py b/lms/djangoapps/django_comment_client/tests/mock_cs_server/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/lms/djangoapps/django_comment_client/tests/mock_cs_server/mock_cs_server.py b/lms/djangoapps/django_comment_client/tests/mock_cs_server/mock_cs_server.py new file mode 100644 index 0000000000..6fbc88fb31 --- /dev/null +++ b/lms/djangoapps/django_comment_client/tests/mock_cs_server/mock_cs_server.py @@ -0,0 +1,76 @@ +from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler +import json +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() diff --git a/lms/djangoapps/django_comment_client/tests/mock_cs_server/test_mock_cs_server.py b/lms/djangoapps/django_comment_client/tests/mock_cs_server/test_mock_cs_server.py new file mode 100644 index 0000000000..9b1977cbff --- /dev/null +++ b/lms/djangoapps/django_comment_client/tests/mock_cs_server/test_mock_cs_server.py @@ -0,0 +1,59 @@ +import unittest +import threading +import json +import urllib2 +from mock_cs_server import MockCommentServiceServer +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)