add albtest.py

albtest.py

@@ -1,96 +1,74 @@
 import unittest
+import os
+import sys
+import logging
 import utils
+from aliyunsdkcore.client import AcsClient
+from alibabacloud_alb20200616.client import Client as Alb20200616Client
+from alibabacloud_tea_openapi import models as open_api_models
 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")
-
+from alibabacloud_alb20200616 import models as alb_20200616_models
+
+# 配置客户端参数
+alb_client_params = {
+    'access_key_id': 'LTAI5tASD5yEZLeC8ffmNebY',
+    'access_key_secret': '1PtsFRdp8viJmI78lEhNZR8MezWZBq',
+    # 'endpoint': 'alb-vpc.cn-hangzhou.aliyuncs.com',
+    'endpoint': 'alb.cn-hangzhou.aliyuncs.com',    #  外网调试
+    'region_id': 'cn-hangzhou'
+}
+
+ecs_client_params = {
+    'access_key_id': 'LTAI4GBWbFvvXoXsSVBe1o9f',
+    'access_key_secret': 'kRAikWitb4kDxaAyBqNrmLmllMEDO3',
+    'region_id': 'cn-hangzhou'
+}
+
+server_group_id = "sgp-ec4gopoclruofsfmxu"  # 测试用的服务器组
+instance_id = "i-bp19n839usecekzn2ig9"  # PaddlePaddle_k8s集群登陆实例，不影响线上服务
+
+
+def connect_client(access_key_id, access_key_secret, region_id):
+    try:
+        clt = AcsClient(ak=access_key_id, secret=access_key_secret, region_id=region_id)
+        return clt
+    except Exception as e:
+        logging.error(e)
+        sys.exit()
+
+def connect_alb_client(access_key_id, access_key_secret, endpoint):
+    config = open_api_models.Config(
+        access_key_id=access_key_id,
+        access_key_secret=access_key_secret,
+        endpoint=endpoint
+    )
+    alb_client = Alb20200616Client(config)
+    return alb_client
+
+class MyTestCase(unittest.TestCase):
+    ecs_client = connect_client(
+        access_key_id=ecs_client_params['access_key_id'],
+        access_key_secret=ecs_client_params['access_key_secret'],
+        region_id=ecs_client_params['region_id']
+    )
+    alb_client = connect_alb_client(
+        access_key_id=alb_client_params['access_key_id'],
+        access_key_secret=alb_client_params['access_key_secret'],
+        endpoint=alb_client_params['endpoint']
+    )
+
+
+    def test_add_servers_to_server_group(self):
+        weight = 0
+        utils.add_servers_to_server_group(self.alb_client, server_group_id, instance_id, weight)
+
+    def test_remove_servers_from_server_group(self):
+        utils.remove_servers_from_server_group(self.alb_client, server_group_id, instance_id)
+
+    def test_set_instance_weight_process_with_alb(self):
+        instance_ids = ["i-bp19n839usecekzn2ig9"]  # PaddlePaddle_k8s集群登陆实例，不影响线上服务
+        weight_list = [(10, 1)]  # 权重和sleep时间
+        utils.set_instance_weight_process_with_alb(self.alb_client, [server_group_id], instance_ids, weight_list)
 
 if __name__ == '__main__':
     unittest.main()