diff options
Diffstat (limited to 'yaksh/tests')
| -rw-r--r-- | yaksh/tests/test_code_server.py | 131 |
1 files changed, 67 insertions, 64 deletions
diff --git a/yaksh/tests/test_code_server.py b/yaksh/tests/test_code_server.py index 47c1da7..a73c12f 100644 --- a/yaksh/tests/test_code_server.py +++ b/yaksh/tests/test_code_server.py @@ -8,9 +8,8 @@ from threading import Thread import unittest from six.moves import urllib -from yaksh.code_server import ServerPool, SERVER_POOL_PORT +from yaksh.code_server import ServerPool, SERVER_POOL_PORT, submit, get_result from yaksh import settings -from yaksh.xmlrpc_clients import CodeServerProxy class TestCodeServer(unittest.TestCase): @@ -19,8 +18,7 @@ class TestCodeServer(unittest.TestCase): def setUpClass(cls): settings.code_evaluators['python']['standardtestcase'] = \ "yaksh.python_assertion_evaluator.PythonAssertionEvaluator" - ports = range(8001, 8006) - server_pool = ServerPool(ports=ports, pool_port=SERVER_POOL_PORT) + server_pool = ServerPool(n=5, pool_port=SERVER_POOL_PORT) cls.server_pool = server_pool cls.server_thread = t = Thread(target=server_pool.run) t.start() @@ -33,70 +31,78 @@ class TestCodeServer(unittest.TestCase): "python_assertion_evaluator.PythonAssertionEvaluator" def setUp(self): - self.code_server = CodeServerProxy() + self.url = 'http://localhost:%s' % SERVER_POOL_PORT def test_infinite_loop(self): # Given - testdata = {'metadata': {'user_answer': 'while True: pass', - 'language': 'python', - 'partial_grading': False - }, - 'test_case_data': [{'test_case':'assert 1==2', - 'test_case_type': 'standardtestcase', - 'weight': 0.0 - }] - } + testdata = { + 'metadata': { + 'user_answer': 'while True: pass', + 'language': 'python', + 'partial_grading': False + }, + 'test_case_data': [ + {'test_case': 'assert 1==2', + 'test_case_type': 'standardtestcase', + 'weight': 0.0} + ] + } # When - result = self.code_server.run_code( - 'python', json.dumps(testdata), '' - ) + submit(self.url, '0', json.dumps(testdata), '') + result = get_result(self.url, '0') # Then - data = json.loads(result) + self.assertTrue(result.get('status') in ['running', 'not started']) + + # When + result = get_result(self.url, '0', block=True) + + # Then + data = json.loads(result.get('result')) self.assertFalse(data['success']) self.assertTrue('infinite loop' in data['error'][0]) def test_correct_answer(self): # Given - testdata = {'metadata': { 'user_answer': 'def f(): return 1', - 'language': 'python', - 'partial_grading': False - }, - 'test_case_data': [{'test_case':'assert f() == 1', - 'test_case_type': 'standardtestcase', - 'weight': 0.0 - }] - } + testdata = { + 'metadata': { + 'user_answer': 'def f(): return 1', + 'language': 'python', + 'partial_grading': False + }, + 'test_case_data': [{'test_case': 'assert f() == 1', + 'test_case_type': 'standardtestcase', + 'weight': 0.0}] + } # When - result = self.code_server.run_code( - 'python', json.dumps(testdata), '' - ) + submit(self.url, '0', json.dumps(testdata), '') + result = get_result(self.url, '0', block=True) # Then - data = json.loads(result) + data = json.loads(result.get('result')) self.assertTrue(data['success']) def test_wrong_answer(self): # Given - testdata = {'metadata': { 'user_answer': 'def f(): return 1', - 'language': 'python', - 'partial_grading': False - }, - 'test_case_data': [{'test_case':'assert f() == 2', - 'test_case_type': 'standardtestcase', - 'weight': 0.0 - }] - } + testdata = { + 'metadata': { + 'user_answer': 'def f(): return 1', + 'language': 'python', + 'partial_grading': False + }, + 'test_case_data': [{'test_case': 'assert f() == 2', + 'test_case_type': 'standardtestcase', + 'weight': 0.0}] + } # When - result = self.code_server.run_code( - 'python', json.dumps(testdata), '' - ) + submit(self.url, '0', json.dumps(testdata), '') + result = get_result(self.url, '0', block=True) # Then - data = json.loads(result) + data = json.loads(result.get('result')) self.assertFalse(data['success']) self.assertTrue('AssertionError' in data['error'][0]) @@ -104,28 +110,27 @@ class TestCodeServer(unittest.TestCase): # Given results = Queue() - def run_code(): + def run_code(uid): """Run an infinite loop.""" - testdata = {'metadata': { 'user_answer': 'while True: pass', - 'language': 'python', - 'partial_grading': False - }, - 'test_case_data': [{'test_case':'assert 1==2', - 'test_case_type': 'standardtestcase', - 'weight': 0.0 - }] - } - result = self.code_server.run_code( - 'python', json.dumps(testdata), '' - ) - results.put(json.loads(result)) + testdata = { + 'metadata': { + 'user_answer': 'while True: pass', + 'language': 'python', + 'partial_grading': False + }, + 'test_case_data': [{'test_case': 'assert 1==2', + 'test_case_type': 'standardtestcase', + 'weight': 0.0}] + } + submit(self.url, uid, json.dumps(testdata), '') + result = get_result(self.url, uid, block=True) + results.put(json.loads(result.get('result'))) N = 10 # When - import time threads = [] for i in range(N): - t = Thread(target=run_code) + t = Thread(target=run_code, args=(str(i),)) threads.append(t) t.start() @@ -142,16 +147,14 @@ class TestCodeServer(unittest.TestCase): def test_server_pool_status(self): # Given - url = "http://localhost:%s/status"%SERVER_POOL_PORT + url = "http://localhost:%s/" % SERVER_POOL_PORT # When response = urllib.request.urlopen(url) data = response.read().decode('utf-8') # Then - expect = 'out of 5 are free' - self.assertTrue(expect in data) - expect = 'Load:' + expect = '5 processes, 0 running, 0 queued' self.assertTrue(expect in data) |