diff --git a/common/djangoapps/terrain/mock_xqueue_server.py b/common/djangoapps/terrain/mock_xqueue_server.py new file mode 100644 index 0000000000..50d77a2f19 --- /dev/null +++ b/common/djangoapps/terrain/mock_xqueue_server.py @@ -0,0 +1,227 @@ +from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler +import json +import urllib +import urlparse + + +class MockXQueueRequestHandler(BaseHTTPRequestHandler): + ''' + A handler for XQueue POST requests. + ''' + + protocol = "HTTP/1.0" + + def do_HEAD(self): + self._send_head() + + def do_POST(self): + ''' + Handle a POST request from the client, interpreted + as either a login request or a submission for grading request. + + Sends back an immediate success/failure response. + If grading is required, it then POSTS back to the client + with grading results, as configured in MockXQueueServer. + ''' + self._send_head() + + # Retrieve the POST data + post_dict = self._post_dict() + + # Send a response indicating success/failure + success = self._send_immediate_response(post_dict) + + # If the client submitted a valid submission request, + # we need to post back to the callback url + # with the grading result + if success and self._is_grade_request(): + self._send_grade_response(post_dict['lms_callback_url'], + post_dict['lms_key']) + + def _send_head(self): + ''' + Send the response code and MIME headers + ''' + if self._is_login_request() or self._is_grade_request(): + self.send_response(200) + else: + self.send_response(500) + + self.send_header('Content-type', 'text/plain') + self.end_headers() + + def _post_dict(self): + ''' + Retrieve the POST parameters from the client as a dictionary + ''' + + try: + length = int(self.headers.getheader('content-length')) + + post_dict = urlparse.parse_qs(self.rfile.read(length)) + + # The POST dict will contain a list of values + # for each key. + # None of our parameters are lists, however, + # so we map [val] --> val + # If the list contains multiple entries, + # we pick the first one + post_dict = dict(map(lambda (key, list_val): (key, list_val[0]), + post_dict.items())) + + except: + # We return an empty dict here, on the assumption + # that when we later check that the request has + # the correct fields, it won't find them, + # and will therefore send an error response + return {} + + return post_dict + + def _send_immediate_response(self, post_dict): + ''' + Check the post_dict for the appropriate fields + for this request (login or grade submission) + If it finds them, inform the client of success. + Otherwise, inform the client of failure + ''' + + # Allow any user to log in, as long as the POST + # dict has a username and password + if self._is_login_request(): + success = 'username' in post_dict and 'password' in post_dict + + elif self._is_grade_request(): + success = ('lms_callback_url' in post_dict and + 'lms_key' in post_dict and + 'queue_name' in post_dict) + else: + success = False + + # Send the response indicating success/failure + response_str = json.dumps({'return_code': 0 if success else 1, + 'content': '' if success else 'Error'}) + + self.wfile.write(response_str) + + return success + + def _send_grade_response(self, postback_url, queuekey): + ''' + POST the grade response back to the client + using the response provided by the server configuration + ''' + response_dict = {'queuekey': queuekey, + 'xqueue_body': self.server.grade_response} + + MockXQueueRequestHandler.post_to_url(postback_url, response_dict) + + def _is_login_request(self): + return 'xqueue/login' in self.path + + def _is_grade_request(self): + return 'xqueue/submit' in self.path + + @staticmethod + def post_to_url(url, param_dict): + ''' + POST *param_dict* to *url* + We make this a separate function so we can easily patch + it during testing. + ''' + urllib.urlopen(url, urllib.urlencode(param_dict)) + + +class MockXQueueServer(HTTPServer): + ''' + A mock XQueue grading server that responds + to POST requests to localhost. + ''' + + def __init__(self, port_num, grade_response_dict): + ''' + Initialize the mock XQueue server instance. + + *port_num* is the localhost port to listen to + + *grade_response_dict* is a dictionary that will be JSON-serialized + and sent in response to XQueue grading requests. + ''' + + self.grade_response = grade_response_dict + + handler = MockXQueueRequestHandler + address = ('', port_num) + HTTPServer.__init__(self, address, handler) + + @property + def grade_response(self): + return self._grade_response + + @grade_response.setter + def grade_response(self, grade_response_dict): + self._grade_response = grade_response_dict + + +# ---------------------------- +# Tests + +import mock +import threading +import unittest + + +class MockXQueueServerTest(unittest.TestCase): + + def setUp(self): + + # Create the server + server_port = 8034 + self.server_url = 'http://127.0.0.1:%d' % server_port + self.server = MockXQueueServer(server_port, + {'correct': True, 'score': 1, 'msg': ''}) + + # 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() + self.server.socket.close() + + def test_login_request(self): + + # Send a login request + login_request = {'username': 'Test', 'password': 'Test'} + response_handle = urllib.urlopen(self.server_url + '/xqueue/login', + urllib.urlencode(login_request)) + response_dict = json.loads(response_handle.read()) + self.assertEqual(response_dict['return_code'], 0) + + def test_grade_request(self): + + # Patch post_to_url() so we can intercept + # outgoing POST requests from the server + MockXQueueRequestHandler.post_to_url = mock.Mock() + + # Send a grade request + callback_url = 'http://127.0.0.1:8000/test_callback' + grade_request = {'lms_callback_url': callback_url, + 'lms_key': 'test_queuekey', + 'queue_name': 'test_queue'} + response_handle = urllib.urlopen(self.server_url + '/xqueue/submit', + urllib.urlencode(grade_request)) + response_dict = json.loads(response_handle.read()) + + # Expect that the response is success + self.assertEqual(response_dict['return_code'], 0) + + # Expect that the server tries to post back the grading info + expected_callback_dict = {'queuekey': 'test_queuekey', + 'xqueue_body': {'correct': True, + 'score': 1, 'msg': ''}} + MockXQueueRequestHandler.post_to_url.assert_called_with(callback_url, + expected_callback_dict)