KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run testcontainers-python automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
