Best Python code snippet using testcontainers-python_python
container.py
Source:container.py
...51 )52 logger.info("Container started: %s", self._container.short_id)53 return self54 def stop(self, force=True, delete_volume=True):55 self.get_wrapped_container().remove(force=force, v=delete_volume)56 def __enter__(self):57 return self.start()58 def __exit__(self, exc_type, exc_val, exc_tb):59 self.stop()60 def __del__(self):61 """62 Try to remove the container in all circumstances63 """64 if self._container is not None:65 try:66 self.stop()67 except: # noqa: E72268 pass69 def get_container_host_ip(self) -> str:70 # infer from docker host71 host = self.get_docker_client().host()72 if not host:73 return "localhost"74 # check testcontainers itself runs inside docker container75 if inside_container():76 # If newly spawned container's gateway IP address from the docker77 # "bridge" network is equal to detected host address, we should use78 # container IP address, otherwise fall back to detected host79 # address. Even it's inside container, we need to double check,80 # because docker host might be set to docker:dind, usually in CI/CD environment81 gateway_ip = self.get_docker_client().gateway_ip(self._container.id)82 if gateway_ip == host:83 return self.get_docker_client().bridge_ip(self._container.id)84 return gateway_ip85 return host86 @wait_container_is_ready()87 def get_exposed_port(self, port) -> str:88 mapped_port = self.get_docker_client().port(self._container.id, port)89 if inside_container():90 gateway_ip = self.get_docker_client().gateway_ip(self._container.id)91 host = self.get_docker_client().host()92 if gateway_ip == host:93 return port94 return mapped_port95 def with_command(self, command: str) -> 'DockerContainer':96 self._command = command97 return self98 def with_name(self, name: str) -> 'DockerContainer':99 self._name = name100 return self101 def with_volume_mapping(self, host: str, container: str,102 mode: str = 'ro') -> 'DockerContainer':103 # '/home/user1/': {'bind': '/mnt/vol2', 'mode': 'rw'}104 mapping = {'bind': container, 'mode': mode}105 self.volumes[host] = mapping106 return self107 def get_wrapped_container(self) -> Container:108 return self._container109 def get_docker_client(self) -> DockerClient:110 return self._docker111 def get_logs(self):112 if not self._container:113 raise ContainerStartException("Container should be started before")114 return self._container.logs(stderr=False), self._container.logs(stdout=False)115 def exec(self, command):116 if not self._container:117 raise ContainerStartException("Container should be started before")...
kafka.py
Source:kafka.py
...46 tarinfo.mtime = time.time()47 tar.addfile(tarinfo, BytesIO(data))48 tar.close()49 tar_stream.seek(0)50 self.get_wrapped_container().put_archive('/', tar_stream)51 def start(self):52 self.with_command(f'sh -c "while [ ! -f {KafkaContainer.TC_START_SCRIPT} ]; do sleep 0.1; done; '53 f'sh {KafkaContainer.TC_START_SCRIPT}"')54 super().start()55 self.tc_start()56 self._connect()...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!