mirror of https://gitee.com/anolis/sysom.git
80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
import time
|
|
import unittest
|
|
from cmg_base import dispatch_service_registry, CmgException, \
|
|
ServiceInstance, dispatch_service_discovery, LoadBalancingStrategy
|
|
|
|
CMG_URL = "redis://localhost:6379"
|
|
|
|
|
|
class MyTestCase(unittest.TestCase):
|
|
"""A test class to test service discovery
|
|
"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
pass
|
|
|
|
def setUp(self) -> None:
|
|
self.registry = dispatch_service_registry(CMG_URL)
|
|
self.discovery = dispatch_service_discovery(CMG_URL, fetch_interval=1)
|
|
self.test_service_name = "CMG_TEST_SERVICE_NAME"
|
|
self.test_service_name_not_exists = "CMG_TEST_SERVICE_NAME_NOT_EXISTS"
|
|
self.test_service_id = "CMG_TEST_SERVICE_ID"
|
|
self.test_service_id2 = "CMG_TEST_SERVICE_ID2"
|
|
self.test_service_id_not_exists = "CMG_TEST_SERVICE_ID_NOT_EXISTS"
|
|
self.registry.register(ServiceInstance(
|
|
service_name=self.test_service_name,
|
|
service_id=self.test_service_id
|
|
))
|
|
self.registry.register(ServiceInstance(
|
|
service_name=self.test_service_name,
|
|
service_id=self.test_service_id2
|
|
))
|
|
|
|
def tearDown(self) -> None:
|
|
self.registry.unregister(self.test_service_id)
|
|
self.registry.unregister(self.test_service_id2)
|
|
|
|
def test_get_services(self):
|
|
service_names = self.discovery.get_services()
|
|
self.assertIn(self.test_service_name, service_names)
|
|
|
|
def test_get_instances(self):
|
|
res = self.discovery.get_instances(self.test_service_name)
|
|
self.assertEqual(len(res), 2)
|
|
self.assertEqual(res[0].service_name, self.test_service_name)
|
|
self.assertEqual(res[1].service_name, self.test_service_name)
|
|
self.assertIn(
|
|
res[0].service_id, [self.test_service_id, self.test_service_id2]
|
|
)
|
|
self.assertIn(
|
|
res[1].service_id, [self.test_service_id, self.test_service_id2]
|
|
)
|
|
|
|
def test_get_not_exists_service_instances(self):
|
|
self.assertEqual(
|
|
len(self.discovery.get_instances(
|
|
self.test_service_name_not_exists)),
|
|
0
|
|
)
|
|
|
|
def test_get_instance(self):
|
|
id1_count = 0
|
|
id2_count = 0
|
|
for i in range(10):
|
|
time.sleep(1)
|
|
service_id = self.discovery.get_instance(
|
|
self.test_service_name,
|
|
LoadBalancingStrategy.RANDOM
|
|
).service_id
|
|
if service_id == self.test_service_id:
|
|
id1_count += 1
|
|
else:
|
|
id2_count += 1
|
|
self.assertNotEqual(id1_count, 0)
|
|
self.assertNotEqual(id2_count, 0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|