scripts/gateway/alb_gateway_unittest.py

import logging
import os
import unittest
from os import write

from gateway import gateway_config, alb_utils, alb_gateway_scaling_j_count
from longvideoapi.longvideoapi_config import server_group_id_list


class MyTestCase(unittest.TestCase):

    def test_send_file_to_ecs(self):
        ecs_client = alb_utils.connect_client(access_key_id=gateway_config.ecs_client_params['access_key_id'],
                                              access_key_secret=gateway_config.ecs_client_params['access_key_secret'],
                                              region_id=gateway_config.ecs_client_params['region_id'])
        instance_ids = ["i-bp19n839usecekzn2ig9"]
        alb_utils.send_file_to_ecs(ecs_client,
                                   instance_id_list=instance_ids,
                                   **gateway_config.start_sh)

    def test_run_command(self):
        ecs_client = alb_utils.connect_client(access_key_id=gateway_config.ecs_client_params['access_key_id'],
                                              access_key_secret=gateway_config.ecs_client_params['access_key_secret'],
                                              region_id=gateway_config.ecs_client_params['region_id'])
        instance_ids = ["i-bp19n839usecekzn2ig9"]
        start_sh_param = "latest"
        server_start_sh = os.path.join(gateway_config.start_sh['target_dir'], gateway_config.start_sh['name'])
        server_start_commend = f"sh {server_start_sh} {start_sh_param}"
        alb_utils.run_command(ecs_client, instance_ids, command=server_start_commend)


    def test_add_servers_to_server_group(self):
        alb_client = alb_utils.connect_alb_client(
            access_key_id=gateway_config.alb_client_params['access_key_id'],
            access_key_secret=gateway_config.alb_client_params['access_key_secret'],
            endpoint=gateway_config.alb_client_params['endpoint']
        )
        instance_ids = ["i-bp19n839usecekzn2ig9"]
        alb_utils.add_servers_to_server_group(alb_client, gateway_config.server_group_id_list, instance_ids, weight=100, port=gateway_config.port)


    def test_list_server_group_servers(self):
        alb_client = alb_utils.connect_alb_client(
            access_key_id=gateway_config.alb_client_params['access_key_id'],
            access_key_secret=gateway_config.alb_client_params['access_key_secret'],
            endpoint=gateway_config.alb_client_params['endpoint']
        )
        online_instance_ids = alb_utils.list_server_group_servers(alb_client, server_group_id=gateway_config.server_group_id_list[0])
        logging.info(online_instance_ids)


    def test_update_server_group_servers_attribute(self):
        alb_client = alb_utils.connect_alb_client(
            access_key_id=gateway_config.alb_client_params['access_key_id'],
            access_key_secret=gateway_config.alb_client_params['access_key_secret'],
            endpoint=gateway_config.alb_client_params['endpoint']
        )
        instance_id = ["i-bp19n839usecekzn2ig9"]
        add_weight_list = [(10, 1), (20, 1), (40, 1)]
        port = gateway_config.port
        alb_utils.update_server_group_servers_attribute(alb_client=alb_client,
                                                        server_group_id_list=gateway_config.server_group_id_list,
                                                        instance_id_list=instance_id,
                                                        weight_list=add_weight_list,
                                                        port=port)

    def test_remove_servers_from_server_group(self):
        alb_client = alb_utils.connect_alb_client(
            access_key_id=gateway_config.alb_client_params['access_key_id'],
            access_key_secret=gateway_config.alb_client_params['access_key_secret'],
            endpoint=gateway_config.alb_client_params['endpoint']
        )
        instance_id = ["i-bp19n839usecekzn2ig9"]
        alb_utils.remove_servers_from_server_group(alb_client,
                                                   server_group_id_list=gateway_config.server_group_id_list,
                                                   instance_ids=instance_id,
                                                   port=gateway_config.port)



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