KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use with_env method in testcontainers-python

Best Python code snippet using testcontainers-python_python

kafka.py

Source:kafka.py

...17    def __init__(self, image=DEFAULT_IMAGE, port_to_expose=DEFAULT_PORT):18        super(KafkaContainer, self).__init__(image)19        self.port_to_expose = port_to_expose20        self.with_exposed_ports(self.port_to_expose)21        self.with_env(22            "KAFKA_LISTENERS",23            "PLAINTEXT://0.0.0.0:{},BROKER://0.0.0.0:9092".format(port_to_expose),24        )25        self.with_env(26            "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP",27            "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT",28        )29        self.with_env("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER")30        self.with_env("KAFKA_BROKER_ID", "1")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"""...

test_account_move_partner_count.py

Source:test_account_move_partner_count.py

...9        # T1 Create partner and account move10        # (Use a new cursor to make sure creations are committed for later transactions)11        with self.registry.cursor() as cr:12            env = self.env(cr=cr)13            self.partner = self.env['res.partner'].with_env(env).create({'name': 'Lucien'})14            user_type_income = env.ref('account.data_account_type_direct_costs')15            self.account_income = self.env['account.account'].with_env(env).create({16                'code': 'TESTCOUNT42',17                'name': 'Sale - Test Account',18                'user_type_id': user_type_income.id19            })20            self.journal = self.env['account.journal'].with_env(env).create({21                'name': 'Journal',22                'code': '7775',23                'type': 'sale',24            })25            self.move = self.env['account.move'].with_env(env).create({26                'partner_id': self.partner.id,27                'type': 'out_invoice',28                'journal_id': self.journal.id,29                'currency_id': self.env.user.company_id.currency_id.id,30            })31            self.line = self.env['account.move.line'].with_env(env).create({32                'move_id': self.move.id,33                'name': 'account move line test',34                'account_id': self.account_income.id,35            })36    def tearDown(self):37        super().tearDown()38        with self.registry.cursor() as cr:39            env = self.env(cr=cr)40            move = self.move.with_env(env)41            move.name = '/'42            move.state = 'draft'43            move.unlink()44            self.journal.with_env(env).unlink()45            self.account_income.with_env(env).unlink()46            self.partner.with_env(env).unlink()47    def test_account_move_count(self):48        # T2 create and post account move49        with self.registry.cursor() as cr:50            env = self.env(cr=cr)51            self.move.with_env(env).post()52            partner = self.partner.with_env(env)53            self.assertEqual(partner.supplier_rank, 0)54            self.assertEqual(partner.customer_rank, 1)55    def test_account_move_count_concurrent(self):56        # T2 Lock the partner row57        with self.registry.cursor() as cr:58            cr.execute("""SELECT id FROM res_partner WHERE id = %s FOR UPDATE""" % self.partner.id)59            # T3 concurrently posts account move and tries to update partner60            with self.registry.cursor() as cr:61                env = self.env(cr=cr)62                self.move.with_env(env).state = 'posted'63                self.assertEqual(self.partner.with_env(env).customer_rank,...

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.