scripts/test/albtest.py

import unittest
import sys
import logging
from util 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

# 配置客户端参数
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
        server_group_id = "sgp-ec4gopoclruofsfmxu"
        instance_id = "i-bp19n839usecekzn2ig9"
        utils.add_servers_to_server_group(self.alb_client, server_group_id, instance_id, weight, port=80)

    def test_list_server_group_servers(self):
        instance_ids = utils.list_server_group_servers(self.alb_client, server_group_id)
        print(instance_ids)


#  todo
    def test_remove_servers_from_server_group(self):
        utils.remove_servers_from_server_group(self.alb_client, server_group_id, instance_id, port=80)




    def test_update_server_group_servers_attribute(self):
        instance_id = ["i-bp19n839usecekzn2ig9"]
        server_group_id = "sgp-ec4gopoclruofsfmxu"
        add_weight_list = [(10, 3), (0, 1)]
        port = "80"
        utils.update_server_group_servers_attribute(self.alb_client, [server_group_id], instance_id, add_weight_list, port)



    def test_run_command(self):
        instance_id = ["i-bp19n839usecekzn2ig9"]
        command="ls"
        utils.run_command(self.ecs_client, instance_ids=instance_id, command=command)






if __name__ == '__main__':
    unittest.main()