Stanislav Hejny
105 lines
4.5 KiB
Python
105 lines
4.5 KiB
Python
from unittest import TestCase
|
|
|
|
import pytest
|
|
|
|
from amqp.adapter.amq_route_factory import AMQRouteFactory
|
|
from amqp.adapter.data_message_factory import DataMessageFactory
|
|
from amqp.model.model import AMQRoute
|
|
from amqp.router.route_database import RouteDatabase
|
|
|
|
ROUTE_1 = "amq::amq1:clever1.book.localhost.#:5:0"
|
|
ROUTE_2 = "amq::amq2:clever2.book.localhost.#:5:0"
|
|
ROUTE_3 = "amq::amq3:clever3.book.localhost.#:5:0"
|
|
ROUTE_7 = "amq::amq7:clever7.book.localhost.#:5:0"
|
|
|
|
|
|
class TestRouteDatabase(TestCase):
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_each_test(self):
|
|
self._database = RouteDatabase()
|
|
if len(self._database.get_routes()) == 0:
|
|
r1 = AMQRouteFactory.from_string(ROUTE_1)
|
|
r2 = AMQRouteFactory.from_string(ROUTE_2)
|
|
r3 = AMQRouteFactory.from_string(ROUTE_3)
|
|
self._database.add_route(r1)
|
|
self._database.add_route(r2)
|
|
self._database.add_route(r3)
|
|
|
|
def test_get_routes(self):
|
|
_amq_factory = AMQRouteFactory()
|
|
_route: AMQRoute = _amq_factory.from_string(
|
|
"amq-clever-pybook::clever-pybook:clever.pybook.localhost.#:5:0"
|
|
)
|
|
_routes = self._database.get_routes()
|
|
self.assertIsNotNone(_routes)
|
|
self.assertEqual(3, len(_routes))
|
|
self._database.add_route(_route)
|
|
_routes = self._database.get_routes()
|
|
self.assertTrue(_route in _routes, "Route was not added")
|
|
self.assertEqual(4, len(_routes))
|
|
|
|
def test_pattern(self):
|
|
pattern = self._database.pattern("")
|
|
self.assertEqual(".*", pattern.pattern)
|
|
pattern = self._database.pattern("..")
|
|
self.assertEqual("[^\\.]*\\.[^\\.]*\\.[^\\.]*", pattern.pattern)
|
|
pattern = self._database.pattern(".#.#")
|
|
self.assertEqual("[^\\.]*\\..*$", pattern.pattern)
|
|
pattern = self._database.pattern("..#")
|
|
self.assertEqual("[^\\.]*\\.[^\\.]*\\..*$", pattern.pattern)
|
|
pattern = self._database.pattern("#")
|
|
self.assertEqual(".*$", pattern.pattern)
|
|
pattern = self._database.pattern("ba.dc.")
|
|
self.assertEqual("ba\\.dc\\.[^\\.]*", pattern.pattern)
|
|
pattern = self._database.pattern(".cd.")
|
|
self.assertEqual("[^\\.]*\\.cd\\.[^\\.]*", pattern.pattern)
|
|
pattern = self._database.pattern("de")
|
|
self.assertEqual("de", pattern.pattern)
|
|
|
|
def test_matches(self):
|
|
self.assertTrue(self._database.matches("..", "aa.bb.cc"))
|
|
self.assertFalse(self._database.matches("..", "aa.bb."))
|
|
self.assertFalse(self._database.matches("..", "aa.bb"))
|
|
self.assertFalse(self._database.matches("..", "aa"))
|
|
self.assertTrue(self._database.matches("..#", "aa.bb.cc"))
|
|
self.assertFalse(self._database.matches("..#", "aa.bb."))
|
|
self.assertFalse(self._database.matches("..#", "aa.bb"))
|
|
self.assertFalse(self._database.matches("..#", "aa"))
|
|
self.assertTrue(self._database.matches(".#", "aa.bb.cc"))
|
|
self.assertTrue(self._database.matches(".#", "aa.bb."))
|
|
self.assertTrue(self._database.matches(".#", "aa.bb"))
|
|
self.assertFalse(self._database.matches(".#", "aa"))
|
|
|
|
def test_wrap_routes(self):
|
|
wrap = self._database.wrap_routes()
|
|
self.assertTrue(ROUTE_1 in wrap)
|
|
self.assertTrue(ROUTE_2 in wrap)
|
|
|
|
def test_find_route(self):
|
|
message = DataMessageFactory.get_instance(111).create_request_message(
|
|
"POST", "clever3-book.localhost", "/aa/bb?cc=d", {}, ""
|
|
)
|
|
route = self._database.find_route(message.domain(), message.path())
|
|
self.assertIsNotNone(route)
|
|
self.assertEqual(ROUTE_3, route.as_string())
|
|
route = self._database.find_route("clever3.book.localhost", "/three-hot-dogs")
|
|
self.assertIsNotNone(route)
|
|
self.assertEqual(ROUTE_3, route.as_string())
|
|
route = self._database.find_route("clever-5.book.localhost", "/three-hot-dogs")
|
|
self.assertIsNone(route)
|
|
|
|
def test_add_route(self):
|
|
origSize = len(self._database.get_routes())
|
|
route = AMQRouteFactory.from_string(ROUTE_7)
|
|
self._database.add_route(route)
|
|
self.assertEqual(origSize + 1, len(self._database.get_routes()))
|
|
|
|
def test_remove_route(self):
|
|
origSize = len(self._database.get_routes())
|
|
route = AMQRouteFactory.from_string(ROUTE_7)
|
|
self._database.add_route(route)
|
|
self.assertEqual(origSize + 1, len(self._database.get_routes()))
|
|
self._database.remove_route(route)
|
|
self.assertEqual(origSize, len(self._database.get_routes()))
|