Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_all_container_names method in localstack

Best Python code snippet using localstack_python

test_lambda_whitebox.py

Source:test_lambda_whitebox.py

...185        func_name = "test_prime_and_destroy_containers"186        func_arn = lambda_api.func_arn(func_name)187        # make sure existing containers are gone188        executor.cleanup()189        self.assertEqual(0, len(executor.get_all_container_names()))190        # deploy and invoke lambda without Docker191        testutil.create_lambda_function(192            func_name=func_name,193            handler_file=TEST_LAMBDA_ENV,194            libs=TEST_LAMBDA_LIBS,195            envvars={"Hello": "World"},196        )197        self.assertEqual(0, len(executor.get_all_container_names()))198        self.assertDictEqual({}, executor.function_invoke_times)199        # invoke a few times.200        durations = []201        num_iterations = 3202        for i in range(0, num_iterations + 1):203            prev_invoke_time = None204            if i > 0:205                prev_invoke_time = executor.function_invoke_times[func_arn]206            start_time = time.time()207            self.lambda_client.invoke(FunctionName=func_name, Payload=b"{}")208            duration = time.time() - start_time209            self.assertEqual(1, len(executor.get_all_container_names()))210            # ensure the last invoke time is being updated properly.211            if i > 0:212                self.assertGreater(executor.function_invoke_times[func_arn], prev_invoke_time)213            else:214                self.assertGreater(executor.function_invoke_times[func_arn], 0)215            durations.append(duration)216        # the first call would have created the container. subsequent calls would reuse and be faster.217        for i in range(1, num_iterations + 1):218            self.assertLess(durations[i], durations[0])219        status = executor.get_docker_container_status(func_arn)220        self.assertEqual(1, status)221        container_network = executor.get_docker_container_network(func_arn)222        self.assertEqual("bridge", container_network)223        executor.cleanup()224        status = executor.get_docker_container_status(func_arn)225        self.assertEqual(0, status)226        self.assertEqual(0, len(executor.get_all_container_names()))227        # clean up228        testutil.delete_lambda_function(func_name)229    @pytest.mark.skipif(230        condition=not isinstance(231            lambda_api.LAMBDA_EXECUTOR, lambda_executors.LambdaExecutorReuseContainers232        ),233        reason="Test only applicable if docker-reuse executor is selected",234    )235    def test_destroy_idle_containers(self):236        executor = lambda_api.LAMBDA_EXECUTOR237        func_name = "test_destroy_idle_containers"238        func_arn = lambda_api.func_arn(func_name)239        # make sure existing containers are gone240        executor.destroy_existing_docker_containers()241        self.assertEqual(0, len(executor.get_all_container_names()))242        # deploy and invoke lambda without Docker243        testutil.create_lambda_function(244            func_name=func_name,245            handler_file=TEST_LAMBDA_ENV,246            libs=TEST_LAMBDA_LIBS,247            envvars={"Hello": "World"},248        )249        self.assertEqual(0, len(executor.get_all_container_names()))250        self.lambda_client.invoke(FunctionName=func_name, Payload=b"{}")251        self.assertEqual(1, len(executor.get_all_container_names()))252        # try to destroy idle containers.253        executor.idle_container_destroyer()254        self.assertEqual(1, len(executor.get_all_container_names()))255        # simulate an idle container256        executor.function_invoke_times[func_arn] = (257            int(time.time() * 1000) - lambda_executors.MAX_CONTAINER_IDLE_TIME_MS258        )259        executor.idle_container_destroyer()260        def assert_container_destroyed():261            self.assertEqual(0, len(executor.get_all_container_names()))262        retry(assert_container_destroyed, retries=3)263        # clean up264        testutil.delete_lambda_function(func_name)265class TestLocalExecutors(unittest.TestCase):266    def test_python3_runtime_multiple_create_with_conflicting_module(self):267        lambda_client = aws_stack.create_external_boto_client("lambda")268        original_do_use_docker = lambda_api.DO_USE_DOCKER269        try:270            # always use the local runner271            lambda_api.DO_USE_DOCKER = False272            python3_with_settings1 = load_file(TEST_LAMBDA_PYTHON3_MULTIPLE_CREATE1, mode="rb")273            python3_with_settings2 = load_file(TEST_LAMBDA_PYTHON3_MULTIPLE_CREATE2, mode="rb")274            lambda_name1 = "test1-%s" % short_uid()275            testutil.create_lambda_function(...

storageApplianceAzure.py

Source:storageApplianceAzure.py

...11            user_input = input("All containers [all], a specific container [con], or object [obj]? ")12            13            if user_input == "all":14                start_time = time.time()15                container_names = get_all_container_names(blob_service_client)16                display_all_containers(blob_service_client, container_names)17                print(str(time.time() - start_time) + " seconds.")            18            elif user_input == "con":19                print("Search> ", end="")20                user_input = input("Container name? ")21                start_time = time.time()22                if not is_valid_container(blob_service_client, user_input):23                    print("Error: This container/container name is invalid. Please try again.")24                    continue 25                display_container(blob_service_client, user_input)26                print(str(time.time() - start_time) + " seconds.") 27            elif user_input == "obj":28                print("Search> ", end="")29                user_input = input("Object name? ")30                start_time = time.time() 31                container_name = find_object_all_containers(blob_service_client, user_input)32                if container_name is not None:33                    print("Container: " + container_name)34                else:35                    print("Object '" + user_input + "' not found")36                print(str(time.time() - start_time) + " seconds.") 37        elif user_input == "download":38            print("Download> ", end="")39            container_name = input("Container name? ")40            if not is_valid_container(blob_service_client, container_name):41                print("Error: This container/container name is invalid. Please try again.")42                continue 43            print("Download> ", end="")44            object_name = input("Object name? ")45            start_time = time.time() 46            if find_object(blob_service_client, container_name, object_name):47                if not os.path.isdir(cfg.task_one['download_dir']):48                    os.mkdir(cfg.task_one['download_dir'])49                download_object_to_dir(blob_service_client, container_name, object_name, cfg.task_one['download_dir'])50                print("Downloaded '" + object_name + "' to " + cfg.task_one['download_dir'])51            else:52                print("Object '" + object_name + "' not found")53            print(str(time.time() - start_time) + " seconds.") 54def is_valid_container(blob_service_client, container_name):55    if container_name not in get_all_container_names(blob_service_client):56        return False57    return True58def find_object_all_containers(blob_service_client, object_name):59    container_names = get_all_container_names(blob_service_client)60    for container in container_names:61        if find_object(blob_service_client, container, object_name):62            return container63    return None64def get_all_container_names(blob_service_client):65    container_names = [container['name'] for container in blob_service_client.list_containers()]66    return container_names67def find_object(blob_service_client, container_name, object_name):68    container_client = blob_service_client.get_container_client(container_name)69    for blob in container_client.list_blobs():70        if blob.name == object_name:71            return True72    73    return False74def download_object_to_dir(blob_service_client, container_name, object_name, dir):75    container_client = blob_service_client.get_container_client(container_name)76    with open(os.path.join(dir, object_name), 'wb') as f:77        f.write(container_client.download_blob(object_name).readall())78def display_all_containers(blob_service_client, container_names):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.