scripts/albtest.py

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
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()