Best Python code snippet using autotest_python
test_locking.py
Source:test_locking.py  
...132    def test(self, connections_mock):133        cursor = connections_mock.set_results((1,), (1,))134        with locking.AdvisoryLock('pudding') as lock:135            self.assertEqual(lock.fullname, "FlippinDatabase:pudding")136            cursor.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", ("FlippinDatabase:pudding", 3))137        cursor.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", ("FlippinDatabase:pudding",))138        self.assertEqual(cursor.execute.call_count, 2)139    def test_nesting_disallowed(self, connections_mock):140        cursor = connections_mock.set_results((1,), (1,), (1,), (1,)) # more than we need, for failure case141        with locking.AdvisoryLock('pudding'):142            cursor.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", ("FlippinDatabase:pudding", 3))143            with self.assertRaises(locking.LockError) as context:144                with locking.AdvisoryLock('cake'):145                    pass146        self.assertEqual(context.exception.args, ("AdvisoryLock-managed contexts may not be nested",))147        cursor.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", ("FlippinDatabase:pudding",))148        self.assertEqual(cursor.execute.call_count, 2)149    def test_nesting_same_disallowed(self, connections_mock):150        cursor = connections_mock.set_results((1,), (1,), (1,), (1,)) # more than we need, for failure case151        with locking.AdvisoryLock('pudding'):152            cursor.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", ("FlippinDatabase:pudding", 3))153            with self.assertRaises(locking.LockError) as context:154                with locking.AdvisoryLock('pudding'):155                    pass156        self.assertEqual(context.exception.args, ("AdvisoryLock-managed contexts may not be nested",))157        cursor.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", ("FlippinDatabase:pudding",))158        self.assertEqual(cursor.execute.call_count, 2)159    def test_nesting_multidb_allowed(self, connections_mock):160        cursor_default = connections_mock.set_results((1,), (1,), using='default')161        cursor_alt = connections_mock.set_results((1,), (1,), using='thisdb')162        with locking.AdvisoryLock('pudding'):163            cursor_default.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", ("FlippinDatabase:pudding", 3))164            with locking.AdvisoryLock('cake', using='thisdb'):165                cursor_alt.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", ("AnotherDatabase:cake", 3))166        cursor_default.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", ("FlippinDatabase:pudding",))167        cursor_alt.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", ("AnotherDatabase:cake",))168        self.assertEqual(cursor_default.execute.call_count, 2)169        self.assertEqual(cursor_alt.execute.call_count, 2)170@patch_connections171class TestAdvisoryLockDecorator(TestCase):172    def test(self, connections_mock):173        cursor = connections_mock.set_results((1,), (1,))174        @locking.AdvisoryLock('pudding')175        def divider(a, b):176            return a / b177        with self.assertRaises(ZeroDivisionError):178            divider(1, 0)179        cursor.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", ("FlippinDatabase:pudding", 3))180        cursor.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", ("FlippinDatabase:pudding",))181        self.assertEqual(cursor.execute.call_count, 2)182class TestLockHelper(TestCase):183    def test_proxy(self):184        lock = locking.lock('pudding', using='thisdb', timeout=10)185        self.assertEqual(lock.nickname, 'pudding')186        self.assertEqual(lock.using, 'thisdb')187        self.assertEqual(lock.timeout, 10)188    @patch_connections189    def test_decorator(self, connections_mock):190        cursor = connections_mock.set_results((1,), (1,), using='thisdb')191        wrapped = locking.lock(lambda: 1 / 0, 'pudding', using='thisdb')192        with self.assertRaises(ZeroDivisionError):193            wrapped()194        cursor.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", ("AnotherDatabase:pudding", 3))195        cursor.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", ("AnotherDatabase:pudding",))196        self.assertEqual(cursor.execute.call_count, 2)197    @patch_connections198    def test_bare_decorator(self, connections_mock):199        cursor = connections_mock.set_results((1,), (1,))200        @locking.lock201        def divider(a, b):202            return a / b203        with self.assertRaises(ZeroDivisionError):204            divider(1, 0)205        name = "FlippinDatabase:core.tests.test_locking:divider"206        cursor.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", (name, 3))207        cursor.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", (name,))208        self.assertEqual(cursor.execute.call_count, 2)209    @patch_connections210    def test_named_method_decorator(self, connections_mock):211        cursor = connections_mock.set_results((1,), (1,))212        class Objectified(object):213            @classmethod214            @locking.lock('pudding')215            def races(cls, a, b):216                return a / b217        with self.assertRaises(ZeroDivisionError):218            Objectified.races(1, 0)219        cursor.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", ("FlippinDatabase:pudding", 3))220        cursor.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", ("FlippinDatabase:pudding",))221        self.assertEqual(cursor.execute.call_count, 2)222class TestLockMethodHelper(TestCase):223    @patch_connections224    def test_instance_decorator(self, connections_mock):225        cursor = connections_mock.set_results((1,), (1,), using='thisdb')226        class Objectified(object):227            @locking.lockingmethod(using='thisdb')228            def races(self, a, b):229                return a / b230        obj = Objectified()231        with self.assertRaises(ZeroDivisionError):232            obj.races(1, 0)233        name = "AnotherDatabase:core.tests.test_locking:Objectified.races"234        cursor.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", (name, 3))235        cursor.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", (name,))236        self.assertEqual(cursor.execute.call_count, 2)237    @patch_connections238    def test_class_decorator(self, connections_mock):239        cursor = connections_mock.set_results((1,), (1,), using='thisdb')240        class Objectified(object):241            @classmethod242            @locking.lockingmethod(using='thisdb')243            def races(cls, a, b):244                return a / b245        with self.assertRaises(ZeroDivisionError):246            Objectified.races(1, 0)247        name = "AnotherDatabase:core.tests.test_locking:Objectified.races"248        cursor.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", (name, 3))249        cursor.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", (name,))250        self.assertEqual(cursor.execute.call_count, 2)251    @patch_connections252    def test_bare_decorator(self, connections_mock):253        cursor = connections_mock.set_results((1,), (1,))254        class Objectified(object):255            @classmethod256            @locking.lockingmethod257            def races(cls, a, b):258                return a / b259        with self.assertRaises(ZeroDivisionError):260            Objectified.races(1, 0)261        name = "FlippinDatabase:core.tests.test_locking:Objectified.races"262        cursor.execute.assert_any_call("SELECT GET_LOCK(%s, %s)", (name, 3))263        cursor.execute.assert_any_call("SELECT RELEASE_LOCK(%s)", (name,))...builder_test.py
Source:builder_test.py  
...127        with self._patch() as patched:128            patched.docker.build_docker_image.return_value = docker_build129            return_val = self.builder.build()130            self.assertTrue(return_val)131            patched.shutil.copy.assert_any_call(f'{library}/Dockerfile', working_dir)132            patched.shutil.copy.assert_any_call(f'{library}/{file1.path}', working_dir)133            patched.shutil.copy.assert_any_call(f'{file2.path}', working_dir)134            patched.changeLoadCustom.assert_called_with(custom_commands)135            patched.changeDir.assert_any_call(working_dir)136            patched.changeDir.assert_any_call(working_dir, not_store=True)137            patched.registry.get_command.assert_any_call(step1.command)138            patched.registry.get_command.assert_any_call(step2.command)139            self.assertTrue(self.test1_command.executed)140            self.assertTrue(self.test2_command.executed)141            self.assertTrue(self.test3_command.executed)142            self.assertEqual(step1.arguments, self.test1_command.args)143            self.assertEqual(step2.arguments, self.test2_command.args)144            self.assertEqual(step3.arguments, self.test3_command.args)145            patched.docker.build_docker_image.assert_called_with(image)146            self.builder._working_directory.cleanup.assert_called()147            patched.changeBack.assert_called()148            # assert info messages149            patched.info.assert_any_call('Copying Dockerfile and required files to build directory')150            patched.info.assert_any_call('Copying Dockerfile Dockerfile from library to build directory')151            patched.info.assert_any_call(f'Copying file {file1.path} from library to build directory')152            patched.info.assert_any_call(f'Copying file {file2.path} to build directory')153            patched.info.assert_any_call('Dockerfile and required files successfully copied to build directory')154            patched.info.assert_any_call(f'Build specified custom commands file {custom_commands}. '155                                         'Loading commands into build')156            patched.info.assert_any_call('Changing working directory to build directory')157            patched.info.assert_any_call('Executing build steps')158            patched.info.assert_any_call(f'Executing build step 1 - {step1.name}')159            patched.info.assert_any_call('Executing build step 2 - name')160            patched.info.assert_any_call(f'Building Docker image with tag {image}')161            patched.info.assert_any_call(f'Docker image with tag {image} built successfully with the following output:')162            patched.info.assert_any_call(f'\tstdout')163            patched.info.assert_any_call('Executing post-build steps')164            patched.info.assert_any_call(f'Executing post-build step 1 - {step3.name}')165            patched.info.assert_any_call('Build finished, changing back to working directory')166    def test_successful_build_without_custom_commands(self):167        docker_build = ExecutionResult(0, 'stdout', '')168        patched: PatchedDependencies169        with self._patch() as patched:170            patched.docker.build_docker_image.return_value = docker_build171            self.builder.config.custom_commands = None172            return_val = self.builder.build()173            self.assertTrue(return_val)174            patched.changeLoadCustom.assert_not_called()175    def test_failed_build_unknown_command(self):176        docker_build = ExecutionResult(0, 'stdout', '')177        patched: PatchedDependencies178        with self._patch() as patched:179            patched.docker.build_docker_image.return_value = docker_build180            unknown_step = models.BuildStep()181            unknown_step.command = 'unknown'182            self.builder.config.steps = [unknown_step]183            with self.assertRaises(dockerwizard.errors.BuildConfigurationError) as e:184                self.builder.build()185            self.assertTrue('Unknown command' in e.exception.message)186    def test_failed_build_command_error(self):187        docker_build = ExecutionResult(0, 'stdout', '')188        patched: PatchedDependencies189        with self._patch() as patched:190            patched.docker.build_docker_image.return_value = docker_build191            self.test1_command.throw_error = True192            return_val = self.builder.build()193            self.assertFalse(return_val)194            patched.error.assert_any_call(f'Failed to execute build step 1 - {step1.name} with error: error')195            patched.error.assert_any_call('See logs to see why the build failed')196    def test_failed_build_docker_error(self):197        failed_docker = ExecutionResult(1, '', 'error')198        patched: PatchedDependencies199        with self._patch() as patched:200            patched.docker.build_docker_image.return_value = failed_docker201            return_val = self.builder.build()202            self.assertFalse(return_val)203            patched.error.assert_any_call('Failed to build Docker image with error error and '204                                          'exit code 1')205            patched.error.assert_any_call('See logs to see why the build failed')206    def test_context_setup(self):207        mock_context = StubContext()208        docker_build = ExecutionResult(0, 'stdout', '')209        patched: PatchedDependencies210        with self._patch() as patched:211            patched.get('context_init').return_value = mock_context212            self._create_builder()213            patched.docker.build_docker_image.return_value = docker_build214            self.builder.build()215            self.assertEqual(self.builder.config, mock_context.config)216            self.assertTrue(mock_context.current_step_set)217            patched.get('context_teardown').assert_called()218if __name__ == '__main__':219    main()telemetry_utils_test.py
Source:telemetry_utils_test.py  
...32        node2['sub_graphs'] = ['sub_graph2']33        node2['sub_graph2'] = sub_graph234        self.init_telemetry_mocks()35        send_op_names_info('framework', graph)36        tm.Telemetry.send_event.assert_any_call('mo', 'op_count', 'framework_a', 5)37        tm.Telemetry.send_event.assert_any_call('mo', 'op_count', 'framework_b', 2)38        tm.Telemetry.send_event.assert_any_call('mo', 'op_count', 'framework_c', 2)39        tm.Telemetry.send_event.assert_any_call('mo', 'op_count', 'framework_d', 1)40    def test_send_shapes_info(self):41        graph = build_graph({**regular_op('placeholder1', {'shape': int64_array([1, 3, 20, 20]), 'type': 'Parameter'}),42                             **regular_op('placeholder2', {'shape': int64_array([2, 4, 10]), 'type': 'Parameter'}),43                             **regular_op('mul', {'shape': int64_array([7, 8]), 'type': 'Multiply'})}, [])44        self.init_telemetry_mocks()45        send_shapes_info('framework', graph)46        tm.Telemetry.send_event.assert_any_call('mo', 'input_shapes', '{fw:framework,shape:"[ 1  3 20 20],[ 2  4 10]"}')47        tm.Telemetry.send_event.assert_any_call('mo', 'partially_defined_shape',48                                                '{partially_defined_shape:0,fw:framework}')49    def test_send_dynamic_shapes_case1(self):50        graph = build_graph({**regular_op('placeholder1', {'shape': int64_array([-1, 3, 20, 20]), 'type': 'Parameter'}),51                             **regular_op('mul', {'shape': int64_array([7, 8]), 'type': 'Multiply'})}, [])52        self.init_telemetry_mocks()53        send_shapes_info('framework', graph)54        tm.Telemetry.send_event.assert_any_call('mo', 'input_shapes', '{fw:framework,shape:"[-1  3 20 20]"}')55        tm.Telemetry.send_event.assert_any_call('mo', 'partially_defined_shape',56                                                '{partially_defined_shape:1,fw:framework}')57    def test_send_dynamic_shapes_case2(self):58        graph = build_graph({**regular_op('placeholder1', {'shape': int64_array([2, 3, 20, 20]), 'type': 'Parameter'}),59                             **regular_op('placeholder2', {'shape': int64_array([7, 4, 10]), 'type': 'Parameter'}),60                             **regular_op('placeholder3', {'shape': int64_array([5, 4, 0]), 'type': 'Parameter'}),61                             **regular_op('mul', {'shape': int64_array([7, 8]), 'type': 'Multiply'})}, [])62        self.init_telemetry_mocks()63        send_shapes_info('framework', graph)64        tm.Telemetry.send_event.assert_any_call('mo', 'input_shapes',65                                                '{fw:framework,shape:"[ 2  3 20 20],[ 7  4 10],[5 4 0]"}')66        tm.Telemetry.send_event.assert_any_call('mo', 'partially_defined_shape',...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!!
