add albtest

albtest.py

@@ -0,0 +1,96 @@
+import unittest
+import utils
+from alibabacloud_alb20200616 import models as alb_models
+
+class TestAddServersToServerGroup(unittest.TestCase):
+    def setUp(self):
+        self.alb_client = FakeAlbClient()  # 使用假客户端
+        self.server_group_id = "sgp-h793418y"  # 假的服务器组 ID
+        self.instance_id = "i-123456"  # 假的实例 ID
+        self.weight = 10  # 权重值
+
+    def test_add_servers_to_server_group_success(self):
+        try:
+            utils.add_servers_to_server_group(self.alb_client, self.server_group_id, self.instance_id, self.weight)
+            print("Server added successfully.")
+        except Exception as e:
+            self.fail(f"add_servers_to_server_group raised an exception: {e}")
+
+
+
+class TestSetInstanceWeightProcessWithAlb(unittest.TestCase):
+    def setUp(self):
+        self.alb_client = FakeAlbClient()  # 使用假客户端
+        self.server_group_id = "sgp-h793418y"  # 假的服务器组 ID
+        self.instance_id = "i-123456"  # 假的实例 ID
+        self.weight = [(0, 1)]  # 假的权重值
+
+    def test_set_instance_weight_success(self):
+        try:
+            utils.set_instance_weight_process_with_alb(self.alb_client, self.server_group_id, self.instance_id,
+                                                       self.weight)
+            print("Instance weight set successfully.")
+        except Exception as e:
+            self.fail(f"set_instance_weight_process_with_alb raised an exception: {e}")
+
+
+
+class TestGetInstanceIds(unittest.TestCase):
+    def setUp(self):
+        self.alb_client = FakeAlbClient()  # 使用假客户端
+        self.server_group_id = "sgp-h793418y"  # 假的服务器组 ID
+    def test_get_instance_ids_success(self):
+        try:
+            instance_ids = utils.get_instance_ids(self.alb_client, self.server_group_id)
+            print(f"Retrieved instance IDs: {instance_ids}")
+            self.assertEqual(instance_ids, ["i-123456", "i-789012"])  # 假数据
+        except Exception as e:
+            self.fail(f"get_instance_ids raised an exception: {e}")
+
+
+
+
+class FakeAlbClient:
+    # 一个假 ALB 客户端，用于测试。
+    def add_servers_to_server_group_with_options(self, request, runtime):
+        if request.server_group_id == "sgp-h793418y" and request.servers:
+            return {"Code": "Success"}
+        else:
+            raise Exception("Failed to add server")
+
+    def add_servers_to_server_group_with_options(self, request, runtime):
+        if request.server_group_id == "sgp-h793418y" and request.servers:
+            return {"Code": "Success"}
+        else:
+            raise Exception("Failed to add server")
+
+    def set_instance_weight_with_options(self, request, runtime):
+        if request.server_group_id == "sgp-h793418y" and request.instance_id == "i-123456" and request.weight is not None:
+            return {"Code": "Success"}
+        else:
+            raise Exception("Failed to set instance weight")
+
+    def get_instance_ids_with_options(self, request, runtime):
+        print(f"Request to get instance IDs for group: {request.server_group_id}")
+        if request.server_group_id == "sgp-h793418y":
+            return {"InstanceIds": ["i-123456", "i-789012"]}
+        else:
+            raise Exception("Failed to get instance IDs")
+
+    def list_server_group_servers_with_options(self, request, runtime):
+        print(f"Request to list servers for group: {request.server_group_id}")
+        if request.server_group_id == "sgp-h793418y":
+            return {
+                "Servers": {
+                    "Server": [
+                        {"ServerId": "i-123456"},
+                        {"ServerId": "i-789012"}
+                    ]
+                }
+            }
+        else:
+            raise Exception("Failed to list server group servers")
+
+
+if __name__ == '__main__':
+    unittest.main()

utils.py

@@ -279,29 +279,29 @@ def get_ip_address(ecs_client, instance_id):
     return ip_address
 
 
-def set_weight_for_instances(client, slb_id, instance_id_list, weight):
-    """
-    同时设置多台服务器的slb权重，权重一样
-    :param client: 客户端连接
-    :param slb_id: slb_id
-    :param instance_id_list: 服务器id list
-    :param weight: 权重值
-    :return: None
-    """
-    for i in range(len(instance_id_list) // 20 + 1):
-        instances_list = instance_id_list[i * 20:(i + 1) * 20]
-        if len(instances_list) == 0:
-            return
-        BackendServers = [{"ServerId": instance_id, "Weight": weight} for instance_id in instances_list]
-        request = CommonRequest()
-        request.set_accept_format('json')
-        request.set_domain('slb.aliyuncs.com')
-        request.set_version('2014-05-15')
-        request.set_method('POST')
-        request.set_action_name('SetBackendServers')
-        request.add_query_param('BackendServers', BackendServers)
-        request.add_query_param('LoadBalancerId', slb_id)
-        response = send_request(client=client, request=request)
+# def set_weight_for_instances(client, slb_id, instance_id_list, weight):
+#     """
+#     同时设置多台服务器的slb权重，权重一样
+#     :param client: 客户端连接
+#     :param slb_id: slb_id
+#     :param instance_id_list: 服务器id list
+#     :param weight: 权重值
+#     :return: None
+#     """
+#     for i in range(len(instance_id_list) // 20 + 1):
+#         instances_list = instance_id_list[i * 20:(i + 1) * 20]
+#         if len(instances_list) == 0:
+#             return
+#         BackendServers = [{"ServerId": instance_id, "Weight": weight} for instance_id in instances_list]
+#         request = CommonRequest()
+#         request.set_accept_format('json')
+#         request.set_domain('slb.aliyuncs.com')
+#         request.set_version('2014-05-15')
+#         request.set_method('POST')
+#         request.set_action_name('SetBackendServers')
+#         request.add_query_param('BackendServers', BackendServers)
+#         request.add_query_param('LoadBalancerId', slb_id)
+#         response = send_request(client=client, request=request)
 
 
 def connect_alb_client(access_key_id, access_key_secret, endpoint):
@@ -467,80 +467,80 @@ def get_instances_status(ecs_client, instance_ids):
     return response
 
 
-def set_instance_weight_process(client, slb_id, instance_id_list, weight_list):
-    """
-    修改服务器的权重值
-    :param client: slb客户端连接·
-    :param slb_id: slb id
-    :param instance_id_list: instance id list
-    :param weight_list: 权重修改列表 type-list [(weight, sleep_time), ...]
-    :return:
-    """
-    for weight, sleep_time in weight_list:
-        logging.info(f"weight = {weight}")
-        flag = True
-        while flag:
-            try:
-                set_weight_for_instances(client=client, slb_id=slb_id, instance_id_list=instance_id_list, weight=weight)
-                time.sleep(sleep_time)
-                flag = False
-            except Exception as e:
-                time.sleep(10)
-                continue
-
-
-def add_backend_servers(client, slb_id, instances):
-    """
-    服务器挂载到负载均衡(必须是状态为运行中的后端服务器才可以加入负载均衡实例，每次调用最多可添加20个后端服务器)
-    :param client:
-    :param slb_id:
-    :param instances: 实例列表 [(instance_id, ip), ...]
-    :return:
-    """
-    try:
-        for i in range(len(instances) // 20 + 1):
-            instances_list = instances[i * 20:(i + 1) * 20]
-            if len(instances_list) == 0:
-                return
-            request = AddBackendServersRequest()
-            request.set_accept_format('json')
-            request.set_LoadBalancerId(slb_id)
-            backend_servers = [
-                {"ServerId": instance_id, "Weight": "0", "Type": "ecs", "ServerIp": ip_address}
-                for instance_id, ip_address in instances_list]
-            request.set_BackendServers(backend_servers)
-            response = client.do_action_with_exception(request)
-            return response
-    except Exception as e:
-        logging.error(e)
-        sys.exit()
-
-
-def remove_backend_servers(client, slb_id, instances):
-    """
-    服务器从负载均衡移除(一次调用最多可以移除20个后端服务器)
-    :param client:
-    :param slb_id:
-    :param instances: 实例列表 [instance_id, ...]
-    :return:
-    """
-    try:
-        for i in range(len(instances) // 20 + 1):
-            instances_list = instances[i * 20:(i + 1) * 20]
-            if len(instances_list) == 0:
-                return
-            request = RemoveBackendServersRequest()
-            request.set_accept_format('json')
-            request.set_LoadBalancerId(slb_id)
-            backend_servers = [
-                {"ServerId": instance_id, "Weight": "0", "Type": "ecs"}
-                for instance_id in instances_list]
-            request.set_BackendServers(backend_servers)
-            response = client.do_action_with_exception(request)
-            return response
-    except Exception as e:
-        logging.error(e)
-        sys.exit()
+# def set_instance_weight_process(client, slb_id, instance_id_list, weight_list):
+#     """
+#     修改服务器的权重值
+#     :param client: slb客户端连接·
+#     :param slb_id: slb id
+#     :param instance_id_list: instance id list
+#     :param weight_list: 权重修改列表 type-list [(weight, sleep_time), ...]
+#     :return:
+#     """
+#     for weight, sleep_time in weight_list:
+#         logging.info(f"weight = {weight}")
+#         flag = True
+#         while flag:
+#             try:
+#                 set_weight_for_instances(client=client, slb_id=slb_id, instance_id_list=instance_id_list, weight=weight)
+#                 time.sleep(sleep_time)
+#                 flag = False
+#             except Exception as e:
+#                 time.sleep(10)
+#                 continue
+
+
+# def add_backend_servers(client, slb_id, instances):
+#     """
+#     服务器挂载到负载均衡(必须是状态为运行中的后端服务器才可以加入负载均衡实例，每次调用最多可添加20个后端服务器)
+#     :param client:
+#     :param slb_id:
+#     :param instances: 实例列表 [(instance_id, ip), ...]
+#     :return:
+#     """
+#     try:
+#         for i in range(len(instances) // 20 + 1):
+#             instances_list = instances[i * 20:(i + 1) * 20]
+#             if len(instances_list) == 0:
+#                 return
+#             request = AddBackendServersRequest()
+#             request.set_accept_format('json')
+#             request.set_LoadBalancerId(slb_id)
+#             backend_servers = [
+#                 {"ServerId": instance_id, "Weight": "0", "Type": "ecs", "ServerIp": ip_address}
+#                 for instance_id, ip_address in instances_list]
+#             request.set_BackendServers(backend_servers)
+#             response = client.do_action_with_exception(request)
+#             return response
+#     except Exception as e:
+#         logging.error(e)
+#         sys.exit()
+
+
+# def remove_backend_servers(client, slb_id, instances):
+#     """
+#     服务器从负载均衡移除(一次调用最多可以移除20个后端服务器)
+#     :param client:
+#     :param slb_id:
+#     :param instances: 实例列表 [instance_id, ...]
+#     :return:
+#     """
+#     try:
+#         for i in range(len(instances) // 20 + 1):
+#             instances_list = instances[i * 20:(i + 1) * 20]
+#             if len(instances_list) == 0:
+#                 return
+#             request = RemoveBackendServersRequest()
+#             request.set_accept_format('json')
+#             request.set_LoadBalancerId(slb_id)
+#             backend_servers = [
+#                 {"ServerId": instance_id, "Weight": "0", "Type": "ecs"}
+#                 for instance_id in instances_list]
+#             request.set_BackendServers(backend_servers)
+#             response = client.do_action_with_exception(request)
+#             return response
+#     except Exception as e:
+#         logging.error(e)
+#         sys.exit()
 
 
 def set_weight_for_instances_alb(alb_client, server_group_id, instance_id_list, weights):