Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use execute_calls method in autotest

Best Python code snippet using autotest_python

test_timestamp_file.py

Source:test_timestamp_file.py

1# Copyright 2019 Red Hat, Inc.2# All Rights Reserved.3#4#    Licensed under the Apache License, Version 2.0 (the "License"); you may5#    not use this file except in compliance with the License. You may obtain6#    a copy of the License at7#8#         http://www.apache.org/licenses/LICENSE-2.09#10#    Unless required by applicable law or agreed to in writing, software11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13#    License for the specific language governing permissions and limitations14#    under the License.15from unittest import mock16from ansible.errors import AnsibleActionFail17from ansible.errors import AnsibleActionSkip18from ansible.playbook.play_context import PlayContext19from tests import base as tests_base20from plugins.action import timestamp_file21class TestTimestampFile(tests_base.TestCase):22    def test_run(self):23        mock_task = mock.MagicMock()24        mock_task.async_val = None25        mock_task.action = "timestamp_file"26        mock_task.args = dict(path='foo.log')27        mock_connection = mock.MagicMock()28        play_context = PlayContext()29        action = timestamp_file.ActionModule(mock_task,30                                             mock_connection,31                                             play_context,32                                             None,33                                             None,34                                             None)35        mock_datetime = mock.MagicMock()36        mock_datetime.return_value = 'foo'37        action._get_date_string = mock_datetime38        mock_execute = mock.MagicMock()39        mock_execute.side_effect = [{'stat': {'exists': True}},40                                    {'stat': {'exists': False}},41                                    {'dest': 'foo.log.foo',42                                     'failed': False,43                                     'changed': True}]44        action._execute_module = mock_execute45        result = action.run()46        execute_calls = [47            mock.call(module_args={'path': 'foo.log'},48                      module_name='stat',49                      task_vars={}),50            mock.call(module_args={'path': 'foo.log.foo'},51                      module_name='stat',52                      task_vars={}),53            mock.call(module_args={'src': 'foo.log',54                                   'dest': 'foo.log.foo',55                                   'remote_src': True},56                      module_name='copy',57                      task_vars={})58        ]59        self.assertEqual(3, mock_execute.call_count)60        mock_execute.assert_has_calls(execute_calls)61        expected_result = {'dest': 'foo.log.foo', 'changed': True}62        self.assertEqual(expected_result, result)63    def test_run_source_missing_skips(self):64        mock_task = mock.MagicMock()65        mock_task.async_val = None66        mock_task.action = "timestamp_file"67        mock_task.args = dict(path='foo.log')68        mock_connection = mock.MagicMock()69        play_context = PlayContext()70        action = timestamp_file.ActionModule(mock_task,71                                             mock_connection,72                                             play_context,73                                             None,74                                             None,75                                             None)76        mock_datetime = mock.MagicMock()77        mock_datetime.return_value = 'foo'78        action._get_date_string = mock_datetime79        mock_execute = mock.MagicMock()80        mock_execute.side_effect = [{'stat': {'exists': False}}]81        action._execute_module = mock_execute82        self.assertRaises(AnsibleActionSkip, action.run)83        execute_calls = [84            mock.call(module_args={'path': 'foo.log'},85                      module_name='stat',86                      task_vars={})87        ]88        self.assertEqual(1, mock_execute.call_count)89        mock_execute.assert_has_calls(execute_calls)90    def test_run_destination_exists_fails(self):91        mock_task = mock.MagicMock()92        mock_task.async_val = None93        mock_task.action = "timestamp_file"94        mock_task.args = dict(path='foo.log')95        mock_connection = mock.MagicMock()96        play_context = PlayContext()97        action = timestamp_file.ActionModule(mock_task,98                                             mock_connection,99                                             play_context,100                                             None,101                                             None,102                                             None)103        mock_datetime = mock.MagicMock()104        mock_datetime.return_value = 'foo'105        action._get_date_string = mock_datetime106        mock_execute = mock.MagicMock()107        mock_execute.side_effect = [{'stat': {'exists': True}},108                                    {'stat': {'exists': True}}]109        action._execute_module = mock_execute110        self.assertRaises(AnsibleActionFail, action.run)111        execute_calls = [112            mock.call(module_args={'path': 'foo.log'},113                      module_name='stat',114                      task_vars={}),115            mock.call(module_args={'path': 'foo.log.foo'},116                      module_name='stat',117                      task_vars={})118        ]119        self.assertEqual(2, mock_execute.call_count)120        mock_execute.assert_has_calls(execute_calls)121    def test_run_destination_exists_force(self):122        mock_task = mock.MagicMock()123        mock_task.async_val = None124        mock_task.action = "timestamp_file"125        mock_task.args = dict(path='foo.log', force=True)126        mock_connection = mock.MagicMock()127        play_context = PlayContext()128        action = timestamp_file.ActionModule(mock_task,129                                             mock_connection,130                                             play_context,131                                             None,132                                             None,133                                             None)134        mock_datetime = mock.MagicMock()135        mock_datetime.return_value = 'foo'136        action._get_date_string = mock_datetime137        mock_execute = mock.MagicMock()138        mock_execute.side_effect = [{'stat': {'exists': True}},139                                    {'stat': {'exists': True}},140                                    {'dest': 'foo.log.foo',141                                     'failed': False,142                                     'changed': True}]143        action._execute_module = mock_execute144        result = action.run()145        execute_calls = [146            mock.call(module_args={'path': 'foo.log'},147                      module_name='stat',148                      task_vars={}),149            mock.call(module_args={'path': 'foo.log.foo'},150                      module_name='stat',151                      task_vars={}),152            mock.call(module_args={'src': 'foo.log',153                                   'dest': 'foo.log.foo',154                                   'remote_src': True},155                      module_name='copy',156                      task_vars={})157        ]158        self.assertEqual(3, mock_execute.call_count)159        mock_execute.assert_has_calls(execute_calls)160        expected_result = {'dest': 'foo.log.foo', 'changed': True}...

test_property.py

Source:test_property.py

1import unittest2from unittest.mock import patch3from utils import my_sql_connector4from services import property_services5class TestPropertyServices(unittest.TestCase):6    def setUp(self):7        self.request_args = {"city": "bogotÃ¡", "year": 2000, "status_id": 3}8    def get_patched_connector(self):9        class PatchedCursor:10            def __init__(self) -> None:11                self.execute_calls = 012                self.fetchall_calls = 013                self.info = [14                    (15                        1,16                        "calle 23 #45-67",17                        "bogotÃ¡",18                        120000000,19                        "Hermoso apartamento en el centro de la ciudad",20                        2000,21                        "3",22                    ),23                    (24                        53,25                        "calle 23 #45-67q",26                        "bogotÃ¡",27                        120000000,28                        "Hermoso apartamento en el centro de la ciudad",29                        2000,30                        "3",31                    ),32                    (33                        2,34                        "calle 2 #45-67",35                        "medellin",36                        120000000,37                        "Hermoso apartamento en el centro de la ciudad",38                        2010,39                        "4",40                    ),41                    (42                        54,43                        "calle 23 #45-67q",44                        "bogotÃ¡",45                        120000000,46                        "Hermoso apartamento en el centro de la ciudad",47                        2011,48                        "5",49                    ),50                ]51                self.filtered_response = self.info.copy()52            def execute(self, sql: str):53                self.execute_calls += 154                self.where = sql.split("where")[1].strip()55                extra_filters = self.where.split("and")56                extra_filters.pop(0)57                self.filtered_response = self.info.copy()58                for extra_filter in extra_filters:59                    extra_filter.strip()60                    value = extra_filter.split("=")[1].strip()61                    if "city" in extra_filter:62                        self.filtered_response = list(63                            filter(64                                lambda i: i[2] == value.replace("'", ""),65                                self.filtered_response,66                            )67                        )68                    if "year" in extra_filter:69                        self.filtered_response = list(70                            filter(lambda i: i[5] == int(value), self.filtered_response)71                        )72                    if "status_id" in extra_filter:73                        self.filtered_response = list(74                            filter(lambda i: i[6] == value, self.filtered_response)75                        )76            def fetchall(self):77                self.fetchall_calls += 178                return self.filtered_response79        class PatchedConnector:80            def __init__(self) -> None:81                self.cursor = PatchedCursor()82                self.close_all_calls = 083            def close_all(self) -> None:84                self.close_all_calls += 185                pass86        self.connector = PatchedConnector()87        return self.connector88    def test_find_available_houses(self):89        with patch.object(my_sql_connector, "DatabaseHandler") as mocked_connector:90            mocked_connector.side_effect = self.get_patched_connector91            # No args call92            response = property_services.find_available_houses()93            self.assertEqual(94                self.connector.cursor.where, "status_id in ('3', '4', '5')"95            )96            self.assertEqual(self.connector.cursor.execute_calls, 1)97            self.assertEqual(self.connector.cursor.fetchall_calls, 1)98            self.assertEqual(self.connector.close_all_calls, 1)99            self.assertEqual(len(response), 4)100            # City arg call101            response = property_services.find_available_houses({"city": "bogotÃ¡"})102            self.assertEqual(103                self.connector.cursor.where,104                "status_id in ('3', '4', '5') and city = 'bogotÃ¡'",105            )106            self.assertEqual(self.connector.cursor.execute_calls, 1)107            self.assertEqual(self.connector.cursor.fetchall_calls, 1)108            self.assertEqual(self.connector.close_all_calls, 1)109            self.assertEqual(len(response), 3)110            # Year arg call111            response = property_services.find_available_houses({"year": 2011})112            self.assertEqual(113                self.connector.cursor.where,114                "status_id in ('3', '4', '5') and year = 2011",115            )116            self.assertEqual(self.connector.cursor.execute_calls, 1)117            self.assertEqual(self.connector.cursor.fetchall_calls, 1)118            self.assertEqual(self.connector.close_all_calls, 1)119            self.assertEqual(len(response), 1)120            # Status arg call121            response = property_services.find_available_houses({"status_id": 4})122            self.assertEqual(123                self.connector.cursor.where,124                "status_id in ('3', '4', '5') and status_id = 4",125            )126            self.assertEqual(self.connector.cursor.execute_calls, 1)127            self.assertEqual(self.connector.cursor.fetchall_calls, 1)128            self.assertEqual(self.connector.close_all_calls, 1)129            self.assertEqual(len(response), 1)130            # All args call131            response = property_services.find_available_houses(self.request_args)132            self.assertEqual(133                self.connector.cursor.where,134                "status_id in ('3', '4', '5') and city = 'bogotÃ¡' and year = 2000 and status_id = 3",135            )136            self.assertEqual(self.connector.cursor.execute_calls, 1)137            self.assertEqual(self.connector.cursor.fetchall_calls, 1)138            self.assertEqual(self.connector.close_all_calls, 1)...

test_db.py

Source:test_db.py

...17        self._cursor.fetchall.return_value = [(EXPLAIN,)]18    def cursor(self):19        return self._cursor20    @property21    def execute_calls(self):22        return self._cursor.execute.call_args_list23# all of these should still be valid operations we should proxy through24# to the underlying deque object25def test_deque_appendleft():26    log = RecordedQueryLog(collections.deque(), FakeDatabase)27    log.appendleft(QUERY)28    assert len(log) == 129def test_deque_count():30    log = RecordedQueryLog(collections.deque(), FakeDatabase)31    log.append(QUERY)32    assert log.count('x') == 033    assert log.count(QUERY) == 134def test_deque_clear():35    log = RecordedQueryLog(collections.deque(), FakeDatabase)...

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.