How to use get_bootstrap_server method in testcontainers-python

Best Python code snippet using testcontainers-python_python

test_detection_service.py

Source:test_detection_service.py

...12        value=json.dumps(message),13    )14    detection_service.producer.flush()15def create_topics(container):16    create_topic(container.get_bootstrap_server(), OUTPUT_TOPIC)17    create_topic(container.get_bootstrap_server(), SCAN_TOPIC)18    create_topic(container.get_bootstrap_server(), FISHING_TOPIC)19def get_output_consumer(kafka):20    output_consumer = get_consumer(kafka.get_bootstrap_server())21    output_consumer.subscribe([OUTPUT_TOPIC])22    return output_consumer23def get_fishing_topic(kafka):24    fishing_consumer = get_consumer(kafka.get_bootstrap_server())25    fishing_consumer.subscribe([FISHING_TOPIC])26    return fishing_consumer27def get_scan_consumer(container):28    scan_consumer = get_consumer(container.get_bootstrap_server())29    scan_consumer.subscribe([SCAN_TOPIC])30    return scan_consumer31class TestDetectionService(unittest.TestCase):32    def test_no_engine_send_to_scan_topic(self):33        with KafkaContainer("confluentinc/cp-kafka:latest") as container:34            create_topics(container)35            detection_service.producer = get_producer(container.get_bootstrap_server())36            send_message({"transaction_id": 1, "is_phishing": False})37            detection_service.producer.flush()38            output_consumer = get_output_consumer(container)39            detection_service.consume_message(output_consumer)40            output_consumer.close()41            scan_consumer = get_scan_consumer(container)42            msg = scan_consumer.poll(5.0)43            assert msg is not None44            assert (45                msg.value().decode("utf-8")46                == '{"transaction_id": 1, "is_phishing": false}'47            )48    def test_engine_exists_phishing_false(self):49        with KafkaContainer("confluentinc/cp-kafka:latest") as kafka:50            create_topics(kafka)51            detection_service.producer = get_producer(kafka.get_bootstrap_server())52            send_message(53                {54                    "transaction_id": 1,55                    "is_phishing": False,56                    "engine_name": "engine1",57                }58            )59            output_consumer = get_output_consumer(kafka)60            detection_service.consume_message(output_consumer)61            output_consumer.close()62            fishing_consumer = get_fishing_topic(kafka)63            msg = fishing_consumer.poll(5.0)64            assert msg is None65            scan_consumer = get_scan_consumer(kafka)66            msg = scan_consumer.poll(5.0)67            assert msg is None68    def test_engine_exists_phishing_true(self):69        with KafkaContainer(70            "confluentinc/cp-kafka:latest"71        ) as kafka, RedisContainer() as redis_container:72            create_topics(kafka)73            detection_service.producer = get_producer(kafka.get_bootstrap_server())74            detection_service.redis = redis.Redis(75                host="localhost", port=int(redis_container.get_exposed_port(6379)), db=076            )77            output_consumer = get_output_consumer(kafka)78            for x in range(2):79                send_message(80                    {81                        "transaction_id": 1,82                        "is_phishing": True,83                        "engine_name": "engine1",84                    }85                )86                detection_service.consume_message(output_consumer)87            output_consumer.close()...

kafka.py

Source:kafka.py

...31        self.with_env("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")32        self.with_env("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", "1")33        self.with_env("KAFKA_LOG_FLUSH_INTERVAL_MESSAGES", "10000000")34        self.with_env("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0")35    def get_bootstrap_server(self):36        return f"{self.get_container_host_ip()}:{self.get_exposed_port(self.port_to_expose)}"37    def start(self):38        self.with_command(39            f'sh -c "while [ ! -f {self.START_SCRIPT_PATH} ]; do sleep 0.1; done; sh {self.START_SCRIPT_PATH}"'40        )41        super().start()42        self._start_kafka()43        self.is_ready()44        return self45    def _start_kafka(self):46        start_script = (47            dedent(48                f"""49            #!/usr/bin/env bash50            set -e51            echo "clientPort=2181" > zookeeper.properties52            echo "dataDir=/var/lib/zookeeper/data" >> zookeeper.properties53            echo "dataLogDir=/var/lib/zookeeper/log" >> zookeeper.properties54            zookeeper-server-start zookeeper.properties &55            export KAFKA_ZOOKEEPER_CONNECT="localhost:2181"56            KAFKA_ADVERTISED_LISTENERS="PLAINTEXT://localhost:{self.get_exposed_port(self.port_to_expose)},BROKER://$(hostname -i):9092"57            export KAFKA_ADVERTISED_LISTENERS58            . /etc/confluent/docker/bash-config59            /etc/confluent/docker/configure60            /etc/confluent/docker/launch61        """62            )63            .strip()64            .encode("utf-8")65        )66        self.copy(start_script, self.START_SCRIPT_PATH)67    def copy(self, content: bytes, path: str):68        with BytesIO() as archive, tarfile.TarFile(fileobj=archive, mode="w") as tar:69            tarinfo = tarfile.TarInfo(name=path)70            tarinfo.size = len(content)71            tarinfo.mtime = time.time()72            tar.addfile(tarinfo, BytesIO(content))73            archive.seek(0)74            self.get_wrapped_container().put_archive("/", archive)75    @wait_container_is_ready()76    def is_ready(self) -> bool:77        consumer = KafkaConsumer(78            group_id="test",79            bootstrap_servers=self.get_bootstrap_server(),80        )81        if not consumer.topics():82            raise KafkaError(83                f"Could not connect to Kafka at {self.get_bootstrap_server()}"84            )...

main.py

Source:main.py

...29    return parser30def get_topic():31    parser = get_config_parser()32    return parser.get('destination', 'topic')33def get_bootstrap_server():34    parser = get_config_parser()35    return parser.get('destination', 'bootstrap_server')36def publish_on_topic(user_details: UserDetails):37    global producer38    if producer is None:39        producer = KeyboardInputProducer(bootstrap_server=get_bootstrap_server(), topic=get_topic())40    producer.publish_message(user_details)41def send_message(message):42    if message == '' or message is None: return43    ip = get_ip_address()44    user_details = UserDetails(ip, message)45    add_to_cache(user_details)46def clear_cache():47    global cache48    cache = []49def add_to_cache(details: UserDetails):50    global cache51    global cache_size52    cache.append(details)53    if len(cache) > cache_size:...

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.