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

How to use create_mock_function method in autotest

Best Python code snippet using autotest_python

subcommand_unittest.py

Source:subcommand_unittest.py

...56        self.god.check_playback()57    def _setup_fork_start_parent(self):58        self.god.stub_function(subcommand.os, 'fork')59        subcommand.os.fork.expect_call().and_return(1000)60        func = self.god.create_mock_function('func')61        cmd = _create_subcommand(func, [])62        cmd.fork_start()63        return cmd64    def test_fork_start_parent(self):65        cmd = self._setup_fork_start_parent()66        self.assertEquals(cmd.pid, 1000)67        self.god.check_playback()68    def _setup_fork_start_child(self):69        self.god.stub_function(subcommand.os, 'pipe')70        self.god.stub_function(subcommand.os, 'fork')71        self.god.stub_function(subcommand.os, 'close')72        self.god.stub_function(subcommand.os, 'write')73        self.god.stub_function(subcommand.cPickle, 'dumps')74        self.god.stub_function(subcommand.os, '_exit')75    def test_fork_start_child(self):76        self._setup_fork_start_child()77        func = self.god.create_mock_function('func')78        fork_hook = self.god.create_mock_function('fork_hook')79        join_hook = self.god.create_mock_function('join_hook')80        subcommand.subcommand.register_fork_hook(fork_hook)81        subcommand.subcommand.register_join_hook(join_hook)82        cmd = _create_subcommand(func, (1, 2))83        subcommand.os.pipe.expect_call().and_return((10, 20))84        subcommand.os.fork.expect_call().and_return(0)85        subcommand.os.close.expect_call(10)86        fork_hook.expect_call(cmd)87        func.expect_call(1, 2).and_return(True)88        subcommand.cPickle.dumps.expect_call(True,89                subcommand.cPickle.HIGHEST_PROTOCOL).and_return('True')90        subcommand.os.write.expect_call(20, 'True')91        subcommand.os.close.expect_call(20)92        join_hook.expect_call(cmd)93        subcommand.os._exit.expect_call(0)94        cmd.fork_start()95        self.god.check_playback()96    def test_fork_start_child_error(self):97        self._setup_fork_start_child()98        self.god.stub_function(subcommand.logging, 'exception')99        func = self.god.create_mock_function('func')100        cmd = _create_subcommand(func, (1, 2))101        error = Exception('some error')102        subcommand.os.pipe.expect_call().and_return((10, 20))103        subcommand.os.fork.expect_call().and_return(0)104        subcommand.os.close.expect_call(10)105        func.expect_call(1, 2).and_raises(error)106        subcommand.logging.exception.expect_call('function failed')107        subcommand.cPickle.dumps.expect_call(error,108                subcommand.cPickle.HIGHEST_PROTOCOL).and_return('error')109        subcommand.os.write.expect_call(20, 'error')110        subcommand.os.close.expect_call(20)111        subcommand.os._exit.expect_call(1)112        cmd.fork_start()113        self.god.check_playback()114    def _setup_poll(self):115        cmd = self._setup_fork_start_parent()116        self.god.stub_function(subcommand.os, 'waitpid')117        return cmd118    def test_poll_running(self):119        cmd = self._setup_poll()120        (subcommand.os.waitpid.expect_call(1000, subcommand.os.WNOHANG)121                .and_raises(subcommand.os.error('waitpid')))122        self.assertEquals(cmd.poll(), None)123        self.god.check_playback()124    def test_poll_finished_success(self):125        cmd = self._setup_poll()126        (subcommand.os.waitpid.expect_call(1000, subcommand.os.WNOHANG)127                .and_return((1000, 0)))128        self.assertEquals(cmd.poll(), 0)129        self.god.check_playback()130    def test_poll_finished_failure(self):131        cmd = self._setup_poll()132        self.god.stub_function(cmd, '_handle_exitstatus')133        (subcommand.os.waitpid.expect_call(1000, subcommand.os.WNOHANG)134                .and_return((1000, 10)))135        cmd._handle_exitstatus.expect_call(10).and_raises(Exception('fail'))136        self.assertRaises(Exception, cmd.poll)137        self.god.check_playback()138    def test_wait_success(self):139        cmd = self._setup_poll()140        (subcommand.os.waitpid.expect_call(1000, 0)141                .and_return((1000, 0)))142        self.assertEquals(cmd.wait(), 0)143        self.god.check_playback()144    def test_wait_failure(self):145        cmd = self._setup_poll()146        self.god.stub_function(cmd, '_handle_exitstatus')147        (subcommand.os.waitpid.expect_call(1000, 0)148                .and_return((1000, 10)))149        cmd._handle_exitstatus.expect_call(10).and_raises(Exception('fail'))150        self.assertRaises(Exception, cmd.wait)151        self.god.check_playback()152class real_subcommand_test(unittest.TestCase):153    """Test actually running subcommands (without mocking)."""154    def _setup_subcommand(self, func, *args):155        cmd = subcommand.subcommand(func, args)156        cmd.fork_start()157        return cmd158    def test_fork_waitfor_no_timeout(self):159        """Test fork_waitfor success with no timeout."""160        cmd = self._setup_subcommand(lambda: None)161        self.assertEquals(cmd.fork_waitfor(), 0)162    def test_fork_waitfor_timeout(self):163        """Test fork_waitfor success with a timeout."""164        cmd = self._setup_subcommand(lambda: None)165        self.assertEquals(cmd.fork_waitfor(timeout=60), 0)166    def test_fork_waitfor_exception(self):167        """Test fork_waitfor failure with an exception."""168        cmd = self._setup_subcommand(lambda: None, 'foo')169        with self.assertRaises(error.AutoservSubcommandError):170          cmd.fork_waitfor(timeout=60)171    def test_fork_waitfor_timeout_fail(self):172        """Test fork_waitfor timing out."""173        cmd = self._setup_subcommand(lambda: time.sleep(60))174        with self.assertRaises(error.AutoservSubcommandError):175          cmd.fork_waitfor(timeout=1)176class parallel_test(unittest.TestCase):177    def setUp(self):178        self.god = mock.mock_god()179        self.god.stub_function(subcommand.cPickle, 'load')180    def tearDown(self):181        self.god.unstub_all()182    def _get_cmd(self, func, args):183        cmd = _create_subcommand(func, args)184        cmd.result_pickle = self.god.create_mock_class(file, 'file')185        return self.god.create_mock_class(cmd, 'subcommand')186    def _get_tasklist(self):187        return [self._get_cmd(lambda x: x * 2, (3,)),188                self._get_cmd(lambda: None, [])]189    def _setup_common(self):190        tasklist = self._get_tasklist()191        for task in tasklist:192            task.fork_start.expect_call()193        return tasklist194    def test_success(self):195        tasklist = self._setup_common()196        for task in tasklist:197            task.fork_waitfor.expect_call(timeout=None).and_return(0)198            (subcommand.cPickle.load.expect_call(task.result_pickle)199                    .and_return(6))200            task.result_pickle.close.expect_call()201        subcommand.parallel(tasklist)202        self.god.check_playback()203    def test_failure(self):204        tasklist = self._setup_common()205        for task in tasklist:206            task.fork_waitfor.expect_call(timeout=None).and_return(1)207            (subcommand.cPickle.load.expect_call(task.result_pickle)208                    .and_return(6))209            task.result_pickle.close.expect_call()210        self.assertRaises(subcommand.error.AutoservError, subcommand.parallel,211                          tasklist)212        self.god.check_playback()213    def test_timeout(self):214        self.god.stub_function(subcommand.time, 'time')215        tasklist = self._setup_common()216        timeout = 10217        subcommand.time.time.expect_call().and_return(1)218        for task in tasklist:219            subcommand.time.time.expect_call().and_return(1)220            task.fork_waitfor.expect_call(timeout=timeout).and_return(None)221            (subcommand.cPickle.load.expect_call(task.result_pickle)222                    .and_return(6))223            task.result_pickle.close.expect_call()224        self.assertRaises(subcommand.error.AutoservError, subcommand.parallel,225                          tasklist, timeout=timeout)226        self.god.check_playback()227    def test_return_results(self):228        tasklist = self._setup_common()229        tasklist[0].fork_waitfor.expect_call(timeout=None).and_return(0)230        (subcommand.cPickle.load.expect_call(tasklist[0].result_pickle)231                .and_return(6))232        tasklist[0].result_pickle.close.expect_call()233        error = Exception('fail')234        tasklist[1].fork_waitfor.expect_call(timeout=None).and_return(1)235        (subcommand.cPickle.load.expect_call(tasklist[1].result_pickle)236                .and_return(error))237        tasklist[1].result_pickle.close.expect_call()238        self.assertEquals(subcommand.parallel(tasklist, return_results=True),239                          [6, error])240        self.god.check_playback()241class test_parallel_simple(unittest.TestCase):242    def setUp(self):243        self.god = mock.mock_god()244        self.god.stub_function(subcommand, 'parallel')245        ctor = self.god.create_mock_function('subcommand')246        self.god.stub_with(subcommand, 'subcommand', ctor)247    def tearDown(self):248        self.god.unstub_all()249    def test_simple_success(self):250        func = self.god.create_mock_function('func')251        func.expect_call(3)252        subcommand.parallel_simple(func, (3,))253        self.god.check_playback()254    def test_simple_failure(self):255        func = self.god.create_mock_function('func')256        error = Exception('fail')257        func.expect_call(3).and_raises(error)258        self.assertRaises(Exception, subcommand.parallel_simple, func, (3,))259        self.god.check_playback()260    def test_simple_return_value(self):261        func = self.god.create_mock_function('func')262        result = 1000263        func.expect_call(3).and_return(result)264        self.assertEquals(subcommand.parallel_simple(func, (3,),265                                                     return_results=True),266                          [result])267        self.god.check_playback()268    def test_default_subdirs_constructor(self):269        func = self.god.create_mock_function('func')270        args = range(4)271        for arg in args:272            subcommand.subcommand.expect_call(273                    func, [arg], str(arg)).and_return(arg)274        subcommand.parallel.expect_call(args, None, return_results=False)275        subcommand.parallel_simple(func, args)276        self.god.check_playback()277    def test_nolog_skips_subdirs(self):278        func = self.god.create_mock_function('func')279        args = range(3)280        for arg in args:281            subcommand.subcommand.expect_call(282                    func, [arg], None).and_return(arg)283        subcommand.parallel.expect_call(args, None, return_results=False)284        subcommand.parallel_simple(func, args, log=False)285        self.god.check_playback()286    def test_custom_subdirs_constructor(self):287        func = self.god.create_mock_function('func')288        args = range(7)289        subdirs = ['subdir%s' % arg for arg in args]290        for arg, subdir in zip(args, subdirs):291            subcommand.subcommand.expect_call(292                    func, [arg], subdir).and_return(arg)293        subcommand.parallel.expect_call(args, None, return_results=False)294        subcommand.parallel_simple(295                func, args, subdir_name_constructor=lambda x: 'subdir%s' % x)296        self.god.check_playback()297if __name__ == '__main__':...

channel_test.py

Source:channel_test.py

...10        self.client = Mock(spec_set=Client)11        self.channel_id = ChannelId('/test')12        self.channel = Channel(self.client, self.channel_id)13        self.mock_message = Message(data='dummy')14    def create_mock_function(self, name='mock', **kwargs):15        mock = Mock(**kwargs)16        mock.__name__ = name17        return mock18    def test_repr(self):19        assert repr(self.channel) == self.channel_id20    def test_channel_id(self):21        assert self.channel.channel_id is self.channel_id22    def test_is_meta(self):23        assert not self.channel.is_meta24        channel = Channel(self.client, ChannelId.META_HANDSHAKE)25        assert channel.is_meta26    def test_is_wild(self):27        assert not self.channel.is_wild28        channel = Channel(self.client, '/test/*')29        assert channel.is_wild30        channel = Channel(self.client, '/*')31        assert channel.is_wild32        channel = Channel(self.client, '/test/**')33        assert not channel.is_wild34        channel = Channel(self.client, '*')35        assert not channel.is_wild36    def test_is_wild_deep(self):37        assert not self.channel.is_wild_deep38        channel = Channel(self.client, '/test/**')39        assert channel.is_wild_deep40        channel = Channel(self.client, '/**')41        assert channel.is_wild_deep42        channel = Channel(self.client, '/*')43        assert not channel.is_wild_deep44        channel = Channel(self.client, '**')45        assert not channel.is_wild_deep46    def test_parts(self):47        assert self.channel.parts == ['test']48        channel = Channel(self.client, '/test/some/channel')49        assert channel.parts == ['test', 'some', 'channel']50        channel = Channel(self.client, '')51        assert channel.parts == []52    def test_has_subscriptions(self):53        assert not self.channel.has_subscriptions54        mock_listener = self.create_mock_function()55        self.channel.add_listener(mock_listener)56        assert not self.channel.has_subscriptions57        mock_subscription = self.create_mock_function()58        self.channel.subscribe(mock_subscription)59        assert self.channel.has_subscriptions60        assert self.channel.remove_listener(function=mock_listener)61        assert self.channel.has_subscriptions62        assert self.channel.unsubscribe(function=mock_subscription)63        assert not self.channel.has_subscriptions64    def test_add_listener(self):65        mock_listener = self.create_mock_function()66        listener_id = self.channel.add_listener(mock_listener, 1, foo='bar')67        assert not mock_listener.called68        self.channel.notify_listeners(self.channel, self.mock_message)69        mock_listener.assert_called_once_with(self.channel, self.mock_message, 1, foo='bar')70        assert self.channel.remove_listener(id=listener_id)71    def test_clear_listeners(self):72        self.channel.clear_listeners()73        mock_listener = self.create_mock_function()74        mock_subscription = self.create_mock_function()75        listener_id = self.channel.add_listener(mock_listener)76        self.channel.subscribe(mock_subscription)77        self.channel.clear_listeners()78        assert not self.channel.remove_listener(id=listener_id)79        self.channel.notify_listeners(self.channel, self.mock_message)80        assert not mock_listener.called81        mock_subscription.assert_called_once_with(self.channel, self.mock_message)82    def test_clear_subscriptions(self):83        self.channel.clear_subscriptions()84        mock_listener = self.create_mock_function()85        mock_subscription = self.create_mock_function()86        self.channel.add_listener(mock_listener)87        subscription_id = self.channel.subscribe(mock_subscription)88        self.channel.clear_subscriptions()89        assert not self.channel.unsubscribe(id=subscription_id)90        self.channel.notify_listeners(self.channel, self.mock_message)91        mock_listener.assert_called_once_with(self.channel, self.mock_message)92        assert not mock_subscription.called93    def test_get_wilds(self):94        channel = Channel(self.client, '/test/some/channel')95        assert channel.get_wilds() == [96            '/test/some/*',97            '/test/some/**',98            '/test/**',99            '/**'100        ]101        channel = Channel(self.client, '/')102        assert channel.get_wilds() == [103            '/*',104            '/**'105        ]106        channel = Channel(self.client, '')107        assert channel.get_wilds() == []108    def test_notify_listeners(self):109        mock_listener = self.create_mock_function()110        mock_subscription_1 = self.create_mock_function()111        mock_subscription_2 = self.create_mock_function(side_effect=Exception())112        mock_subscription_3 = self.create_mock_function()113        self.channel.add_listener(mock_listener, 1, foo='bar1')114        self.channel.subscribe(mock_subscription_1, 2, foo='bar2')115        subscription_id = self.channel.subscribe(mock_subscription_2, 3)116        self.channel.subscribe(mock_subscription_3, foo='bar4')117        self.channel.notify_listeners(self.channel, self.mock_message)118        mock_listener.assert_called_once_with(self.channel, self.mock_message, 1, foo='bar1')119        mock_subscription_1.assert_called_once_with(self.channel, self.mock_message, 2, foo='bar2')120        mock_subscription_2.assert_called_once_with(self.channel, self.mock_message, 3)121        mock_subscription_3.assert_called_once_with(self.channel, self.mock_message, foo='bar4')122        self.client.fire.assert_called_once_with(123            self.client.EVENT_LISTENER_EXCEPTION,124            Listener(125                id=subscription_id,126                function=mock_subscription_2,127                extra_args=(3,),128                extra_kwargs={}129            ),130            self.mock_message,131            mock_subscription_2.side_effect132        )133    def test_notify_listeners_without_data(self):134        mock_listener = self.create_mock_function()135        mock_subscription = self.create_mock_function()136        self.mock_message.data = None137        self.channel.add_listener(mock_listener)138        self.channel.subscribe(mock_subscription)139        self.channel.notify_listeners(self.channel, self.mock_message)140        mock_listener.assert_called_once_with(self.channel, self.mock_message)141        assert not mock_subscription.called142    def test_notify_listeners_with_other_channel(self):143        mock_listener = self.create_mock_function()144        mock_subscription = self.create_mock_function()145        other_channel = Channel(self.client, '/other')146        self.channel.add_listener(mock_listener)147        self.channel.subscribe(mock_subscription)148        self.channel.notify_listeners(other_channel, self.mock_message)149        mock_listener.assert_called_once_with(other_channel, self.mock_message)150        mock_subscription.assert_called_once_with(other_channel, self.mock_message)151    def test_publish(self):152        self.channel.publish('dummy')153        self.client.send.assert_called_once_with({154            'channel': '/test',155            'data': 'dummy'156        })157        self.channel.publish('dummy', properties={'id': '1'})158        self.client.send.assert_called_with({159            'channel': '/test',160            'data': 'dummy',161            'id': '1'162        })163    def test_remove_listener(self):164        # Add a listener165        mock_listener = self.create_mock_function()166        listener_id = self.channel.add_listener(mock_listener)167        # Check validation of optional arguments168        self.assertRaises(ValueError, self.channel.remove_listener)169        self.assertRaises(ValueError, self.channel.remove_listener, id=listener_id, function=mock_listener)170        # Make sure non-matches are handled correctly171        assert not self.channel.remove_listener(id=listener_id - 1)172        assert not self.channel.remove_listener(function=self.create_mock_function())173        # Test removal by ID174        assert self.channel.remove_listener(id=listener_id)175        self.channel.notify_listeners(self.channel, self.mock_message)176        assert not mock_listener.called177        # Test removal by function178        mock_listener_2 = self.create_mock_function()179        self.channel.add_listener(mock_listener)180        self.channel.add_listener(mock_listener_2)181        self.channel.add_listener(mock_listener)182        self.channel.notify_listeners(self.channel, self.mock_message)183        assert mock_listener.call_count == 2184        assert mock_listener_2.call_count == 1185        assert self.channel.remove_listener(function=mock_listener)186        self.channel.notify_listeners(self.channel, self.mock_message)187        assert mock_listener.call_count == 2188        assert mock_listener_2.call_count == 2189    def test_subscribe(self):190        mock_subscription = self.create_mock_function()191        subscription_id = self.channel.subscribe(mock_subscription, 1, foo='bar')192        self.client.send.assert_called_once_with({193            'channel': ChannelId.META_SUBSCRIBE,194            'subscription': self.channel_id195        })196        self.channel.subscribe(mock_subscription)197        assert self.client.send.call_count == 1198        self.channel.notify_listeners(self.channel, self.mock_message)199        assert mock_subscription.call_args_list == [200            ((self.channel, self.mock_message, 1), {'foo': 'bar'}),201            ((self.channel, self.mock_message),)202        ]203        assert self.channel.unsubscribe(id=subscription_id)204        assert self.channel.unsubscribe(function=mock_subscription)205    def test_subscribe_with_properties(self):206        mock_subscription = self.create_mock_function()207        self.channel.subscribe(mock_subscription, 1, foo='bar', properties={'id': '1'})208        self.client.send.assert_called_once_with({209            'channel': ChannelId.META_SUBSCRIBE,210            'subscription': self.channel_id,211            'id': '1'212        })213        self.channel.notify_listeners(self.channel, self.mock_message)214        mock_subscription.assert_called_once_with(self.channel, self.mock_message, 1, foo='bar')215    def test_unsubscribe(self):216        mock_subscription = self.create_mock_function()217        subscription_id = self.channel.subscribe(mock_subscription)218        self.client.send.reset_mock()219        self.assertRaises(ValueError, self.channel.unsubscribe)220        self.assertRaises(ValueError, self.channel.unsubscribe, id=subscription_id, function=mock_subscription)221        assert not self.channel.unsubscribe(id=subscription_id - 1)222        assert not self.channel.unsubscribe(function=self.create_mock_function())223        assert self.channel.unsubscribe(id=subscription_id)224        self.client.send.assert_called_once_with({225            'channel': ChannelId.META_UNSUBSCRIBE,226            'subscription': self.channel_id227        })228        self.channel.notify_listeners(self.channel, self.mock_message)229        assert not mock_subscription.called230        self.channel.subscribe(mock_subscription)231        self.channel.subscribe(mock_subscription)232        self.client.send.reset_mock()233        self.channel.notify_listeners(self.channel, self.mock_message)234        assert mock_subscription.call_count == 2235        assert self.channel.unsubscribe(function=mock_subscription)236        assert self.client.send.call_count == 1237        self.channel.notify_listeners(self.channel, self.mock_message)238        assert mock_subscription.call_count == 2239    def test_unsubscribe_with_properties(self):240        mock_subscription = self.create_mock_function()241        self.channel.subscribe(mock_subscription)242        self.client.send.reset_mock()243        assert self.channel.unsubscribe(function=mock_subscription, properties={'id': '1'})244        self.client.send.assert_called_once_with({245            'channel': ChannelId.META_UNSUBSCRIBE,246            'subscription': self.channel_id,247            'id': '1'248        })249        self.channel.notify_listeners(self.channel, self.mock_message)...

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.