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

How to use wait_for_logs method in testcontainers-python

Best Python code snippet using testcontainers-python_python

test_event_log.py

Source:test_event_log.py

...48    # Since the logging is all async, the tests would be a nightmare without49    # enforcing events to be logged serially as they happen.50    # In order to do this, we wait by poking at the DB in a loop until the51    # expected number of records exists in the DB.52    def wait_for_logs(self, game_id, expected_nr):53        c = self.db.cursor()54        while True:55            time.sleep(0.005)56            c.execute('SELECT COUNT(*) FROM event_logs WHERE game_id = ?', [game_id])57            if c.fetchone()[0] == expected_nr:58                break59    @defer.inlineCallbacks60    def test01_log_game_events(self):61        owner_id = 4262        player1_id = 8463        player2_id = 8664        sentence = 'The Sentence.'65        # 1 - Should log creation.66        game_id = yield self.game.create(owner_id)67        self.wait_for_logs(game_id, 1)68        # 2 - Should log card set.69        owner = yield self.game.player2game(owner_id)70        winner_card = owner['cards'][0]71        yield self.game.set_card(owner_id, winner_card)72        self.wait_for_logs(game_id, 2)73        # 3 -Should log sentence set.74        yield self.game.set_sentence(owner_id, sentence)75        self.wait_for_logs(game_id, 3)76        # 4 - Should log invitation (player1 only).77        yield self.game.invite([player1_id])78        self.wait_for_logs(game_id, 4)79        # 5 & 6 - Should log joining (both players).80        yield self.game.participate(player1_id)81        self.wait_for_logs(game_id, 5)82        yield self.game.participate(player2_id)83        self.wait_for_logs(game_id, 6)84        # 7 & 8 - Should log players picking cards.85        player1 = yield self.game.player2game(player1_id)86        picked_card1 = player1['cards'][0]87        yield self.game.pick(player1_id, picked_card1)88        self.wait_for_logs(game_id, 7)89        player2 = yield self.game.player2game(player2_id)90        picked_card2 = player1['cards'][0]91        yield self.game.pick(player2_id, picked_card2)92        self.wait_for_logs(game_id, 8)93        # 9 - Should log game moved into voting.94        yield self.game.voting(owner_id)95        self.wait_for_logs(game_id, 9)96        # 10 & 11 - Should log players voting for cards.97        yield self.game.vote(player1_id, winner_card)98        self.wait_for_logs(game_id, 10)99        yield self.game.vote(player2_id, picked_card1)100        self.wait_for_logs(game_id, 11)101        # 12 - Should log game completed.102        yield self.game.complete(owner_id)103        self.wait_for_logs(game_id, 12)104        # Now lets check the logs table.105        c = self.db.cursor()106        # Wait until the logs are written (they're written asynchronously)107        c.execute('SELECT event_type, player_id, data FROM event_logs WHERE game_id = ? ORDER BY timestamp', [game_id])108        # Twelve events should be logged.109        rows = c.fetchall()110        self.assertEqual((event_log.GAME_CREATED, owner_id, None), rows[0])111        self.assertEqual((event_log.OWNER_CHOSE_CARD, owner_id, str(winner_card)), rows[1])112        self.assertEqual((event_log.OWNER_WROTE_STORY, owner_id, sentence), rows[2])113        self.assertEqual((event_log.PLAYER_INVITED, owner_id, str(player1_id)), rows[3])114        self.assertEqual((event_log.PLAYER_JOINED, player1_id, None), rows[4])115        self.assertEqual((event_log.PLAYER_JOINED, player2_id, None), rows[5])116        self.assertEqual((event_log.PLAYER_PICKED_CARD, player1_id, str(picked_card1)), rows[6])117        self.assertEqual((event_log.PLAYER_PICKED_CARD, player2_id, str(picked_card2)), rows[7])118        self.assertEqual((event_log.GAME_MOVED_TO_VOTING, owner_id, None), rows[8])119        self.assertEqual((event_log.PLAYER_VOTED, player1_id, str(winner_card)), rows[9])120        self.assertEqual((event_log.PLAYER_VOTED, player2_id, str(picked_card1)), rows[10])121        self.assertEqual((event_log.GAME_COMPLETED, owner_id, None), rows[11])122        c.close()123    @defer.inlineCallbacks124    def test02_log_canceled_game_events(self):125        owner_id = 4126        player_id = 8127        sentence = 'Some Sentence.'128        # 1 - Should log creation.129        game_id = yield self.game.create(owner_id)130        self.wait_for_logs( game_id, 1)131        # 2 - Should log card set.132        owner = yield self.game.player2game(owner_id)133        winner_card = owner['cards'][0]134        yield self.game.set_card(owner_id, winner_card)135        self.wait_for_logs( game_id, 2)136        # 3 -Should log sentence set.137        yield self.game.set_sentence(owner_id, sentence)138        self.wait_for_logs( game_id, 3)139        # 4 - Should log joining.140        yield self.game.participate(player_id)141        self.wait_for_logs( game_id, 4)142        # 5 - Should log player leaving.143        yield self.game.leave([player_id])144        self.wait_for_logs( game_id, 5)145        # 6 - Should log game cancelation.146        yield self.game.cancel()147        self.wait_for_logs(game_id, 6)148        # Now lets check the logs table.149        c = self.db.cursor()150        c.execute('SELECT event_type, player_id, data FROM event_logs WHERE game_id = ? ORDER BY timestamp', [game_id])151        # Twelve events should be logged.152        rows = c.fetchall()153        self.assertEqual((event_log.GAME_CREATED, owner_id, None), rows[0])154        self.assertEqual((event_log.OWNER_CHOSE_CARD, owner_id, str(winner_card)), rows[1])155        self.assertEqual((event_log.OWNER_WROTE_STORY, owner_id, sentence), rows[2])156        self.assertEqual((event_log.PLAYER_JOINED, player_id, None), rows[3])157        self.assertEqual((event_log.PLAYER_LEFT, player_id, None), rows[4])158        self.assertEqual((event_log.GAME_CANCELED, owner_id, None), rows[5])159        c.close()160    def test03_log_query_functions(self):161        c = self.db.cursor()...

functional_test_tls_docker_network.py

Source:functional_test_tls_docker_network.py

...23        """24        first = DockerContainer(image='nginx:1.19').with_env('INFR_VIRTUAL_HOST', 'duckduckgo.com,bing.com').start()25        second = DockerContainer(image='nginx:1.19').with_env('INFR_VIRTUAL_HOST', 'google.com').start()26        try:27            wait_for_logs(first, 'ready for start up')28            wait_for_logs(second, 'ready for start up')29            os.environ['PATH'] = path + ':' + os.environ['PATH']30            check = TlsDockerNetworkCheck(param_type='environment', param_name='INFR_VIRTUAL_HOST', alert_days_before=1)\31                .main()32        finally:33            first.stop()34            second.stop()35        self.assertIn('Domain google.com is OK', check[0])36        self.assertIn('Domain bing.com is OK', check[0])37        self.assertIn('Domain duckduckgo.com is OK', check[0])38        self.assertTrue(check[1])39    def test_containers_matched_by_label(self):40        """41        There are two docker containers. Matching is by label org.riotkit.domain: {{ domain }}42        Checks:43            - Parsing of the labels syntax44        """45        first = DockerContainer(image='nginx:1.19').with_kwargs(labels={'org.riotkit.domain': 'duckduckgo.com'}).start()46        second = DockerContainer(image='nginx:1.19')\47            .with_kwargs(labels={'org.riotkit.domain': 'riseup.net,bing.com'}).start()48        try:49            wait_for_logs(first, 'ready for start up')50            wait_for_logs(second, 'ready for start up')51            os.environ['PATH'] = path + ':' + os.environ['PATH']52            check = TlsDockerNetworkCheck(param_type='label', param_name='org.riotkit.domain', alert_days_before=1) \53                .main()54        finally:55            first.stop()56            second.stop()57        self.assertIn('Domain duckduckgo.com is OK', check[0])58        self.assertIn('Domain bing.com is OK', check[0])59        self.assertIn('Domain riseup.net is OK', check[0])60        self.assertTrue(check[1])61    def test_no_domains_found_will_result_in_error_status(self):62        check = TlsDockerNetworkCheck(param_type='label', param_name='invalid-label-name', alert_days_before=1) \63            .main()64        self.assertEqual('No any domains found, maybe the containers are not running?', check[0])...

conftest.py

Source:conftest.py

1import logging2import pytest3from src import cylinder4@pytest.fixture()5def foo_site_app():6    def dir_map_func(request):  # pylint: disable=unused-argument7        return "test_sites", "foo_site"8    app = cylinder.get_app(dir_map_func, logging.DEBUG)9    app.wait_for_logs = True10    # other setup can go here11    return app12    # clean up / reset resources here13@pytest.fixture()14def minimum_site_app():15    def dir_map_func(request):  # pylint: disable=unused-argument16        return "test_sites", "minimum_site"17    app = cylinder.get_app(dir_map_func, logging.DEBUG)18    app.wait_for_logs = True19    return app20@pytest.fixture()21def no_hook_fail_site_app():22    def dir_map_func(request):  # pylint: disable=unused-argument23        return "test_sites", "no_hook_fail_site"24    app = cylinder.get_app(dir_map_func, logging.DEBUG)25    app.wait_for_logs = True26    return app27@pytest.fixture()28def tiny_queue_app():29    def dir_map_func(request):  # pylint: disable=unused-argument30        return "test_sites", "minimum_site"31    app = cylinder.get_app(dir_map_func, logging.DEBUG, log_queue_length=1)32    app.wait_for_logs = True33    return app34@pytest.fixture()35def foo_site_client(foo_site_app):  # pylint: disable=redefined-outer-name36    return foo_site_app.test_client()37@pytest.fixture()38def minimum_site_client(minimum_site_app):  # pylint: disable=redefined-outer-name39    return minimum_site_app.test_client()40@pytest.fixture()41def no_hook_fail_site_client(42    no_hook_fail_site_app,43):  # pylint: disable=redefined-outer-name44    return no_hook_fail_site_app.test_client()45@pytest.fixture()46def tiny_queue_client(tiny_queue_app):  # pylint: disable=redefined-outer-name...

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.