hurui200320
bde74c82b5
This MR fixed the pipeline so it can run on forgejo runner. Currently we have two pipelines set up: + One pipeline will run pytest to test the code and collect coverage data, this pipeline will run on all branches and PRs. + One pipeline will build the wheel file and publish it to the company's pypl repo, this pipeline will only run on the master branch and develop branch. This RP also made the following changes to make the pipeline working: + Remove `setup.py` and `requirements.txt`, now the project is fully embraced `pyproject.toml`, which according to ChatGPT, is the most modern way of managing a python project. + Move all test files to `tests` folder, so it won't be packed into the final wheel file and affect the coverage rate. + Change version to `X.Y.Z-dev`, before building the wheel file, the pipeline will automatically add a UTC timestamp after it, so it becomes `X.Y.Z-devYYYYMMDDHHmmSS`. This is added due to the limitation of the pypl repo, unlike maven snapshot versions, you can't push the same version twice with our pypl repository. Reviewed-on: #10 Reviewed-by: Stanislav Hejny <stanislav.hejny@cleverthis.com> Co-authored-by: Rui Hu <rui.hu@cleverthis.com> Co-committed-by: Rui Hu <rui.hu@cleverthis.com>
124 lines
5.0 KiB
Python
124 lines
5.0 KiB
Python
from typing import Set
|
|
from unittest import TestCase
|
|
|
|
import pytest
|
|
|
|
from amqp.adapter.data_message_factory import DataMessageFactory
|
|
from amqp.adapter.amq_route_factory import AMQRouteFactory
|
|
from amqp.model.model import AMQRoute
|
|
from amqp.router.router_base import RouterBase
|
|
from amqp.router.utils import sanitize_as_rabbitmq_name
|
|
|
|
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_6 = "amq::amq6:clever6.book.localhost.#:5:0"
|
|
ROUTE_7 = "amq::amq7:clever7.book.localhost.#:5:0"
|
|
ROUTE_8 = "amq::amq8:clever8.book.localhost.#:5:0"
|
|
ROUTE_9 = "amq::amq9:clever9.book.localhost.#:5:0"
|
|
|
|
|
|
class TestRouterBase(TestCase):
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_each_test(self):
|
|
self.base = RouterBase()
|
|
self.factory = DataMessageFactory.get_instance(111)
|
|
if len(self.base.get_routes()) == 0:
|
|
r1 = AMQRouteFactory.from_string(ROUTE_1)
|
|
r2 = AMQRouteFactory.from_string(ROUTE_2)
|
|
r3 = AMQRouteFactory.from_string(ROUTE_3)
|
|
self.base.add_route(r1)
|
|
self.base.add_route(r2)
|
|
self.base.add_route(r3)
|
|
|
|
def test_sanitize_as_rabbitmq_name(self):
|
|
self.assertEqual("aa.bb.cc", sanitize_as_rabbitmq_name("aa.bb/cc"))
|
|
self.assertEqual("aa.bb.cc", sanitize_as_rabbitmq_name("aa-bb/cc"))
|
|
self.assertEqual("aa", sanitize_as_rabbitmq_name("aa&bb/cc"))
|
|
self.assertEqual("aa", sanitize_as_rabbitmq_name("aa+bb/cc"))
|
|
self.assertEqual("aa", sanitize_as_rabbitmq_name("aa%bb/cc"))
|
|
self.assertEqual("aa#bb.cc", sanitize_as_rabbitmq_name("aa#bb/cc"))
|
|
self.assertEqual("aa#bb.cc.#", sanitize_as_rabbitmq_name("aa#bb/cc.#"))
|
|
self.assertEqual("aa", sanitize_as_rabbitmq_name("aa?bb////cc"))
|
|
self.assertEqual("aa.bb.cc", sanitize_as_rabbitmq_name("aa///bb////cc"))
|
|
|
|
def test_get_routing_key(self):
|
|
message = self.factory.create_request_message(
|
|
"POST", "test.me.localhost", "/aa/bb?cc=d", {}, ""
|
|
)
|
|
rk = self.base.get_routing_key(message)
|
|
self.assertEqual("test.me.localhost.aa.bb", rk)
|
|
|
|
def test_wrap_routes(self):
|
|
wrap = self.base.wrap_routes()
|
|
self.assertTrue(ROUTE_1 in wrap)
|
|
self.assertTrue(ROUTE_2 in wrap)
|
|
|
|
def test_wrap_own_routes(self):
|
|
_route: AMQRoute = AMQRouteFactory.from_string(ROUTE_7)
|
|
self.base.add_own_route(_route)
|
|
o_route = self.base.wrap_own_routes()
|
|
self.assertEqual(ROUTE_7, o_route)
|
|
|
|
def test_unwrap_route_list(self):
|
|
_routes = self.base.get_routes()
|
|
wrapped = self.base.wrap_routes()
|
|
unwrapped = self.base.unwrap_route_list(wrapped)
|
|
self.assertEqual(len(_routes), len(unwrapped))
|
|
self.assertTrue(unwrapped.pop() in _routes)
|
|
self.assertTrue(unwrapped.pop() in _routes)
|
|
|
|
def test_get_routes(self):
|
|
_routes = self.base.get_routes()
|
|
self.assertEqual(3, len(_routes))
|
|
|
|
def test_find_route(self):
|
|
_route = self.base.find_route("clever3.book.localhost", "/three-hot-dogs")
|
|
self.assertIsNotNone(_route)
|
|
self.assertEqual(_route.as_string(), ROUTE_3)
|
|
_route = self.base.find_route("clever-5.book.localhost", "/three-hot-dogs")
|
|
self.assertIsNone(_route)
|
|
|
|
def test_find_route_by_message(self):
|
|
message = DataMessageFactory.get_instance(111).create_request_message(
|
|
"POST", "clever3-book.localhost", "/aa/bb?cc=d", {}, ""
|
|
)
|
|
_route: AMQRoute = self.base.find_route_by_message(message)
|
|
self.assertIsNotNone(_route)
|
|
self.assertEqual(_route.as_string(), ROUTE_3)
|
|
|
|
def test_add_routes(self):
|
|
init_routes: Set[AMQRoute] = self.base.get_routes().copy()
|
|
self.base.add_routes(ROUTE_6 + ",amq-dlq:DLX:DeadLetterQueue:#:0:0,:::::,:::")
|
|
modified = self.base.get_routes()
|
|
self.assertEqual(len(init_routes) + 1, len(modified))
|
|
wr = self.base.wrap_routes()
|
|
self.assertFalse("DLX" in wr)
|
|
|
|
def test_remove_routes(self):
|
|
init_routes: Set[AMQRoute] = self.base.get_routes().copy()
|
|
toRemove: AMQRoute = init_routes.pop()
|
|
self.base.remove_routes(toRemove.as_string())
|
|
modified = self.base.get_routes()
|
|
self.assertEqual(len(init_routes), len(modified))
|
|
|
|
def test_remove_route(self):
|
|
origSize = len(self.base.get_routes())
|
|
_route = AMQRouteFactory.from_string(ROUTE_7)
|
|
self.base.add_route(_route)
|
|
self.assertEqual(origSize + 1, len(self.base.get_routes()))
|
|
self.base.remove_route(_route)
|
|
self.assertEqual(origSize, len(self.base.get_routes()))
|
|
|
|
def test_add_route(self):
|
|
origSize = len(self.base.get_routes())
|
|
_route = AMQRouteFactory.from_string(ROUTE_7)
|
|
self.base.add_route(_route)
|
|
self.assertEqual(origSize + 1, len(self.base.get_routes()))
|
|
|
|
def test_add_own_route(self):
|
|
_route = AMQRouteFactory.from_string(ROUTE_8)
|
|
self.base.add_own_route(_route)
|
|
self.assertTrue(ROUTE_8 in self.base.wrap_own_routes())
|