KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use connect_container_to_network method in localstack

Best Python code snippet using localstack_python
__init__.py
Source:__init__.py
1# -*- coding: utf-8 -*-2'''3Common logic used by the docker state and execution module4This module contains logic to accommodate docker/salt CLI usage, as well as5input as formatted by states.6'''7# Import Python libs8from __future__ import absolute_import, print_function, unicode_literals9import copy10import logging11# Import Salt libs12import salt.utils.args13import salt.utils.data14import salt.utils.docker.translate15from salt.utils.docker.translate.helpers import split as _split16from salt.exceptions import CommandExecutionError, SaltInvocationError17from salt.utils.args import get_function_argspec as _argspec18# Import 3rd-party libs19from salt.ext import six20try:21    import docker22    HAS_DOCKER_PY = True23except ImportError:24    HAS_DOCKER_PY = False25# These next two imports are only necessary to have access to the needed26# functions so that we can get argspecs for the container config, host config,27# and networking config (see the get_client_args() function).28try:29    import docker.types30except ImportError:31    pass32try:33    import docker.utils34except ImportError:35    pass36NOTSET = object()37# Default timeout as of docker-py 1.0.038CLIENT_TIMEOUT = 6039# Timeout for stopping the container, before a kill is invoked40SHUTDOWN_TIMEOUT = 1041log = logging.getLogger(__name__)42def get_client_args(limit=None):43    if not HAS_DOCKER_PY:44        raise CommandExecutionError('docker Python module not imported')45    limit = salt.utils.args.split_input(limit or [])46    ret = {}47    if not limit or any(x in limit for x in48            ('create_container', 'host_config', 'connect_container_to_network')):49        try:50            ret['create_container'] = \51                _argspec(docker.APIClient.create_container).args52        except AttributeError:53            try:54                ret['create_container'] = \55                    _argspec(docker.Client.create_container).args56            except AttributeError:57                raise CommandExecutionError(58                    'Coult not get create_container argspec'59                )60        try:61            ret['host_config'] = \62                _argspec(docker.types.HostConfig.__init__).args63        except AttributeError:64            try:65                ret['host_config'] = \66                    _argspec(docker.utils.create_host_config).args67            except AttributeError:68                raise CommandExecutionError(69                    'Could not get create_host_config argspec'70                )71        try:72            ret['connect_container_to_network'] = \73                _argspec(docker.types.EndpointConfig.__init__).args74        except AttributeError:75            try:76                ret['connect_container_to_network'] = \77                    _argspec(docker.utils.utils.create_endpoint_config).args78            except AttributeError:79                try:80                    ret['connect_container_to_network'] = \81                        _argspec(docker.utils.create_endpoint_config).args82                except AttributeError:83                    raise CommandExecutionError(84                        'Could not get connect_container_to_network argspec'85                    )86    for key, wrapped_func in (87            ('logs', docker.api.container.ContainerApiMixin.logs),88            ('create_network', docker.api.network.NetworkApiMixin.create_network)):89        if not limit or key in limit:90            try:91                func_ref = wrapped_func92                if six.PY2:93                    try:94                        # create_network is decorated, so we have to dig into the95                        # closure created by functools.wraps96                        ret[key] = \97                            _argspec(func_ref.__func__.__closure__[0].cell_contents).args98                    except (AttributeError, IndexError):99                        # functools.wraps changed (unlikely), bail out100                        ret[key] = []101                else:102                    try:103                        # functools.wraps makes things a little easier in Python 3104                        ret[key] = _argspec(func_ref.__wrapped__).args105                    except AttributeError:106                        # functools.wraps changed (unlikely), bail out107                        ret[key] = []108            except AttributeError:109                # Function moved, bail out110                ret[key] = []111    if not limit or 'ipam_config' in limit:112        try:113            ret['ipam_config'] = _argspec(docker.types.IPAMPool.__init__).args114        except AttributeError:115            try:116                ret['ipam_config'] = _argspec(docker.utils.create_ipam_pool).args117            except AttributeError:118                raise CommandExecutionError('Could not get ipam args')119    for item in ret:120        # The API version is passed automagically by the API code that imports121        # these classes/functions and is not an arg that we will be passing, so122        # remove it if present. Similarly, don't include "self" if it shows up123        # in the arglist.124        for argname in ('version', 'self'):125            try:126                ret[item].remove(argname)127            except ValueError:128                pass129    # Remove any args in host or endpoint config from the create_container130    # arglist. This keeps us from accidentally allowing args that docker-py has131    # moved from the create_container function to the either the host or132    # endpoint config.133    for item in ('host_config', 'connect_container_to_network'):134        for val in ret.get(item, []):135            try:136                ret['create_container'].remove(val)137            except ValueError:138                # Arg is not in create_container arglist139                pass140    for item in ('create_container', 'host_config', 'connect_container_to_network'):141        if limit and item not in limit:142            ret.pop(item, None)143    try:144        ret['logs'].remove('container')145    except (KeyError, ValueError, TypeError):146        pass147    return ret148def translate_input(translator,149                    skip_translate=None,150                    ignore_collisions=False,151                    validate_ip_addrs=True,152                    **kwargs):153    '''154    Translate CLI/SLS input into the format the API expects. The ``translator``155    argument must be a module containing translation functions, within156    salt.utils.docker.translate. A ``skip_translate`` kwarg can be passed to157    control which arguments are translated. It can be either a comma-separated158    list or an iterable containing strings (e.g. a list or tuple), and members159    of that tuple will have their translation skipped. Optionally,160    skip_translate can be set to True to skip *all* translation.161    '''162    kwargs = copy.deepcopy(salt.utils.args.clean_kwargs(**kwargs))163    invalid = {}164    collisions = []165    if skip_translate is True:166        # Skip all translation167        return kwargs168    else:169        if not skip_translate:170            skip_translate = ()171        else:172            try:173                skip_translate = _split(skip_translate)174            except AttributeError:175                pass176            if not hasattr(skip_translate, '__iter__'):177                log.error('skip_translate is not an iterable, ignoring')178                skip_translate = ()179    try:180        # Using list(kwargs) here because if there are any invalid arguments we181        # will be popping them from the kwargs.182        for key in list(kwargs):183            real_key = translator.ALIASES.get(key, key)184            if real_key in skip_translate:185                continue186            # ipam_pools is designed to be passed as a list of actual187            # dictionaries, but if each of the dictionaries passed has a single188            # element, it will be incorrectly repacked.189            if key != 'ipam_pools' and salt.utils.data.is_dictlist(kwargs[key]):190                kwargs[key] = salt.utils.data.repack_dictlist(kwargs[key])191            try:192                kwargs[key] = getattr(translator, real_key)(193                    kwargs[key],194                    validate_ip_addrs=validate_ip_addrs,195                    skip_translate=skip_translate)196            except AttributeError:197                log.debug('No translation function for argument \'%s\'', key)198                continue199            except SaltInvocationError as exc:200                kwargs.pop(key)201                invalid[key] = exc.strerror202        try:203            translator._merge_keys(kwargs)204        except AttributeError:205            pass206        # Convert CLI versions of commands to their docker-py counterparts207        for key in translator.ALIASES:208            if key in kwargs:209                new_key = translator.ALIASES[key]210                value = kwargs.pop(key)211                if new_key in kwargs:212                    collisions.append(new_key)213                else:214                    kwargs[new_key] = value215        try:216            translator._post_processing(kwargs, skip_translate, invalid)217        except AttributeError:218            pass219    except Exception as exc:220        error_message = exc.__str__()221        log.error(222            'Error translating input: \'%s\'', error_message, exc_info=True)223    else:224        error_message = None225    error_data = {}226    if error_message is not None:227        error_data['error_message'] = error_message228    if invalid:229        error_data['invalid'] = invalid230    if collisions and not ignore_collisions:231        for item in collisions:232            error_data.setdefault('collisions', []).append(233                '\'{0}\' is an alias for \'{1}\', they cannot both be used'234                .format(translator.ALIASES_REVMAP[item], item)235            )236    if error_data:237        raise CommandExecutionError(238            'Failed to translate input', info=error_data)239    return kwargs240def create_ipam_config(*pools, **kwargs):241    '''242    Builds an IP address management (IPAM) config dictionary243    '''244    kwargs = salt.utils.args.clean_kwargs(**kwargs)245    try:246        # docker-py 2.0 and newer247        pool_args = salt.utils.args.get_function_argspec(248            docker.types.IPAMPool.__init__).args249        create_pool = docker.types.IPAMPool250        create_config = docker.types.IPAMConfig251    except AttributeError:252        # docker-py < 2.0253        pool_args = salt.utils.args.get_function_argspec(254            docker.utils.create_ipam_pool).args255        create_pool = docker.utils.create_ipam_pool256        create_config = docker.utils.create_ipam_config257    for primary_key, alias_key in (('driver', 'ipam_driver'),258                                   ('options', 'ipam_opts')):259        if alias_key in kwargs:260            alias_val = kwargs.pop(alias_key)261            if primary_key in kwargs:262                log.warning(263                    'docker.create_ipam_config: Both \'%s\' and \'%s\' '264                    'passed. Ignoring \'%s\'',265                    alias_key, primary_key, alias_key266                )267            else:268                kwargs[primary_key] = alias_val269    if salt.utils.data.is_dictlist(kwargs.get('options')):270        kwargs['options'] = salt.utils.data.repack_dictlist(kwargs['options'])271    # Get all of the IPAM pool args that were passed as individual kwargs272    # instead of in the *pools tuple273    pool_kwargs = {}274    for key in list(kwargs):275        if key in pool_args:276            pool_kwargs[key] = kwargs.pop(key)277    pool_configs = []278    if pool_kwargs:279        pool_configs.append(create_pool(**pool_kwargs))280    pool_configs.extend([create_pool(**pool) for pool in pools])281    if pool_configs:282        # Sanity check the IPAM pools. docker-py's type/function for creating283        # an IPAM pool will allow you to create a pool with a gateway, IP284        # range, or map of aux addresses, even when no subnet is passed.285        # However, attempting to use this IPAM pool when creating the network286        # will cause the Docker Engine to throw an error.287        if any('Subnet' not in pool for pool in pool_configs):288            raise SaltInvocationError('A subnet is required in each IPAM pool')289        else:290            kwargs['pool_configs'] = pool_configs291    ret = create_config(**kwargs)292    pool_dicts = ret.get('Config')293    if pool_dicts:294        # When you inspect a network with custom IPAM configuration, only295        # arguments which were explictly passed are reflected. By contrast,296        # docker-py will include keys for arguments which were not passed in297        # but set the value to None. Thus, for ease of comparison, the below298        # loop will remove all keys with a value of None from the generated299        # pool configs.300        for idx, _ in enumerate(pool_dicts):301            for key in list(pool_dicts[idx]):302                if pool_dicts[idx][key] is None:303                    del pool_dicts[idx][key]...
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.