Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use get_docker_client method in testcontainers-python

Best Python code snippet using testcontainers-python_python

docker_lib.py

Source:docker_lib.py

...25    DOCKER_ZONE: 'defaultzone',26    DOCKER_LAUNCHED: []27}28client=None29def get_docker_client():30    global client31    if not client:32        try:33            client = docker.from_env()34        except Exception:35            if cli_exception.CliException.is_debug():36                logging.exception("Exception occurs during docker client creation.")37            raise cli_exception.CliException("Exception occurs during docker client creation.")38    return client39def get_setting(key):40    return config_global.get_attr(key, default_settings.get(key))41def add_set(key, list_to_add):42    old_list = get_setting(key)43    config_global.set_attr(key, old_list + list_to_add)44def install():45    hooks.register(HOOK_NAME_TEST_START, HOOK_TEST_START)46    hooks.register(HOOK_NAME_TEST_STOP, HOOK_TEST_STOP)47def uninstall():48    hooks.unregister(HOOK_NAME_TEST_START, HOOK_TEST_START)49    hooks.unregister(HOOK_NAME_TEST_STOP, HOOK_TEST_STOP)50def status():51    installed = "installed" if hooks.is_registered(HOOK_NAME_TEST_START, HOOK_TEST_START) else "not installed"52    print("configuration:")53    for k in default_settings:54        print(k + " = " + str(get_setting(k)))55    print()56    print("docker hooks is " + installed)57    print()58    check_images(DOCKER_LG_IMAGE)59    check_images(DOCKER_CONTROLLER_IMAGE)60    check_docker()61    launched_container = get_setting(DOCKER_LAUNCHED)62    if launched_container:63        print("\ncontainers :")64        for c_id in launched_container:65            print(container_status(c_id))66    else:67        try:68            check_zone(get_setting(DOCKER_ZONE))69        except cli_exception.CliException as ex:70            print(str(ex))71def container_status(container_id):72    try:73        container = get_docker_client().containers.get(container_id)74        return container.name + ' [' + container.status + ']'75    except docker.errors.NotFound:76        return container_id + ' [NOT FOUND]'77def check_images(key):78    image_name = get_setting(key)79    image_status = " is not pulled" if len(get_docker_client().images.list(image_name)) == 0 else " is pulled"80    print("image " + image_name + image_status)81def up(no_wait):82    zone = get_setting(DOCKER_ZONE)83    check_zone(zone)84    start_infra(85        zone,86        int(get_setting(DOCKER_CONTROLLER_DEFAULT_COUNT)),87        int(get_setting(DOCKER_LG_DEFAULT_COUNT)),88        no_wait,89        'manual'90    )91def check_zone(zone_set):92    zone = get_zone(zone_set)93    if not zone:94        raise cli_exception.CliException("zone " + zone_set + " doesn't exist")95    if zone.get('type') != "STATIC":96        raise cli_exception.CliException("zone " + zone_set + " is not static !")97    if zone.get('controller'):98        raise cli_exception.CliException(99            "controller with " + zone_set + " zone is not empty. Try neoload docker clean")100    if zone.get('loadgenerators'):101        logging.warning("lg zone with " + zone_set + " is not empty. Try neoload docker clean")102def get_zone(zone_id):103    for zone in zones.get_zones():104        if zone.get('id') == zone_id:105            return zone106    return None107def down():108    stop_infra()109def compute_hosts(lg_containers):110    hosts = {}111    for lg_container in lg_containers:112        lg_container.reload()  # refresh data after launch.113        hosts[lg_container.name] = lg_container.attrs['NetworkSettings']['IPAddress']114    return hosts115def generate_conf_ctrl(zone, lg_containers):116    env = {"MODE": "Managed"}117    env.update(common_env(zone))118    return {119        'env': env,120        'name_prefix': 'ctrl',121        'hosts': compute_hosts(lg_containers)122    }123def generate_conf_lg(zone):124    return {125        'env': common_env(zone),126        'name_prefix': 'lg'127    }128def stop_infra():129    for container_id in get_setting(DOCKER_LAUNCHED):130        try:131            container = get_docker_client().containers.get(container_id)132            container.stop()133        except docker.errors.NotFound:134            pass135    forget()136def common_env(zone):137    return {138        "NEOLOADWEB_URL": user_data.get_user_data().get_url(),139        "NEOLOADWEB_TOKEN": user_data.get_user_data().get_token(),140        "ZONE": zone141    }142def launch_ctrl(count, zone, lg_containers, reason):143    image = pull_if_needed(get_setting(DOCKER_CONTROLLER_IMAGE))144    return start_container(image, generate_conf_ctrl(zone, lg_containers), count, reason)145def launch_lg(count, zone, reason):146    image = pull_if_needed(get_setting(DOCKER_LG_IMAGE))147    return start_container(image, generate_conf_lg(zone), count, reason)148def extract_number(prefix, element):149    num, number_match = re.subn("^" + prefix + "-([0-9]+)-.*$", r"\g<1>", element.name)150    return int(num) if number_match > 0 else 0151def max_number(prefix):152    containers_list = get_docker_client().containers.list(all=True)153    return max(map(lambda c: extract_number(prefix, c), containers_list)) if containers_list else 0154def start_container(image, configuration, count, reason):155    prefix = configuration.get('name_prefix')156    number = max_number(prefix)157    containers = []158    for i in range(count):159        name = prefix + '-' + str(number + i + 1) + '-' + socket.gethostname().lower()160        container = get_docker_client().containers.run(161            image=image.id,162            name=name,163            hostname=name,164            labels={'launched-by-neoload-cli': reason},165            detach=True,166            extra_hosts=configuration.get('hosts', {}),167            auto_remove=True,168            environment=configuration.get('env', {})169        )170        containers.append(container)171    return containers172def make_waiting(containers_name, zone):173    waiting_list = []174    names = list(map(lambda c: c['name'], zone.get('controllers') + zone.get('loadgenerators')))175    for c_name in containers_name:176        if c_name not in names:177            waiting_list.append(c_name)178    return waiting_list179def wait_is_up(containers, zone_id):180    containers_name = map(lambda c: c.name, containers)181    while True:182        time.sleep(5)183        waiting = make_waiting(containers_name, get_zone(zone_id))184        if waiting:185            print("Waiting: " + str(waiting))186        else:187            break188def start_infra(zone_id, ctrl_count: int, lg_count: int, no_wait, reason):189    containers = []190    try:191        containers = launch_lg(lg_count, zone_id, reason)192        containers += launch_ctrl(ctrl_count, zone_id, containers, reason)193        add_set(DOCKER_LAUNCHED, list(map(lambda c: c.id, containers)))194        if not no_wait:195            wait_is_up(containers, zone_id)196    except docker.errors.DockerException as ex:197        for container in containers:198            container.stop()199        logging.error("Unexpected error in 'attach':" + str(ex))200def extract_lg_number(test_settings_json, zone):201    number_lg = test_settings_json.get('lgZoneIds').get(zone)202    return number_lg if number_lg else 0203def hook_test_start(test_settings_json):204    zone = get_setting(DOCKER_ZONE)205    if test_settings_json.get('controllerZoneId') == zone:206        number_lg = extract_lg_number(test_settings_json, zone)207        zone_obj = get_zone(zone)208        if number_lg <= len(zone_obj.get('loadgenerators')) and 1 <= len(zone_obj.get('controllers')):209            print("zone is already up", file=sys.stderr)210            return211        print("Launch docker containers", file=sys.stderr)212        check_zone(zone)213        start_infra(214            zone,215            1,216            number_lg,217            False,218            test_settings_json.get('id')219        )220def hook_test_stop():221    print("Stop docker containers", file=sys.stderr)222    stop_infra()223def pull_if_needed(image_name):224    try:225        return get_docker_client().images.get(image_name)226    except docker.errors.ImageNotFound:227        print("Pulling [" + image_name + "]", file=sys.stderr)228        return get_docker_client().images.pull(image_name)229def check_docker():230    preempt_msg = "Unexpected error in 'try_docker_system':"231    try:232        preempt_msg = "Could not ping the Docker host."233        get_docker_client().ping()234        preempt_msg = "Could not obtain version info from the Docker host."235        get_docker_client().version()236        preempt_msg = "Could not list containers on the Docker host."237        get_docker_client().containers.list()238    except Exception:239        exc_type, exc_value, exc_traceback = sys.exc_info()240        full_msg = preempt_msg + repr(traceback.format_exception(exc_type, exc_value, exc_traceback))241        lower = full_msg.lower()242        if 'connectionerror' in lower and 'permission denied' in lower:243            msg = preempt_msg + " Do you have rights (i.e. sudo first)?"244        elif 'connection refused' in lower:245            msg = "Docker installed, but connection to dockerd failed."246        else:247            msg = "Docker-specific error: " + full_msg248        tools.system_exit({'message': msg, 'code': 2})249def clean():250    list_to_clean = get_docker_client().containers.list(all=True, filters={'label': 'launched-by-neoload-cli'})251    for container in list_to_clean:252        container.stop()253    forget()254def kill():255    list_to_clean = get_docker_client().containers.list(all=True, filters={'label': 'launched-by-neoload-cli'})256    for container in list_to_clean:257        container.remove(force=True, v=True)258    forget()259def forget():260    config_global.set_attr(DOCKER_LAUNCHED, None)261def set_client(new_client):262    global client...

launch.py

Source:launch.py

...25LOGGER = logging.getLogger(__name__)26BUILTIN_DISCOVERY_MODELS = "builtin-discovery-models"27CUSTOM_DISCOVERY_INTERPRETER = "custom-discovery-interpreter"28BUILTIN_NLPROCESSOR_MODELS = "builtin-nlprocessor-models"29def get_docker_client():30    """31    Returns docker client.32    Invoke this repeatedly instead of persisting an object in python33    """34    return docker.client.from_env()35class RetryRequest(Retry):36    """37    Custom retry class with max backoff set to TIMEOUT from client.env38    """39    BACKOFF_MAX = float(env["TIMEOUT"])40retries = RetryRequest(41    total=int(env["RETRIES"]), backoff_factor=1.0, status_forcelist=[500, 502, 503, 504]42)43request_session = requests.Session()44request_session.mount("http://", HTTPAdapter(max_retries=retries))45class LaunchTarget(Enum):46    everything = "everything"47    discovery = "discovery"48    nlprocessor = "nlprocessor"49def pull_image(image):50    """51    Try to pull the most up to date version of this image.52    If internet connectivity is down, don't error.53    """54    try:55        if env.get("AUTOMATICALLY_PULL_IMAGES").title() == "True":56            get_docker_client().images.pull(image)57    except requests.exceptions.HTTPError:58        LOGGER.warning(59            "Unable to pull latest %s. Using image as found in local registry", image60        )61def fix_volume_permissions(volume, folder):62    busybox_tag = env["BUSYBOX_TAG"]63    busybox_image = f"busybox:{busybox_tag}"64    pull_image(busybox_image)65    get_docker_client().containers.run(66        busybox_image,67        command=f"chmod -R 777 {folder}",68        user="root",69        remove=True,70        detach=False,71        volumes={volume.name: {"bind": folder, "mode": "rw"}},72    )73def load_builtin_discovery_models():74    """75    Create a docker volume,76    attach it to a busybox container,77    and load the GK interpreter models into that container78    """79    # copy default models in80    remove_volume(BUILTIN_DISCOVERY_MODELS)81    builtin_models = get_docker_client().volumes.create(name=BUILTIN_DISCOVERY_MODELS)82    folder = "/models"83    fix_volume_permissions(builtin_models, folder)84    initdiscovery_tag = env["INIT_DISCOVERY_TAG"]85    initdiscovery_project = "docker.greenkeytech.com/greenkey-discovery-sdk-private"86    initdiscovery_image = f"{initdiscovery_project}:{initdiscovery_tag}"87    pull_image(initdiscovery_image)88    initcontainer = get_docker_client().containers.run(89        initdiscovery_image,90        auto_remove=False,91        detach=True,92        volumes={builtin_models.name: {"bind": folder, "mode": "rw"}},93    )94    initcontainer.start()95    initcontainer.wait()96    initcontainer.remove(force=True)97def load_builtin_nlprocessor_models():98    """99    Create a docker volume,100    attach it to a busybox container,101    and load the GK nlprocessor models into that container102    """103    # copy default models in104    remove_volume(BUILTIN_NLPROCESSOR_MODELS)105    builtin_models = get_docker_client().volumes.create(name=BUILTIN_NLPROCESSOR_MODELS)106    folder = "/models/transformers"107    fix_volume_permissions(builtin_models, folder)108    initnlprocessor_tag = env["INIT_NLPROCESSOR_TAG"]109    initnlprocessor_project = "docker.greenkeytech.com/nlpmodelcontroller"110    initnlprocessor_image = f"{initnlprocessor_project}:{initnlprocessor_tag}"111    pull_image(initnlprocessor_image)112    initcontainer = get_docker_client().containers.run(113        initnlprocessor_image,114        auto_remove=False,115        detach=True,116        volumes={builtin_models.name: {"bind": folder, "mode": "rw"}},117    )118    initcontainer.start()119    initcontainer.wait()120    initcontainer.remove(force=True)121def load_custom_model(interpreter_directory):122    """123    Create a docker volume, attach it to a busybox container, and load our interpreter into that container124    """125    if not os.path.isdir(interpreter_directory):126        raise Exception(f"Interpreter directory {interpreter_directory} not found")127    # copy custom model in128    remove_volume(CUSTOM_DISCOVERY_INTERPRETER)129    vol = get_docker_client().volumes.create(name=CUSTOM_DISCOVERY_INTERPRETER)130    source = interpreter_directory131    destination = "/data"132    busybox_image = f"busybox:{env['BUSYBOX_TAG']}"133    pull_image(busybox_image)134    busybox = get_docker_client().containers.run(135        busybox_image,136        "sleep infinity",137        auto_remove=False,138        detach=True,139        volumes={vol.name: {"bind": destination, "mode": "rw"}},140    )141    file_like_object = io.BytesIO()142    with tarfile.open(fileobj=file_like_object, mode="w") as tar:143        for fl in glob.glob(os.path.join(source, "*")):144            tar.add(fl, arcname=fl.replace(source, ""))145    file_like_object.seek(0)146    data = file_like_object.read()147    busybox.put_archive(destination, data)148    busybox.remove(force=True)149def create_dummy_custom_model():150    """151    Create a docker volume that contains nothing so docker-compose still works152    """153    # copy custom model in154    remove_volume(CUSTOM_DISCOVERY_INTERPRETER)155    get_docker_client().volumes.create(name=CUSTOM_DISCOVERY_INTERPRETER)156def wait_on_service(address):157    return request_session.get(158        "/".join([address, "ping"]), timeout=float(env["TIMEOUT"])159    )160def wait_for_everything(target):161    """162    Wait for all services as necessary163    """164    if target in [LaunchTarget.discovery, LaunchTarget.everything]:165        wait_on_service(":".join([env["DISCOVERY_HOST"], env["DISCOVERY_PORT"]]))166    if target in [LaunchTarget.nlprocessor, LaunchTarget.everything]:167        wait_on_service(":".join([env["NLPROCESSOR_HOST"], env["NLPROCESSOR_PORT"]]))168def check_for_license():169    """170    Codify license check171    """172    assert (173        env["LICENSE_KEY"] != ""174    ), "LICENSE_KEY needed. Please set in your environment or client.env"175def create_volumes(target, interpreter_directory):176    """177    Create volumes as needed178    """179    if interpreter_directory:180        load_custom_model(interpreter_directory)181    else:182        create_dummy_custom_model()183    if target in [LaunchTarget.discovery, LaunchTarget.everything]:184        load_builtin_discovery_models()185    if target in [LaunchTarget.nlprocessor, LaunchTarget.everything]:186        load_builtin_nlprocessor_models()187def stand_up_compose(target):188    """189    Stand up compose based on target190    """191    args = ["docker-compose", "--env-file", "client.env"]192    if target in [LaunchTarget.discovery, LaunchTarget.everything]:193        discovery_image = f"docker.greenkeytech.com/discovery:{env['DISCOVERY_TAG']}"194        pull_image(discovery_image)195        args += ["-f", "discovery.yaml"]196    if target in [LaunchTarget.nlprocessor, LaunchTarget.everything]:197        nlprocessor_image = (198            f"docker.greenkeytech.com/nlprocessor:{env['NLPROCESSOR_TAG']}"199        )200        pull_image(nlprocessor_image)201        args += ["-f", "nlprocessor.yaml"]202    args += ["up", "-d", "--force-recreate"]203    with patch.object(sys, "argv", args):204        docker_compose()205def prune_volumes():206    """207    Prune any volumes not in use208    """209    get_docker_client().volumes.prune()210def prune_containers():211    """212    Prune any containers which have exited213    """214    get_docker_client().containers.prune()215def launch_docker_compose(target="everything", interpreter_directory=None):216    """217    Launch docker compose with either both discovery and nlprocessor or just one218    """219    check_for_license()220    target = LaunchTarget(target)221    prune_containers()222    prune_volumes()223    create_volumes(target, interpreter_directory)224    stand_up_compose(target)225    # block until ready226    wait_for_everything(target)227def remove_volume(volume_name):228    """229    Try to remove a docker volume230    """231    try:232        vol = get_docker_client().volumes.get(volume_name)233        for container in get_docker_client().containers.list():234            if any(235                vol.name == mount.get("Name") for mount in container.attrs["Mounts"]236            ):237                container.remove(force=True)238        vol.remove()239    except docker.errors.NotFound:240        pass241def teardown_docker_compose():242    """243    Teardown docker compose244    """245    args = [246        "docker-compose",247        "--env-file",...

docker.py

Source:docker.py

1import docker 2import sqlite33def get_docker_client():4    return docker.from_env()5def get_app_ddl(app_image: str) -> str:6    """7    Return the ddl of the app8    """9    # create a docker client10    client = get_docker_client()11    # create a container from the app_image12    logs = client.containers.run(app_image, 'schema', auto_remove=True)13    # read the output of the container14    ddl = logs.decode('utf-8')15    # remove the container16    return ddl17def get_app_config_spec(app_image: str) -> str:18    """19    Return the configuration spec of the app20    """21    # create a docker client22    client = get_docker_client()23    # create a container from the app_image24    logs = client.containers.run(app_image, 'spec', auto_remove=True)25    # read the output of the container26    spec = logs.decode('utf-8')27    # remove the container28    return spec29def run_app_image(app_image, input_db_path, config_path, log_path):30    """31    Run the app image32    """33    # create a docker client34    client = get_docker_client()35    # mount the input_db_path and config_path to the container36    volumes = {37        input_db_path: {38            'bind': '/input.db',39            'mode': 'ro'40        },41        # mount as a file42        config_path: {43            'bind': '/config.json',44            'mode': 'ro'45        }46    }47    command = "run --input-db-path /input.db --config /config.json"48    # create a container from the app_image and mount the volumes49    container = client.containers.run(app_image, command, volumes=volumes, auto_remove=True, detach=True, )50    # redirect the output of the container to log_path51    with open(log_path, 'w') as f:52        for line in container.logs(stream=True):53            f.write(line.decode('utf-8'))54def verify_if_docker_image_exists(image_name: str) -> bool:55    """56    Return True if the docker image exists57    """58    client = get_docker_client()59    try:60        client.images.get(image_name)61        return True62    except docker.errors.ImageNotFound:63        return False64def setup_temp_db(db_path, app_image):65    """66    Create tables in db_path according to the DDL as specified by app_image67    """68    ddl = get_app_ddl(app_image)69    conn = sqlite3.connect(db_path)70    conn.executescript(ddl)71    conn.commit()72    conn.close()

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.