from queue import Queue import random import threading import unittest from coapclient import HelperClient from coapserver import CoAPServer from coapthon import defines from coapthon.messages.option import Option from coapthon.messages.request import Request from coapthon.messages.response import Response __author__ = 'Giacomo Tanganelli' __version__ = "2.0" class Tests(unittest.TestCase): def setUp(self): self.server_address = ("::1", 5683) self.current_mid = random.randint(1, 1000) self.server_mid = random.randint(1000, 2000) self.server = CoAPServer("::1", 5683) self.server_thread = threading.Thread(target=self.server.listen, args=(10,)) self.server_thread.start() self.queue = Queue() def tearDown(self): self.server.close() self.server_thread.join(timeout=25) self.server = None def _test_with_client(self, message_list): # pragma: no cover client = HelperClient(self.server_address) for message, expected in message_list: if message is not None: received_message = client.send_request(message) if expected is not None: if expected.type is not None: self.assertEqual(received_message.type, expected.type) if expected.mid is not None: self.assertEqual(received_message.mid, expected.mid) self.assertEqual(received_message.code, expected.code) if expected.source is not None: self.assertEqual(received_message.source, self.server_address) if expected.token is not None: self.assertEqual(received_message.token, expected.token) if expected.payload is not None: self.assertEqual(received_message.payload, expected.payload) if expected.options: self.assertEqual(len(received_message.options), len(expected.options)) for o in expected.options: assert isinstance(o, Option) option_value = getattr(expected, o.name.lower().replace("-", "_")) option_value_rec = getattr(received_message, o.name.lower().replace("-", "_")) self.assertEqual(option_value, option_value_rec) client.stop() def _test_with_client_observe(self, message_list): # pragma: no cover client = HelperClient(self.server_address) for message, expected in message_list: if message is not None: client.send_request(message, self.client_callback) if expected is not None: received_message = self.queue.get() if expected.type is not None: self.assertEqual(received_message.type, expected.type) if expected.mid is not None: self.assertEqual(received_message.mid, expected.mid) self.assertEqual(received_message.code, expected.code) if expected.source is not None: self.assertEqual(received_message.source, self.server_address) if expected.token is not None: self.assertEqual(received_message.token, expected.token) if expected.payload is not None: self.assertEqual(received_message.payload, expected.payload) if expected.options: self.assertEqual(len(received_message.options), len(expected.options)) for o in expected.options: assert isinstance(o, Option) option_value = getattr(expected, o.name.lower().replace("-", "_")) option_value_rec = getattr(received_message, o.name.lower().replace("-", "_")) self.assertEqual(option_value, option_value_rec) client.stop() def client_callback(self, response): print("Callback") self.queue.put(response) def test_not_allowed(self): print("TEST_NOT_ALLOWED") path = "/void" req = Request() req.code = defines.Codes.GET.number req.uri_path = path req.type = defines.Types["CON"] req._mid = self.current_mid req.destination = self.server_address expected = Response() expected.type = defines.Types["ACK"] expected._mid = self.current_mid expected.code = defines.Codes.METHOD_NOT_ALLOWED.number expected.token = None exchange1 = (req, expected) self.current_mid += 1 req = Request() req.code = defines.Codes.POST.number req.uri_path = path req.type = defines.Types["CON"] req._mid = self.current_mid req.destination = self.server_address expected = Response() expected.type = defines.Types["ACK"] expected._mid = self.current_mid expected.code = defines.Codes.METHOD_NOT_ALLOWED.number expected.token = None exchange2 = (req, expected) self.current_mid += 1 req = Request() req.code = defines.Codes.PUT.number req.uri_path = path req.type = defines.Types["CON"] req._mid = self.current_mid req.destination = self.server_address expected = Response() expected.type = defines.Types["ACK"] expected._mid = self.current_mid expected.code = defines.Codes.METHOD_NOT_ALLOWED.number expected.token = None exchange3 = (req, expected) self.current_mid += 1 req = Request() req.code = defines.Codes.DELETE.number req.uri_path = path req.type = defines.Types["CON"] req._mid = self.current_mid req.destination = self.server_address expected = Response() expected.type = defines.Types["ACK"] expected._mid = self.current_mid expected.code = defines.Codes.METHOD_NOT_ALLOWED.number expected.token = None exchange4 = (req, expected) self.current_mid += 1 self._test_with_client([exchange1, exchange2, exchange3, exchange4]) if __name__ == '__main__': unittest.main()