How to use test_docker_kwargs method in testcontainers-python

Best Python code snippet using testcontainers-python_python

test_clients_docker.py

Source:test_clients_docker.py

1import docker2import unittest3from rigelcore.clients import DockerClient4from rigelcore.exceptions import (5    DockerAPIError,6    DockerOperationError,7    InvalidDockerClientInstanceError,8    InvalidDockerImageNameError,9)10from unittest.mock import MagicMock, Mock, patch11class DockerClientTesting(unittest.TestCase):12    """13    Test suite for rigelcore.clients.DockerClient class.14    """15    def test_invalid_docker_client_instance_error(self) -> None:16        """17        Ensure that InvalidDockerClientInstanceError is thrown18        if an invalid instance of docker.client.DockerClient is provided.19        """20        with self.assertRaises(InvalidDockerClientInstanceError):21            DockerClient('invalid_docker_client_instance')22    @patch('rigelcore.clients.docker.docker.from_env')23    def test_docker_from_env_error(self, docker_mock: Mock) -> None:24        """25        Ensure that an instance of DockerAPIError is thrown26        if an error while trying to connect to the Docker daemon.27        """28        docker_mock.side_effect = docker.errors.DockerException()29        with self.assertRaises(DockerAPIError):30            DockerClient()31    @patch('rigelcore.clients.docker.docker.from_env')32    def test_docker_get_image_error(self, docker_mock: Mock) -> None:33        """34        Ensure that an instance of DockerAPIError is thrown35        if an error occurs while searching for a Docker image using the Docker API.36        """37        test_exception = docker.errors.DockerException()38        docker_client_mock = MagicMock()39        docker_client_mock.images.get.side_effect = test_exception40        docker_mock.return_value = docker_client_mock41        with self.assertRaises(DockerAPIError) as context:42            docker_client = DockerClient()43            docker_client.get_image('test_docker_image_name')44        self.assertEqual(context.exception.kwargs['exception'], test_exception)45    @patch('rigelcore.clients.docker.docker.from_env')46    def test_docker_get_image_none(self, docker_mock: Mock) -> None:47        """48        Ensure that the mechanism to retrieve existing Docker images49        works as expected when the specified Docker image does not exist.50        """51        test_exception = docker.errors.ImageNotFound(message='test_error_message')52        docker_client_mock = MagicMock()53        docker_client_mock.images.get.side_effect = test_exception54        docker_mock.return_value = docker_client_mock55        docker_client = DockerClient()56        image = docker_client.get_image('test_docker_image_name')57        self.assertIsNone(image)58    @patch('rigelcore.clients.docker.docker.from_env')59    def test_docker_get_image(self, docker_mock: Mock) -> None:60        """61        Ensure that the mechanism to retrieve existing Docker images works as expected.62        """63        test_docker_image = docker.models.images.Image()64        docker_client_mock = MagicMock()65        docker_client_mock.images.get.return_value = test_docker_image66        docker_mock.return_value = docker_client_mock67        docker_client = DockerClient()68        image = docker_client.get_image('test_docker_image_name')69        self.assertEqual(image, test_docker_image)70    @patch('rigelcore.clients.docker.DockerClient.print_logs')71    @patch('rigelcore.clients.docker.docker.from_env')72    def test_client_image_build(self, docker_mock: Mock, print_mock: Mock) -> None:73        """74        Ensure that the creation of Docker images works as expected.75        """76        test_context_path = 'test_context_path'77        test_dockerfile_path = 'test_dockerfile_path'78        test_image = 'test_image'79        test_buildargs = {'TEST_VARIABLE': 'TEST_VALUE'}80        docker_api_mock_return_value = 'TestImageObject'81        docker_api_mock = MagicMock()82        docker_api_mock.api.build.return_value = docker_api_mock_return_value83        docker_mock.return_value = docker_api_mock84        docker_client = DockerClient()85        docker_client.build_image(test_context_path, test_dockerfile_path, test_image, test_buildargs)86        docker_api_mock.api.build.assert_called_once_with(87            path=test_context_path,88            decode=True,89            rm=True,90            dockerfile=test_dockerfile_path,91            tag=test_image,92            buildargs=test_buildargs93        )94        print_mock.assert_called_once_with(docker_api_mock_return_value)95    @patch('rigelcore.clients.docker.docker.from_env')96    def test_client_image_build_error(self, docker_mock: Mock) -> None:97        """98        Ensure that an instance of DockerAPIError is thrown99        if an error occurs while trying to build a Docker image.100        """101        test_exception = docker.errors.DockerException()102        docker_api_mock = MagicMock()103        docker_api_mock.api.build.side_effect = test_exception104        docker_mock.return_value = docker_api_mock105        with self.assertRaises(DockerAPIError) as context:106            docker_client = DockerClient()107            docker_client.build_image(108                'test_context_path',109                'test_dockerfile_path',110                'test_image',111                {'TEST_VARIABLE': 'TEST_VALUE'}112            )113        self.assertEqual(context.exception.kwargs['exception'], test_exception)114    def test_invalid_docker_image_name_error(self) -> None:115        """116        Ensure that InvalidDockerImageNameError is thrown if an invalid target image117        name is provided to function 'tag'.118        """119        test_target_image = 'invalid:image:name'120        with self.assertRaises(InvalidDockerImageNameError) as context:121            docker_client = DockerClient()122            docker_client.tag_image('test_source_image', test_target_image)123        self.assertEqual(context.exception.kwargs['image'], test_target_image)124    @patch('rigelcore.clients.docker.docker.from_env')125    def test_docker_image_name_split_default_tag(self, docker_mock: Mock) -> None:126        """127        Ensure that the target Docker image name is properly split when128        it does not contain character ':'.129        """130        test_source_image = 'test_source_image'131        test_target_image = 'test_target_image'132        docker_api_mock = MagicMock()133        docker_mock.return_value = docker_api_mock134        docker_client = DockerClient()135        docker_client.tag_image(test_source_image, test_target_image)136        docker_api_mock.api.tag.assert_called_once_with(137            image=test_source_image,138            repository=test_target_image,139            tag='latest'140        )141    @patch('rigelcore.clients.docker.docker.from_env')142    def test_docker_image_name_split_semicolon(self, docker_mock: Mock) -> None:143        """144        Ensure that the target Docker image name is properly split when145        it contains character ':'.146        """147        test_source_image = 'test_source_image'148        test_target_image = 'test_target_image'149        test_target_tag = 'test_target_tag'150        complete_test_target_image = f'{test_target_image}:{test_target_tag}'151        docker_api_mock = MagicMock()152        docker_mock.return_value = docker_api_mock153        docker_client = DockerClient()154        docker_client.tag_image(test_source_image, complete_test_target_image)155        docker_api_mock.api.tag.assert_called_once_with(156            image=test_source_image,157            repository=test_target_image,158            tag=test_target_tag159        )160    @patch('rigelcore.clients.docker.docker.from_env')161    def test_docker_image_tag_api_error(self, docker_mock: Mock) -> None:162        """163        Ensure that an instance of DockerAPIError is thrown164        if an error occurs while attempting to tag a Docker image.165        """166        test_exception = docker.errors.DockerException()167        docker_api_mock = MagicMock()168        docker_api_mock.api.tag.side_effect = test_exception169        docker_mock.return_value = docker_api_mock170        with self.assertRaises(DockerAPIError) as context:171            docker_client = DockerClient()172            docker_client.tag_image(173                'test_source_image',174                'test_target_image'175            )176        self.assertEqual(context.exception.kwargs['exception'], test_exception)177    @patch('rigelcore.clients.docker.docker.from_env')178    def test_invalid_image_registry_error(self, docker_mock: Mock) -> None:179        """180        Ensure that DockerAPIError is thrown181        if an error occurs while attempting to login to a Docker registry.182        """183        test_exception = docker.errors.DockerException()184        docker_api_mock = MagicMock()185        docker_api_mock.api.login.side_effect = test_exception186        docker_mock.return_value = docker_api_mock187        with self.assertRaises(DockerAPIError) as context:188            docker_client = DockerClient()189            docker_client.login('test_registry', 'test_username', 'test_password')190        self.assertEqual(context.exception.kwargs['exception'], test_exception)191    @patch('rigelcore.clients.docker.docker.from_env')192    def test_client_login(self, docker_mock: Mock) -> None:193        """194        Ensure that login information is properly passed.195        """196        test_registry = 'test_registry'197        test_username = 'test_username'198        test_password = 'test_password'199        docker_api_mock = MagicMock()200        docker_mock.return_value = docker_api_mock201        docker_client = DockerClient()202        docker_client.login(test_registry, test_username, test_password)203        docker_api_mock.api.login.assert_called_once_with(204            username=test_username,205            password=test_password,206            registry=test_registry207        )208    @patch('rigelcore.clients.docker.DockerClient.print_logs')209    @patch('rigelcore.clients.docker.docker.from_env')210    def test_client_image_push(self, docker_mock: Mock, print_mock: Mock) -> None:211        """212        Ensure that the deploy of Docker images works as expected.213        """214        test_image = 'test_image'215        docker_api_mock_return_value = 'TestImageObject'216        docker_api_mock = MagicMock()217        docker_api_mock.api.push.return_value = docker_api_mock_return_value218        docker_mock.return_value = docker_api_mock219        docker_client = DockerClient()220        docker_client.push_image(test_image)221        docker_api_mock.api.push.assert_called_once_with(222            test_image,223            stream=True,224            decode=True,225        )226        print_mock.assert_called_once_with(docker_api_mock_return_value)227    @patch('rigelcore.clients.docker.DockerClient.print_logs')228    @patch('rigelcore.clients.docker.docker.from_env')229    def test_client_image_push_api_error(self, docker_mock: Mock, print_mock: Mock) -> None:230        """231        Ensure that an instance of DockerAPIError is thrown232        if an error occurs while attempting to push a Docker image to a registry.233        """234        test_exception = docker.errors.DockerException()235        docker_api_mock = MagicMock()236        docker_api_mock.api.push.side_effect = test_exception237        docker_mock.return_value = docker_api_mock238        with self.assertRaises(DockerAPIError) as context:239            docker_client = DockerClient()240            docker_client.push_image('test_image')241        self.assertEqual(context.exception.kwargs['exception'], test_exception)242    @patch('rigelcore.clients.docker.DockerLogPrinter.log')243    @patch('rigelcore.clients.docker.docker.from_env')244    def test_docker_operation_error(self, docker_mock: Mock, logger_mock: Mock) -> None:245        """246        Ensure that DockerOperationError is thrown whenever an error log is found.247        """248        test_error_message = 'Test error log message.'249        log = {'error': test_error_message}250        image_mock = MagicMock()251        image_mock.__iter__.return_value = [log]252        with self.assertRaises(DockerOperationError) as context:253            docker_client = DockerClient()254            docker_client.print_logs(image_mock)255        self.assertEqual(context.exception.kwargs['msg'], test_error_message)256        logger_mock.assert_called_once_with(log)257    @patch('rigelcore.clients.docker.docker.from_env')258    def test_docker_network_exists_true(self, docker_mock: Mock) -> None:259        """260        Ensure that the mechanism to verify if a Docker network already exists261        is working as expected.262        """263        test_docker_network = 'test_docker_network'264        docker_networks_mock = MagicMock()265        docker_networks_mock.networks.list.return_value = [docker.models.networks.Network()]266        docker_mock.return_value = docker_networks_mock267        docker_client = DockerClient()268        network = docker_client.get_network(test_docker_network)269        docker_networks_mock.networks.list.assert_called_once_with(names=[test_docker_network])270        self.assertIsNotNone(network)271    @patch('rigelcore.clients.docker.docker.from_env')272    def test_docker_network_exists_false(self, docker_mock: Mock) -> None:273        """274        Ensure that the mechanism to verify if a Docker network already exists275        is working as expected.276        """277        test_docker_network = 'test_docker_network'278        docker_networks_mock = MagicMock()279        docker_networks_mock.networks.list.return_value = False280        docker_mock.return_value = docker_networks_mock281        docker_client = DockerClient()282        network_exists = docker_client.get_network(test_docker_network)283        docker_networks_mock.networks.list.assert_called_once_with(names=[test_docker_network])284        self.assertIsNone(network_exists)285    @patch('rigelcore.clients.docker.docker.from_env')286    @patch('rigelcore.clients.docker.DockerClient.get_network')287    def test_create_docker_network_new(self, network_mock: Mock, docker_mock: Mock) -> None:288        """289        Ensure that the mechanism to create new Docker networks works as expected.290        """291        test_docker_network_name = 'test_docker_network_name'292        test_docker_network_driver = 'test_docker_network_driver'293        docker_networks_mock = MagicMock()294        docker_mock.return_value = docker_networks_mock295        network_mock.return_value = None296        docker_client = DockerClient()297        docker_client.create_network(298            test_docker_network_name,299            test_docker_network_driver300        )301        network_mock.assert_called_once_with(test_docker_network_name)302        docker_networks_mock.networks.create.assert_called_once_with(303            test_docker_network_name,304            driver=test_docker_network_driver305        )306    @patch('rigelcore.clients.docker.docker.from_env')307    @patch('rigelcore.clients.docker.DockerClient.get_network')308    def test_create_docker_network_existent(self, network_mock: Mock, docker_mock: Mock) -> None:309        """310        Ensure that the mechanism to create new Docker networks first311        verifies if a given Docker network exists before creating it.312        """313        test_docker_network_name = 'test_docker_network_name'314        test_docker_network_driver = 'test_docker_network_driver'315        docker_networks_mock = MagicMock()316        docker_mock.return_value = docker_networks_mock317        network_mock.return_value = True318        docker_client = DockerClient()319        docker_client.create_network(320            test_docker_network_name,321            test_docker_network_driver322        )323        network_mock.assert_called_once_with(test_docker_network_name)324        docker_networks_mock.networks.create.assert_not_called()325    @patch('rigelcore.clients.docker.docker.from_env')326    @patch('rigelcore.clients.docker.DockerClient.get_network')327    def test_docker_create_network_api_error_(self, network_mock: Mock, docker_mock: Mock) -> None:328        """329        Ensure that an instance of DockerAPIError is thrown330        if an error occurs while creating a new Docker network using the Docker API.331        """332        test_exception = docker.errors.DockerException()333        docker_network_mock = MagicMock()334        docker_network_mock.networks.create.side_effect = test_exception335        docker_mock.return_value = docker_network_mock336        network_mock.return_value = None337        with self.assertRaises(DockerAPIError) as context:338            docker_client = DockerClient()339            docker_client.create_network(340                'test_docker_network_name',341                'test_docker_network_driver'342            )343        self.assertEqual(context.exception.kwargs['exception'], test_exception)344    @patch('rigelcore.clients.docker.docker.from_env')345    @patch('rigelcore.clients.docker.DockerClient.get_network')346    def test_docker_remove_network_api_error(self, network_mock: Mock, docker_mock: Mock) -> None:347        """348        Ensure that an instance of DockerAPIError is thrown349        if an error occurs while deleting a Docker network using the Docker API.350        """351        test_network_name = 'test_network_name'352        test_exception = docker.errors.DockerException()353        network_instance_mock = MagicMock()354        network_instance_mock.remove.side_effect = test_exception355        network_mock.return_value = network_instance_mock356        with self.assertRaises(DockerAPIError) as context:357            docker_client = DockerClient()358            docker_client.remove_network(test_network_name)359        network_mock.assert_called_once_with(test_network_name)360        self.assertEqual(context.exception.kwargs['exception'], test_exception)361    @patch('rigelcore.clients.docker.docker.from_env')362    @patch('rigelcore.clients.docker.DockerClient.get_network')363    def test_docker_remove_network_unexistent(self, network_mock: Mock, docker_mock: Mock) -> None:364        """365        Ensure that a network delete Docker API call is only made if a given network exists.366        """367        test_network_name = 'test_network_name'368        network_instance_mock = MagicMock()369        network_instance_mock.__bool__.return_value = False370        network_mock.return_value = network_instance_mock371        docker_client = DockerClient()372        docker_client.remove_network(test_network_name)373        network_mock.assert_called_once_with(test_network_name)374        network_instance_mock.remove.assert_not_called()375    @patch('rigelcore.clients.docker.docker.from_env')376    @patch('rigelcore.clients.docker.DockerClient.get_network')377    def test_docker_remove_network_existent(self, network_mock: Mock, docker_mock: Mock) -> None:378        """379        Ensure that a network delete Docker API call is only made if a given network exists.380        """381        test_network_name = 'test_network_name'382        network_instance_mock = MagicMock()383        network_instance_mock.__bool__.return_value = True384        network_mock.return_value = network_instance_mock385        docker_client = DockerClient()386        docker_client.remove_network(test_network_name)387        network_mock.assert_called_once_with(test_network_name)388        network_instance_mock.remove.assert_called_once()389    @patch('rigelcore.clients.docker.docker.from_env')390    def test_docker_get_container_api_error(self, docker_mock: Mock) -> None:391        """392        Ensure that an instance of DockerAPIError is thrown393        if an error occurs while retrieving a Docker container information using the Docker API.394        """395        test_exception = docker.errors.DockerException()396        docker_clients_mock = MagicMock()397        docker_clients_mock.containers.list.side_effect = test_exception398        docker_mock.return_value = docker_clients_mock399        with self.assertRaises(DockerAPIError) as context:400            docker_client = DockerClient()401            docker_client.get_container('test_docker_container')402        self.assertEqual(context.exception.kwargs['exception'], test_exception)403    @patch('rigelcore.clients.docker.docker.from_env')404    def test_docker_get_container_exists(self, docker_mock: Mock) -> None:405        """406        Ensure that the mechanism to retrieve Docker containers works as expected.407        """408        test_docker_container_name = 'test_docker_container_name'409        test_docker_container_instance = docker.models.containers.Container()410        docker_clients_mock = MagicMock()411        docker_clients_mock.containers.list.return_value = [test_docker_container_instance]412        docker_mock.return_value = docker_clients_mock413        docker_client = DockerClient()414        container = docker_client.get_container(test_docker_container_name)415        docker_clients_mock.containers.list.assert_called_once_with(filters={'name': test_docker_container_name})416        self.assertEqual(container, test_docker_container_instance)417    @patch('rigelcore.clients.docker.docker.from_env')418    def test_docker_get_container_unexistent(self, docker_mock: Mock) -> None:419        """420        Ensure that the mechanism to retrieve Docker containers works as expected421        if the specified Docker container does not exist.422        """423        test_docker_container_name = 'test_docker_container_name'424        docker_clients_mock = MagicMock()425        docker_clients_mock.containers.list.return_value = []426        docker_mock.return_value = docker_clients_mock427        docker_client = DockerClient()428        container = docker_client.get_container(test_docker_container_name)429        docker_clients_mock.containers.list.assert_called_once_with(filters={'name': test_docker_container_name})430        self.assertIsNone(container)431    @patch('rigelcore.clients.docker.docker.from_env')432    @patch('rigelcore.clients.docker.DockerClient.get_container')433    def test_docker_remove_container_api_error(self, container_mock: Mock, docker_mock: Mock) -> None:434        """435        Ensure that a DockerAPIError instance is thrown436        if an error occurs while removing a Docker container using Docker API calls.437        """438        test_exception = docker.errors.DockerException()439        container_instance_mock = MagicMock()440        container_instance_mock.__bool__.return_value = True441        container_instance_mock.remove.side_effect = test_exception442        container_mock.return_value = container_instance_mock443        with self.assertRaises(DockerAPIError) as context:444            docker_client = DockerClient()445            docker_client.remove_container('test_docker_container_name')446        self.assertEqual(context.exception.kwargs['exception'], test_exception)447    @patch('rigelcore.clients.docker.docker.from_env')448    @patch('rigelcore.clients.docker.DockerClient.get_container')449    def test_docker_remove_container_exists(self, container_mock: Mock, docker_mock: Mock) -> None:450        """451        Ensure that the mecanism to delete Docker containers works as expected.452        """453        test_docker_container_name = 'test_docker_container_name'454        container_instance_mock = MagicMock()455        container_instance_mock.__bool__.return_value = True456        container_mock.return_value = container_instance_mock457        docker_client = DockerClient()458        docker_client.remove_container(test_docker_container_name)459        container_mock.assert_called_once_with(test_docker_container_name)460        container_instance_mock.remove.assert_called_once_with(v=True, force=True)461    @patch('rigelcore.clients.docker.docker.from_env')462    @patch('rigelcore.clients.docker.DockerClient.get_container')463    def test_docker_remove_container_unexistent(self, container_mock: Mock, docker_mock: Mock) -> None:464        """465        Ensure that the mecanism to delete Docker containers works as expected466        when the specified Docker container does not exist.467        """468        test_docker_container_name = 'test_docker_container_name'469        container_instance_mock = MagicMock()470        container_instance_mock.__bool__.return_value = False471        container_mock.return_value = container_instance_mock472        docker_client = DockerClient()473        docker_client.remove_container(test_docker_container_name)474        container_mock.assert_called_once_with(test_docker_container_name)475        container_instance_mock.remove.assert_not_called()476    @patch('rigelcore.clients.docker.docker.from_env')477    @patch('rigelcore.clients.docker.DockerClient.get_container')478    def test_docker_run_container_api_error(self, container_mock: Mock, docker_mock: Mock) -> None:479        """480        Ensure that a DockerAPIError instance is thrown481        if an error occurs while running a Docker container using Docker API calls.482        """483        test_exception = docker.errors.DockerException()484        container_instance_mock = MagicMock()485        container_instance_mock.__bool__.return_value = False486        container_mock.return_value = container_instance_mock487        docker_client_mock = MagicMock()488        docker_client_mock.containers.run.side_effect = test_exception489        docker_mock.return_value = docker_client_mock490        with self.assertRaises(DockerAPIError) as context:491            docker_client = DockerClient()492            docker_client.run_container(493                'test_docker_container_name',494                'test_docker_image_name'495            )496        self.assertEqual(context.exception.kwargs['exception'], test_exception)497    @patch('rigelcore.clients.docker.docker.from_env')498    @patch('rigelcore.clients.docker.DockerClient.get_container')499    def test_docker_run_container_exists(self, container_mock: Mock, docker_mock: Mock) -> None:500        """501        Ensure that a Docker container is only run if no other Docker container502        exists with the same name.503        """504        container_instance_mock = MagicMock()505        container_instance_mock.__bool__.return_value = True506        container_mock.return_value = container_instance_mock507        docker_client = DockerClient()508        container = docker_client.run_container(509            'test_docker_container_name',510            'test_docker_image_name'511        )512        self.assertEqual(container, container_instance_mock)513    @patch('rigelcore.clients.docker.docker.from_env')514    @patch('rigelcore.clients.docker.DockerClient.get_container')515    def test_docker_run_container_unexistent(self, container_mock: Mock, docker_mock: Mock) -> None:516        """517        Ensure that the mechanism to run Docker containers works as expected.518        """519        test_docker_container_instance = docker.models.containers.Container()520        container_instance_mock = MagicMock()521        container_instance_mock.__bool__.return_value = False522        container_mock.return_value = container_instance_mock523        docker_client_mock = MagicMock()524        docker_client_mock.containers.run.return_value = test_docker_container_instance525        docker_mock.return_value = docker_client_mock526        docker_client = DockerClient()527        container = docker_client.run_container(528            'test_docker_container_name',529            'test_docker_image_name'530        )531        self.assertEqual(container, test_docker_container_instance)532    @patch('rigelcore.clients.docker.docker.from_env')533    @patch('rigelcore.clients.docker.DockerClient.get_container')534    def test_docker_run_container_call(self, container_mock: Mock, docker_mock: Mock) -> None:535        """536        Ensure that the call to the Docker API is done properly when running Docker containers.537        """538        test_docker_container_name = 'test_docker_container_name'539        test_docker_image = 'test_docker_image'540        test_docker_kwargs = {541            "test_kwargs_field_str": 'test_docker_command',542            "test_kwargs_field_bool":  False,543        }544        container_instance_mock = MagicMock()545        container_instance_mock.__bool__.return_value = False546        container_mock.return_value = container_instance_mock547        docker_client_mock = MagicMock()548        docker_mock.return_value = docker_client_mock549        docker_client = DockerClient()550        docker_client.run_container(551            test_docker_container_name,552            test_docker_image,553            **test_docker_kwargs554        )555        test_docker_kwargs['name'] = test_docker_container_name556        docker_client_mock.containers.run.assert_called_once_with(557            test_docker_image,558            **test_docker_kwargs559        )560    @patch('rigelcore.clients.docker.docker.from_env')561    @patch('rigelcore.clients.docker.DockerClient.get_container')562    def test_wait_container_status_unexistent(self, container_mock: Mock, docker_mock: Mock) -> None:563        """564        Ensure that an instance of DockerAPIError is thrown565        if a Docker container to watch does not exist.566        """567        container_instance_mock = MagicMock()568        container_instance_mock.__bool__.return_value = False569        container_mock.return_value = container_instance_mock570        with self.assertRaises(DockerAPIError):571            docker_client = DockerClient()572            docker_client.wait_for_container_status('test_container_name', 'test_status')573    @patch('rigelcore.clients.docker.docker.from_env')574    @patch('rigelcore.clients.docker.DockerClient.get_container')575    @patch('rigelcore.clients.docker.time.sleep')576    def test_wait_container_status_timeout(self, sleep_mock: Mock, container_mock: Mock, docker_mock: Mock) -> None:577        """578        Ensure that an instance of DockerAPIError is thrown579        if a Docker container to watch does not exist.580        """581        container_instance_mock = MagicMock()582        container_instance_mock.__bool__.return_value = True583        container_instance_mock.status = 'test_invalid_status'584        container_mock.return_value = container_instance_mock585        with self.assertRaises(DockerAPIError):586            docker_client = DockerClient()587            docker_client.wait_for_container_status('test_container_name', 'test_status')588    @patch('rigelcore.clients.docker.docker.from_env')589    @patch('rigelcore.clients.docker.DockerClient.get_container')590    @patch('rigelcore.clients.docker.time.sleep')591    def test_wait_container_status_loop(self, sleep_mock: Mock, container_mock: Mock, docker_mock: Mock) -> None:592        """593        Ensure that the mechanism to wait for a given container status value594        to change works as expected.595        """596        test_desired_status = 'test_desired_status'597        container_instance_mock = MagicMock()598        container_instance_mock.__bool__.return_value = True599        container_instance_mock.status = 'test_invalid_status'600        container_mock.return_value = container_instance_mock601        def update_container_status(time: int) -> None:602            container_instance_mock.status = test_desired_status603        sleep_mock.side_effect = update_container_status604        docker_client = DockerClient()605        docker_client.wait_for_container_status('test_container_name', test_desired_status)606        sleep_mock.assert_called_once_with(docker_client.DOCKER_RUN_WAIT_STATUS)607if __name__ == '__main__':...

test_new_docker_api.py

Source:test_new_docker_api.py

...19    with container:20        url = container.get_connection_url()21        pattern = r'mysql\+pymysql:\/\/demo:test@[\w,.]+:(3306|32785)\/custom_db'22        assert re.match(pattern, url)23def test_docker_kwargs():24    code_dir = Path(__file__).parent25    container_first = DockerContainer("nginx:latest")26    container_first.with_volume_mapping(code_dir, '/code')27    container_second = DockerContainer("nginx:latest")28    with container_first:29        container_second.with_kwargs(volumes_from=[container_first._container.short_id])30        with container_second:31            files_first = container_first.exec('ls /code').output.decode('utf-8').strip()32            files_second = container_second.exec('ls /code').output.decode('utf-8').strip()...

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.