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

How to use _create_subcommand method in autotest

Best Python code snippet using autotest_python

subcommand_unittest.py

Source:subcommand_unittest.py

...4import common5from autotest_lib.client.common_lib import error6from autotest_lib.client.common_lib.test_utils import mock7from autotest_lib.server import subcommand8def _create_subcommand(func, args):9    # to avoid __init__10    class wrapper(subcommand.subcommand):11        def __init__(self, func, args):12            self.func = func13            self.args = args14            self.subdir = None15            self.debug = None16            self.pid = None17            self.returncode = None18    return wrapper(func, args)19class subcommand_test(unittest.TestCase):20    def setUp(self):21        self.god = mock.mock_god()22    def tearDown(self):23        self.god.unstub_all()24        # cleanup the hooks25        subcommand.subcommand.fork_hooks = []26        subcommand.subcommand.join_hooks = []27    def test_create(self):28        def check_attributes(cmd, func, args, subdir=None, debug=None,29                             pid=None, returncode=None, fork_hooks=[],30                             join_hooks=[]):31            self.assertEquals(cmd.func, func)32            self.assertEquals(cmd.args, args)33            self.assertEquals(cmd.subdir, subdir)34            self.assertEquals(cmd.debug, debug)35            self.assertEquals(cmd.pid, pid)36            self.assertEquals(cmd.returncode, returncode)37            self.assertEquals(cmd.fork_hooks, fork_hooks)38            self.assertEquals(cmd.join_hooks, join_hooks)39        def func(arg1, arg2):40            pass41        cmd = subcommand.subcommand(func, (2, 3))42        check_attributes(cmd, func, (2, 3))43        self.god.check_playback()44        self.god.stub_function(subcommand.os.path, 'abspath')45        self.god.stub_function(subcommand.os.path, 'exists')46        self.god.stub_function(subcommand.os, 'mkdir')47        subcommand.os.path.abspath.expect_call('dir').and_return('/foo/dir')48        subcommand.os.path.exists.expect_call('/foo/dir').and_return(False)49        subcommand.os.mkdir.expect_call('/foo/dir')50        (subcommand.os.path.exists.expect_call('/foo/dir/debug')51                .and_return(False))52        subcommand.os.mkdir.expect_call('/foo/dir/debug')53        cmd = subcommand.subcommand(func, (2, 3), subdir='dir')54        check_attributes(cmd, func, (2, 3), subdir='/foo/dir',55                         debug='/foo/dir/debug')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)...

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.