scripts/distribution/distribution_unittest.py

import os
import unittest
import alb_utils
import distribution_config
import distribution_update_b
import logging

class MyTestCase(unittest.TestCase):

    def test_create_multiple_instances(self):
        ecs_client = alb_utils.connect_client(access_key_id=distribution_config.ecs_client_params['access_key_id'],
                                              access_key_secret=distribution_config.ecs_client_params['access_key_secret'],
                                              region_id=distribution_config.ecs_client_params['region_id'])

        ess_instance_ids = alb_utils.create_multiple_instances(
            amount=1,
            ecs_client=ecs_client,
            **distribution_config.instance_config_b,
        )
        logging.info(ess_instance_ids)


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

    def test_run_command(self):
        ecs_client = alb_utils.connect_client(access_key_id=distribution_config.ecs_client_params['access_key_id'],
                                              access_key_secret=distribution_config.ecs_client_params['access_key_secret'],
                                              region_id=distribution_config.ecs_client_params['region_id'])
        instance_ids = ["i-bp1gg4mac4z31sjgq7ec"]
        start_sh_param = "latest"
        server_start_sh = os.path.join(distribution_config.start_sh['target_dir'], distribution_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_distribution_health_check(self):
        ecs_client = alb_utils.connect_client(access_key_id=distribution_config.ecs_client_params['access_key_id'],
                                              access_key_secret=distribution_config.ecs_client_params['access_key_secret'],
                                              region_id=distribution_config.ecs_client_params['region_id'])
        instance_ids = "i-bp1gg4mac4z31sjgq7ec"
        distribution_update_b.server_health_check(ecs_client,
                                                  instance_id=instance_ids)


    def test_remove_container_image(self):
        ecs_client = alb_utils.connect_client(access_key_id=distribution_config.ecs_client_params['access_key_id'],
                                              access_key_secret=distribution_config.ecs_client_params[
                                                  'access_key_secret'],
                                              region_id=distribution_config.ecs_client_params['region_id'])
        instance_id = "i-bp1gg4mac4z31sjgq7ec"
        container_name_list = ['distribution']
        distribution_update_b.remove_container_image(ecs_client,
                                                         instance_id,
                                                         container_name_list)

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