2018-12-07 20:44:31 +00:00
|
|
|
from functools import partial
|
2018-12-07 20:35:34 +00:00
|
|
|
from unittest import TestCase
|
2018-12-07 20:44:31 +00:00
|
|
|
|
2018-12-07 23:34:59 +00:00
|
|
|
from ipc_unix.simple import send_to
|
2018-12-07 23:18:29 +00:00
|
|
|
from tests import EchoServer, get_temp_file_path
|
2018-12-07 20:35:34 +00:00
|
|
|
|
|
|
|
|
2018-12-07 23:34:59 +00:00
|
|
|
class SimpleServerTestCase(TestCase):
|
2018-12-07 20:35:34 +00:00
|
|
|
def setUp(self):
|
2018-12-07 23:18:29 +00:00
|
|
|
self.socket_path = get_temp_file_path()
|
2018-12-07 20:35:34 +00:00
|
|
|
self.server = EchoServer(self.socket_path)
|
|
|
|
self.server.serve_in_thread()
|
2018-12-07 23:34:59 +00:00
|
|
|
self.send_to_client = partial(send_to, self.socket_path)
|
2018-12-07 20:35:34 +00:00
|
|
|
|
|
|
|
def tearDown(self):
|
|
|
|
self.server.shutdown()
|
|
|
|
|
|
|
|
def test_sending_dict(self):
|
|
|
|
data = {"foo": "bar"}
|
|
|
|
response = self.send_to_client(data)
|
|
|
|
self.assertEqual(response, data)
|
|
|
|
|
2018-12-07 23:18:29 +00:00
|
|
|
def test_sending_string(self):
|
|
|
|
data = "foo"
|
2018-12-07 20:35:34 +00:00
|
|
|
response = self.send_to_client(data)
|
|
|
|
self.assertEqual(response, data)
|
|
|
|
|
|
|
|
def test_sending_full_buffer(self):
|
|
|
|
data = ["foo"] * 4096 # Pad out the buffer
|
|
|
|
response = self.send_to_client(data)
|
|
|
|
self.assertEqual(response, data)
|
|
|
|
|
|
|
|
def test_sending_empty_payload(self):
|
|
|
|
response = self.send_to_client("")
|
|
|
|
self.assertEqual(response, "")
|
|
|
|
|
|
|
|
def test_multiple_send_to_same_server(self):
|
|
|
|
data = {"foo": "bar"}
|
|
|
|
for _ in range(10):
|
|
|
|
response = self.send_to_client(data)
|
|
|
|
self.assertEqual(response, data)
|