164 lines
6.1 KiB
Python
164 lines
6.1 KiB
Python
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()
|
|
|