diff options
Diffstat (limited to 'tests.py')
| -rw-r--r-- | tests.py | 141 |
1 files changed, 111 insertions, 30 deletions
@@ -36,6 +36,9 @@ from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCRequestHandler +ORIGINAL_HISTORY_SIZE = history.size + + def get_port(family=socket.AF_INET): sock = socket.socket(family, socket.SOCK_STREAM) sock.bind(("localhost", 0)) @@ -53,6 +56,10 @@ class TestCompatibility(unittest.TestCase): self.server = server_set_up(addr=('', self.port)) self.client = Server('http://localhost:%d' % self.port) + def tearDown(self): + self.server.stop() + self.server.join() + # v1 tests forthcoming # Version 2.0 Tests @@ -281,6 +288,13 @@ class InternalTests(unittest.TestCase): def setUp(self): self.port = get_port() self.server = server_set_up(addr=('', self.port)) + self.addCleanup(self.cleanup) + + def cleanup(self): + self.server.stop() + self.server.join() + history.size = ORIGINAL_HISTORY_SIZE + history.clear() def get_client(self): return Server('http://localhost:%d' % self.port) @@ -316,7 +330,10 @@ class InternalTests(unittest.TestCase): def test_single_namespace(self): client = self.get_client() + response = client.namespace.sum(1, 2, 4) + self.assertEqual(7, response) + request = json.loads(history.request) response = json.loads(history.response) verify_request = { @@ -328,8 +345,64 @@ class InternalTests(unittest.TestCase): } verify_request['id'] = request['id'] verify_response['id'] = request['id'] - self.assertTrue(verify_request == request) - self.assertTrue(verify_response == response) + self.assertEqual(verify_request, request) + self.assertEqual(verify_response, response) + + def test_history_defaults_to_20(self): + client = self.get_client() + self.assertEqual(20, history.size) + + for i in range(30): + client.namespace.sum(i, i) + + self.assertEqual(20, len(history.requests)) + self.assertEqual(20, len(history.responses)) + + verify_request = { + "jsonrpc": "2.0", "params": [29, 29], + "method": "namespace.sum" + } + verify_response = {"jsonrpc": "2.0", "result": 58} + + # it should truncate to the *last* 20 + request = json.loads(history.request) + response = json.loads(history.response) + + verify_request["id"] = request["id"] + self.assertEqual(request, verify_request) + + verify_response["id"] = request["id"] + self.assertEqual(response, verify_response) + + def test_history_allows_configurable_size(self): + client = self.get_client() + history.size = 10 + + for i in range(30): + client.namespace.sum(i, i) + + self.assertEqual(10, len(history.requests)) + self.assertEqual(10, len(history.responses)) + + def test_history_allows_unlimited_size(self): + client = self.get_client() + history.size = -1 + + for i in range(40): + client.namespace.sum(i, i) + + self.assertEqual(40, len(history.requests)) + self.assertEqual(40, len(history.responses)) + + def test_history_can_be_disabled(self): + client = self.get_client() + history.size = 0 + + for i in range(40): + client.namespace.sum(i, i) + + self.assertEqual(0, len(history.requests)) + self.assertEqual(0, len(history.responses)) def test_multicall_success(self): multicall = self.get_multicall_client() @@ -365,37 +438,40 @@ class InternalTests(unittest.TestCase): result = sub_service_proxy.add(21, 21) self.assertTrue(result == 42) -if jsonrpc.USE_UNIX_SOCKETS: - # We won't do these tests unless Unix Sockets are supported - - @unittest.skip("Skipping Unix socket tests right now.") - class UnixSocketInternalTests(InternalTests): - """ - These tests run the same internal communication tests, - but over a Unix socket instead of a TCP socket. - """ - def setUp(self): - suffix = "%d.sock" % get_port() - - # Open to safer, alternative processes - # for getting a temp file name... - temp = tempfile.NamedTemporaryFile( - suffix=suffix - ) - self.port = temp.name - temp.close() - self.server = server_set_up( - addr=self.port, - address_family=socket.AF_UNIX - ) +@unittest.skipIf( + not jsonrpc.USE_UNIX_SOCKETS or "SKIP_UNIX_SOCKET_TESTS" in os.environ, + "Skipping Unix socket tests -- unsupported in this environment.") +class UnixSocketInternalTests(InternalTests): + """ + These tests run the same internal communication tests, + but over a Unix socket instead of a TCP socket. + """ + def setUp(self): + super().setUp() + suffix = "%d.sock" % get_port() - def get_client(self): - return Server('unix:/%s' % self.port) + # Open to safer, alternative processes + # for getting a temp file name... + temp = tempfile.NamedTemporaryFile( + suffix=suffix + ) + self.port = temp.name + temp.close() - def tearDown(self): - """ Removes the tempory socket file """ - os.unlink(self.port) + self.server = server_set_up( + addr=self.port, + address_family=socket.AF_UNIX + ) + + def get_client(self): + return Server('unix:/%s' % self.port) + + def tearDown(self): + """ Removes the tempory socket file """ + self.server.stop() + self.server.join() + os.unlink(self.port) class UnixSocketErrorTests(unittest.TestCase): @@ -473,7 +549,12 @@ def server_set_up(addr, address_family=socket.AF_INET): server.register_function(service.summation, 'sum') server.register_function(service.summation, 'notify_sum') server.register_function(service.summation, 'namespace.sum') + + def stop(): + server.shutdown() + server_proc = Thread(target=server.serve_forever) server_proc.daemon = True + server_proc.stop = stop server_proc.start() return server_proc |