scripts/gateway/alb_gateway_unittest.py

import logging
import unittest
from os import write

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


class MyTestCase(unittest.TestCase):
    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()