How to use _runner method in Molotov

Best Python code snippet using molotov_python

test_winrm_base.py

Source:test_winrm_base.py

...27from winrm_runner import winrm_ps_command_runner28class WinRmBaseTestCase(RunnerTestCase):29    def setUp(self):30        super(WinRmBaseTestCase, self).setUpClass()31        self._runner = winrm_ps_command_runner.get_runner()32    def _init_runner(self):33        runner_parameters = {'host': 'host@domain.tld',34                             'username': 'user@domain.tld',35                             'password': 'xyz987'}36        self._runner.runner_parameters = runner_parameters37        self._runner.pre_run()38    def test_win_rm_runner_timout_error(self):39        error = WinRmRunnerTimoutError('test_response')40        self.assertIsInstance(error, Exception)41        self.assertEquals(error.response, 'test_response')42        with self.assertRaises(WinRmRunnerTimoutError):43            raise WinRmRunnerTimoutError('test raising')44    def test_init(self):45        runner = winrm_ps_command_runner.WinRmPsCommandRunner('abcdef')46        self.assertIsInstance(runner, WinRmBaseRunner)47        self.assertIsInstance(runner, ActionRunner)48        self.assertEquals(runner.runner_id, "abcdef")49    @mock.patch('winrm_runner.winrm_base.ActionRunner.pre_run')50    def test_pre_run(self, mock_pre_run):51        runner_parameters = {'host': 'host@domain.tld',52                             'username': 'user@domain.tld',53                             'password': 'abc123',54                             'timeout': 99,55                             'port': 1234,56                             'scheme': 'http',57                             'transport': 'ntlm',58                             'verify_ssl_cert': False,59                             'cwd': 'C:\\Test',60                             'env': {'TEST_VAR': 'TEST_VALUE'},61                             'kwarg_op': '/'}62        self._runner.runner_parameters = runner_parameters63        self._runner.pre_run()64        mock_pre_run.assert_called_with()65        self.assertEquals(self._runner._session, None)66        self.assertEquals(self._runner._host, 'host@domain.tld')67        self.assertEquals(self._runner._username, 'user@domain.tld')68        self.assertEquals(self._runner._password, 'abc123')69        self.assertEquals(self._runner._timeout, 99)70        self.assertEquals(self._runner._read_timeout, 100)71        self.assertEquals(self._runner._port, 1234)72        self.assertEquals(self._runner._scheme, 'http')73        self.assertEquals(self._runner._transport, 'ntlm')74        self.assertEquals(self._runner._winrm_url, 'http://host@domain.tld:1234/wsman')75        self.assertEquals(self._runner._verify_ssl, False)76        self.assertEquals(self._runner._server_cert_validation, 'ignore')77        self.assertEquals(self._runner._cwd, 'C:\\Test')78        self.assertEquals(self._runner._env, {'TEST_VAR': 'TEST_VALUE'})79        self.assertEquals(self._runner._kwarg_op, '/')80    @mock.patch('winrm_runner.winrm_base.ActionRunner.pre_run')81    def test_pre_run_defaults(self, mock_pre_run):82        runner_parameters = {'host': 'host@domain.tld',83                             'username': 'user@domain.tld',84                             'password': 'abc123'}85        self._runner.runner_parameters = runner_parameters86        self._runner.pre_run()87        mock_pre_run.assert_called_with()88        self.assertEquals(self._runner._host, 'host@domain.tld')89        self.assertEquals(self._runner._username, 'user@domain.tld')90        self.assertEquals(self._runner._password, 'abc123')91        self.assertEquals(self._runner._timeout, 60)92        self.assertEquals(self._runner._read_timeout, 61)93        self.assertEquals(self._runner._port, 5986)94        self.assertEquals(self._runner._scheme, 'https')95        self.assertEquals(self._runner._transport, 'ntlm')96        self.assertEquals(self._runner._winrm_url, 'https://host@domain.tld:5986/wsman')97        self.assertEquals(self._runner._verify_ssl, True)98        self.assertEquals(self._runner._server_cert_validation, 'validate')99        self.assertEquals(self._runner._cwd, None)100        self.assertEquals(self._runner._env, {})101        self.assertEquals(self._runner._kwarg_op, '-')102    @mock.patch('winrm_runner.winrm_base.ActionRunner.pre_run')103    def test_pre_run_5985_force_http(self, mock_pre_run):104        runner_parameters = {'host': 'host@domain.tld',105                             'username': 'user@domain.tld',106                             'password': 'abc123',107                             'port': 5985,108                             'scheme': 'https'}109        self._runner.runner_parameters = runner_parameters110        self._runner.pre_run()111        mock_pre_run.assert_called_with()112        self.assertEquals(self._runner._host, 'host@domain.tld')113        self.assertEquals(self._runner._username, 'user@domain.tld')114        self.assertEquals(self._runner._password, 'abc123')115        self.assertEquals(self._runner._timeout, 60)116        self.assertEquals(self._runner._read_timeout, 61)117        # ensure port is still 5985118        self.assertEquals(self._runner._port, 5985)119        # ensure scheme is set back to http120        self.assertEquals(self._runner._scheme, 'http')121        self.assertEquals(self._runner._transport, 'ntlm')122        self.assertEquals(self._runner._winrm_url, 'http://host@domain.tld:5985/wsman')123        self.assertEquals(self._runner._verify_ssl, True)124        self.assertEquals(self._runner._server_cert_validation, 'validate')125        self.assertEquals(self._runner._cwd, None)126        self.assertEquals(self._runner._env, {})127        self.assertEquals(self._runner._kwarg_op, '-')128    @mock.patch('winrm_runner.winrm_base.ActionRunner.pre_run')129    def test_pre_run_none_env(self, mock_pre_run):130        runner_parameters = {'host': 'host@domain.tld',131                             'username': 'user@domain.tld',132                             'password': 'abc123',133                             'env': None}134        self._runner.runner_parameters = runner_parameters135        self._runner.pre_run()136        mock_pre_run.assert_called_with()137        # ensure that env is set to {} even though we passed in None138        self.assertEquals(self._runner._env, {})139    @mock.patch('winrm_runner.winrm_base.ActionRunner.pre_run')140    def test_pre_run_ssl_verify_true(self, mock_pre_run):141        runner_parameters = {'host': 'host@domain.tld',142                             'username': 'user@domain.tld',143                             'password': 'abc123',144                             'verify_ssl_cert': True}145        self._runner.runner_parameters = runner_parameters146        self._runner.pre_run()147        mock_pre_run.assert_called_with()148        self.assertEquals(self._runner._verify_ssl, True)149        self.assertEquals(self._runner._server_cert_validation, 'validate')150    @mock.patch('winrm_runner.winrm_base.ActionRunner.pre_run')151    def test_pre_run_ssl_verify_false(self, mock_pre_run):152        runner_parameters = {'host': 'host@domain.tld',153                             'username': 'user@domain.tld',154                             'password': 'abc123',155                             'verify_ssl_cert': False}156        self._runner.runner_parameters = runner_parameters157        self._runner.pre_run()158        mock_pre_run.assert_called_with()159        self.assertEquals(self._runner._verify_ssl, False)160        self.assertEquals(self._runner._server_cert_validation, 'ignore')161    @mock.patch('winrm_runner.winrm_base.Session')162    def test_get_session(self, mock_session):163        self._runner._session = None164        self._runner._winrm_url = 'https://host@domain.tld:5986/wsman'165        self._runner._username = 'user@domain.tld'166        self._runner._password = 'abc123'167        self._runner._transport = 'ntlm'168        self._runner._server_cert_validation = 'validate'169        self._runner._timeout = 60170        self._runner._read_timeout = 61171        mock_session.return_value = "session"172        result = self._runner._get_session()173        self.assertEquals(result, "session")174        self.assertEquals(result, self._runner._session)175        mock_session.assert_called_with('https://host@domain.tld:5986/wsman',176                                        auth=('user@domain.tld', 'abc123'),177                                        transport='ntlm',178                                        server_cert_validation='validate',179                                        operation_timeout_sec=60,180                                        read_timeout_sec=61)181        # ensure calling _get_session again doesn't create a new one, it reuses the existing182        old_session = self._runner._session183        result = self._runner._get_session()184        self.assertEquals(result, old_session)185    def test_winrm_get_command_output(self):186        self._runner._timeout = 0187        mock_protocol = mock.MagicMock()188        mock_protocol._raw_get_command_output.side_effect = [189            (b'output1', b'error1', 123, False),190            (b'output2', b'error2', 456, False),191            (b'output3', b'error3', 789, True)192        ]193        result = self._runner._winrm_get_command_output(mock_protocol, 567, 890)194        self.assertEquals(result, (b'output1output2output3', b'error1error2error3', 789))195        mock_protocol._raw_get_command_output.assert_has_calls = [196            mock.call(567, 890),197            mock.call(567, 890),198            mock.call(567, 890)199        ]200    def test_winrm_get_command_output_timeout(self):201        self._runner._timeout = 0.1202        mock_protocol = mock.MagicMock()203        def sleep_for_timeout(*args, **kwargs):204            time.sleep(0.2)205            return (b'output1', b'error1', 123, False)206        mock_protocol._raw_get_command_output.side_effect = sleep_for_timeout207        with self.assertRaises(WinRmRunnerTimoutError) as cm:208            self._runner._winrm_get_command_output(mock_protocol, 567, 890)209        timeout_exception = cm.exception210        self.assertEqual(timeout_exception.response.std_out, b'output1')211        self.assertEqual(timeout_exception.response.std_err, b'error1')212        self.assertEqual(timeout_exception.response.status_code, WINRM_TIMEOUT_EXIT_CODE)213        mock_protocol._raw_get_command_output.assert_called_with(567, 890)214    def test_winrm_get_command_output_operation_timeout(self):215        self._runner._timeout = 0.1216        mock_protocol = mock.MagicMock()217        def sleep_for_timeout_then_raise(*args, **kwargs):218            time.sleep(0.2)219            raise WinRMOperationTimeoutError()220        mock_protocol._raw_get_command_output.side_effect = sleep_for_timeout_then_raise221        with self.assertRaises(WinRmRunnerTimoutError) as cm:222            self._runner._winrm_get_command_output(mock_protocol, 567, 890)223        timeout_exception = cm.exception224        self.assertEqual(timeout_exception.response.std_out, b'')225        self.assertEqual(timeout_exception.response.std_err, b'')226        self.assertEqual(timeout_exception.response.status_code, WINRM_TIMEOUT_EXIT_CODE)227        mock_protocol._raw_get_command_output.assert_called_with(567, 890)228    def test_winrm_run_cmd(self):229        mock_protocol = mock.MagicMock()230        mock_protocol.open_shell.return_value = 123231        mock_protocol.run_command.return_value = 456232        mock_protocol._raw_get_command_output.return_value = (b'output', b'error', 9, True)233        mock_session = mock.MagicMock(protocol=mock_protocol)234        self._init_runner()235        result = self._runner._winrm_run_cmd(mock_session, "fake-command",236                                             args=['arg1', 'arg2'],237                                             env={'PATH': 'C:\\st2\\bin'},238                                             cwd='C:\\st2')239        expected_response = Response((b'output', b'error', 9))240        expected_response.timeout = False241        self.assertEquals(result.__dict__, expected_response.__dict__)242        mock_protocol.open_shell.assert_called_with(env_vars={'PATH': 'C:\\st2\\bin'},243                                                    working_directory='C:\\st2')244        mock_protocol.run_command.assert_called_with(123, 'fake-command', ['arg1', 'arg2'])245        mock_protocol._raw_get_command_output.assert_called_with(123, 456)246        mock_protocol.cleanup_command.assert_called_with(123, 456)247        mock_protocol.close_shell.assert_called_with(123)248    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._winrm_get_command_output')249    def test_winrm_run_cmd_timeout(self, mock_get_command_output):250        mock_protocol = mock.MagicMock()251        mock_protocol.open_shell.return_value = 123252        mock_protocol.run_command.return_value = 456253        mock_session = mock.MagicMock(protocol=mock_protocol)254        mock_get_command_output.side_effect = WinRmRunnerTimoutError(Response(('', '', 5)))255        self._init_runner()256        result = self._runner._winrm_run_cmd(mock_session, "fake-command",257                                             args=['arg1', 'arg2'],258                                             env={'PATH': 'C:\\st2\\bin'},259                                             cwd='C:\\st2')260        expected_response = Response(('', '', 5))261        expected_response.timeout = True262        self.assertEquals(result.__dict__, expected_response.__dict__)263        mock_protocol.open_shell.assert_called_with(env_vars={'PATH': 'C:\\st2\\bin'},264                                                    working_directory='C:\\st2')265        mock_protocol.run_command.assert_called_with(123, 'fake-command', ['arg1', 'arg2'])266        mock_protocol.cleanup_command.assert_called_with(123, 456)267        mock_protocol.close_shell.assert_called_with(123)268    def test_winrm_encode(self):269        result = self._runner._winrm_encode('hello world')270        # result translated into UTF-16 little-endian271        self.assertEquals(result, 'aABlAGwAbABvACAAdwBvAHIAbABkAA==')272    def test_winrm_ps_cmd(self):273        result = self._runner._winrm_ps_cmd('abc123==')274        self.assertEquals(result, 'powershell -encodedcommand abc123==')275    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._winrm_run_cmd')276    def test_winrm_run_ps(self, mock_run_cmd):277        mock_run_cmd.return_value = Response(('output', '', 3))278        script = "Get-ADUser stanley"279        result = self._runner._winrm_run_ps("session", script,280                                            env={'PATH': 'C:\\st2\\bin'},281                                            cwd='C:\\st2')282        self.assertEquals(result.__dict__,283                          Response(('output', '', 3)).__dict__)284        expected_ps = ('powershell -encodedcommand ' +285                       b64encode("Get-ADUser stanley".encode('utf_16_le')).decode('ascii'))286        mock_run_cmd.assert_called_with("session",287                                        expected_ps,288                                        env={'PATH': 'C:\\st2\\bin'},289                                        cwd='C:\\st2')290    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._winrm_run_cmd')291    def test_winrm_run_ps_clean_stderr(self, mock_run_cmd):292        mock_run_cmd.return_value = Response(('output', 'error', 3))293        mock_session = mock.MagicMock()294        mock_session._clean_error_msg.return_value = 'e'295        script = "Get-ADUser stanley"296        result = self._runner._winrm_run_ps(mock_session, script,297                                            env={'PATH': 'C:\\st2\\bin'},298                                            cwd='C:\\st2')299        self.assertEquals(result.__dict__,300                          Response(('output', 'e', 3)).__dict__)301        expected_ps = ('powershell -encodedcommand ' +302                       b64encode("Get-ADUser stanley".encode('utf_16_le')).decode('ascii'))303        mock_run_cmd.assert_called_with(mock_session,304                                        expected_ps,305                                        env={'PATH': 'C:\\st2\\bin'},306                                        cwd='C:\\st2')307        mock_session._clean_error_msg.assert_called_with('error')308    def test_translate_response_success(self):309        response = Response(('output1', 'error1', 0))310        response.timeout = False311        result = self._runner._translate_response(response)312        self.assertEquals(result, ('succeeded',313                                   {'failed': False,314                                    'succeeded': True,315                                    'return_code': 0,316                                    'stdout': 'output1',317                                    'stderr': 'error1'},318                                   None))319    def test_translate_response_failure(self):320        response = Response(('output1', 'error1', 123))321        response.timeout = False322        result = self._runner._translate_response(response)323        self.assertEquals(result, ('failed',324                                   {'failed': True,325                                    'succeeded': False,326                                    'return_code': 123,327                                    'stdout': 'output1',328                                    'stderr': 'error1'},329                                   None))330    def test_translate_response_timeout(self):331        response = Response(('output1', 'error1', 123))332        response.timeout = True333        result = self._runner._translate_response(response)334        self.assertEquals(result, ('timeout',335                                   {'failed': True,336                                    'succeeded': False,337                                    'return_code': -1,338                                    'stdout': 'output1',339                                    'stderr': 'error1'},340                                   None))341    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._run_ps_or_raise')342    def test_make_tmp_dir(self, mock_run_ps_or_raise):343        mock_run_ps_or_raise.return_value = {'stdout': ' expected \n'}344        result = self._runner._make_tmp_dir('C:\\Windows\\Temp')345        self.assertEquals(result, 'expected')346        mock_run_ps_or_raise.assert_called_with('''$parent = C:\\Windows\\Temp347$name = [System.IO.Path]::GetRandomFileName()348$path = Join-Path $parent $name349New-Item -ItemType Directory -Path $path | Out-Null350$path''',351                                                ("Unable to make temporary directory for"352                                                 " powershell script"))353    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._run_ps_or_raise')354    def test_rm_dir(self, mock_run_ps_or_raise):355        self._runner._rm_dir('C:\\Windows\\Temp\\testtmpdir')356        mock_run_ps_or_raise.assert_called_with(357            'Remove-Item -Force -Recurse -Path "C:\\Windows\\Temp\\testtmpdir"',358            "Unable to remove temporary directory for powershell script")359    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._upload_chunk')360    @mock.patch('winrm_runner.winrm_base.open')361    @mock.patch('os.path.exists')362    def test_upload_chunk_file(self, mock_os_path_exists, mock_open, mock_upload_chunk):363        mock_os_path_exists.return_value = True364        mock_src_file = mock.MagicMock()365        mock_src_file.read.return_value = "test data"366        mock_open.return_value.__enter__.return_value = mock_src_file367        self._runner._upload('/opt/data/test.ps1', 'C:\\Windows\\Temp\\test.ps1')368        mock_os_path_exists.assert_called_with('/opt/data/test.ps1')369        mock_open.assert_called_with('/opt/data/test.ps1', 'r')370        mock_src_file.read.assert_called_with()371        mock_upload_chunk.assert_has_calls([372            mock.call('C:\\Windows\\Temp\\test.ps1', 'test data')373        ])374    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._upload_chunk')375    @mock.patch('os.path.exists')376    def test_upload_chunk_data(self, mock_os_path_exists, mock_upload_chunk):377        mock_os_path_exists.return_value = False378        self._runner._upload('test data', 'C:\\Windows\\Temp\\test.ps1')379        mock_os_path_exists.assert_called_with('test data')380        mock_upload_chunk.assert_has_calls([381            mock.call('C:\\Windows\\Temp\\test.ps1', 'test data')382        ])383    @mock.patch('winrm_runner.winrm_base.WINRM_UPLOAD_CHUNK_SIZE_BYTES', 2)384    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._upload_chunk')385    @mock.patch('os.path.exists')386    def test_upload_chunk_multiple_chunks(self, mock_os_path_exists, mock_upload_chunk):387        mock_os_path_exists.return_value = False388        self._runner._upload('test data', 'C:\\Windows\\Temp\\test.ps1')389        mock_os_path_exists.assert_called_with('test data')390        mock_upload_chunk.assert_has_calls([391            mock.call('C:\\Windows\\Temp\\test.ps1', 'te'),392            mock.call('C:\\Windows\\Temp\\test.ps1', 'st'),393            mock.call('C:\\Windows\\Temp\\test.ps1', ' d'),394            mock.call('C:\\Windows\\Temp\\test.ps1', 'at'),395            mock.call('C:\\Windows\\Temp\\test.ps1', 'a'),396        ])397    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._run_ps_or_raise')398    def test_upload_chunk(self, mock_run_ps_or_raise):399        self._runner._upload_chunk('C:\\Windows\\Temp\\testtmp.ps1', 'hello world')400        mock_run_ps_or_raise.assert_called_with(401            '''$filePath = "C:\\Windows\\Temp\\testtmp.ps1"402$s = @"403aGVsbG8gd29ybGQ=404"@405$data = [System.Convert]::FromBase64String($s)406Add-Content -value $data -encoding byte -path $filePath407''',408            "Failed to upload chunk of powershell script")409    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._rm_dir')410    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._upload')411    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._make_tmp_dir')412    def test_tmp_script(self, mock_make_tmp_dir, mock_upload, mock_rm_dir):413        mock_make_tmp_dir.return_value = 'C:\\Windows\\Temp\\abc123'414        with self._runner._tmp_script('C:\\Windows\\Temp', 'Get-ChildItem') as tmp:415            self.assertEquals(tmp, 'C:\\Windows\\Temp\\abc123\\script.ps1')416        mock_make_tmp_dir.assert_called_with('C:\\Windows\\Temp')417        mock_upload.assert_called_with('Get-ChildItem',418                                       'C:\\Windows\\Temp\\abc123\\script.ps1')419        mock_rm_dir.assert_called_with('C:\\Windows\\Temp\\abc123')420    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._rm_dir')421    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._upload')422    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._make_tmp_dir')423    def test_tmp_script_cleans_up_when_raises(self, mock_make_tmp_dir, mock_upload,424                                              mock_rm_dir):425        mock_make_tmp_dir.return_value = 'C:\\Windows\\Temp\\abc123'426        mock_upload.side_effect = RuntimeError427        with self.assertRaises(RuntimeError):428            with self._runner._tmp_script('C:\\Windows\\Temp', 'Get-ChildItem') as tmp:429                self.assertEquals(tmp, "can never get here")430        mock_make_tmp_dir.assert_called_with('C:\\Windows\\Temp')431        mock_upload.assert_called_with('Get-ChildItem',432                                       'C:\\Windows\\Temp\\abc123\\script.ps1')433        mock_rm_dir.assert_called_with('C:\\Windows\\Temp\\abc123')434    @mock.patch('winrm.Protocol')435    def test_run_cmd(self, mock_protocol_init):436        mock_protocol = mock.MagicMock()437        mock_protocol._raw_get_command_output.side_effect = [438            (b'output1', b'error1', 0, False),439            (b'output2', b'error2', 0, False),440            (b'output3', b'error3', 0, True)441        ]442        mock_protocol_init.return_value = mock_protocol443        self._init_runner()444        result = self._runner.run_cmd("ipconfig /all")445        self.assertEquals(result, ('succeeded',446                                   {'failed': False,447                                    'succeeded': True,448                                    'return_code': 0,449                                    'stdout': 'output1output2output3',450                                    'stderr': 'error1error2error3'},451                                   None))452    @mock.patch('winrm.Protocol')453    def test_run_cmd_failed(self, mock_protocol_init):454        mock_protocol = mock.MagicMock()455        mock_protocol._raw_get_command_output.side_effect = [456            (b'output1', b'error1', 0, False),457            (b'output2', b'error2', 0, False),458            (b'output3', b'error3', 1, True)459        ]460        mock_protocol_init.return_value = mock_protocol461        self._init_runner()462        result = self._runner.run_cmd("ipconfig /all")463        self.assertEquals(result, ('failed',464                                   {'failed': True,465                                    'succeeded': False,466                                    'return_code': 1,467                                    'stdout': 'output1output2output3',468                                    'stderr': 'error1error2error3'},469                                   None))470    @mock.patch('winrm.Protocol')471    def test_run_cmd_timeout(self, mock_protocol_init):472        mock_protocol = mock.MagicMock()473        self._init_runner()474        self._runner._timeout = 0.1475        def sleep_for_timeout_then_raise(*args, **kwargs):476            time.sleep(0.2)477            return (b'output1', b'error1', 123, False)478        mock_protocol._raw_get_command_output.side_effect = sleep_for_timeout_then_raise479        mock_protocol_init.return_value = mock_protocol480        result = self._runner.run_cmd("ipconfig /all")481        self.assertEquals(result, ('timeout',482                                   {'failed': True,483                                    'succeeded': False,484                                    'return_code': -1,485                                    'stdout': 'output1',486                                    'stderr': 'error1'},487                                   None))488    @mock.patch('winrm.Protocol')489    def test_run_ps(self, mock_protocol_init):490        mock_protocol = mock.MagicMock()491        mock_protocol._raw_get_command_output.side_effect = [492            (b'output1', b'error1', 0, False),493            (b'output2', b'error2', 0, False),494            (b'output3', b'error3', 0, True)495        ]496        mock_protocol_init.return_value = mock_protocol497        self._init_runner()498        result = self._runner.run_ps("Get-Location")499        self.assertEquals(result, ('succeeded',500                                   {'failed': False,501                                    'succeeded': True,502                                    'return_code': 0,503                                    'stdout': 'output1output2output3',504                                    'stderr': 'error1error2error3'},505                                   None))506    @mock.patch('winrm.Protocol')507    def test_run_ps_failed(self, mock_protocol_init):508        mock_protocol = mock.MagicMock()509        mock_protocol._raw_get_command_output.side_effect = [510            (b'output1', b'error1', 0, False),511            (b'output2', b'error2', 0, False),512            (b'output3', b'error3', 1, True)513        ]514        mock_protocol_init.return_value = mock_protocol515        self._init_runner()516        result = self._runner.run_ps("Get-Location")517        self.assertEquals(result, ('failed',518                                   {'failed': True,519                                    'succeeded': False,520                                    'return_code': 1,521                                    'stdout': 'output1output2output3',522                                    'stderr': 'error1error2error3'},523                                   None))524    @mock.patch('winrm.Protocol')525    def test_run_ps_timeout(self, mock_protocol_init):526        mock_protocol = mock.MagicMock()527        self._init_runner()528        self._runner._timeout = 0.1529        def sleep_for_timeout_then_raise(*args, **kwargs):530            time.sleep(0.2)531            return (b'output1', b'error1', 123, False)532        mock_protocol._raw_get_command_output.side_effect = sleep_for_timeout_then_raise533        mock_protocol_init.return_value = mock_protocol534        result = self._runner.run_ps("Get-Location")535        self.assertEquals(result, ('timeout',536                                   {'failed': True,537                                    'succeeded': False,538                                    'return_code': -1,539                                    'stdout': 'output1',540                                    'stderr': 'error1'},541                                   None))542    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._run_ps')543    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._winrm_encode')544    def test_run_ps_params(self, mock_winrm_encode, mock_run_ps):545        mock_winrm_encode.return_value = 'xyz123=='546        mock_run_ps.return_value = "expected"547        self._init_runner()548        result = self._runner.run_ps("Get-Location", '-param1 value1 arg1')549        self.assertEquals(result, "expected")550        mock_winrm_encode.assert_called_with('& {Get-Location} -param1 value1 arg1')551        mock_run_ps.assert_called_with('xyz123==', is_b64=True)552    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._winrm_ps_cmd')553    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._run_ps_script')554    def test_run_ps_large_command_convert_to_script(self, mock_run_ps_script,555                                                    mock_winrm_ps_cmd):556        mock_run_ps_script.return_value = "expected"557        # max length of a command in powershelll558        script = 'powershell -encodedcommand '559        script += '#' * (WINRM_MAX_CMD_LENGTH + 1 - len(script))560        mock_winrm_ps_cmd.return_value = script561        self._init_runner()562        result = self._runner.run_ps('$PSVersionTable')563        self.assertEquals(result, "expected")564        mock_run_ps_script.assert_called_with('$PSVersionTable', None)565    @mock.patch('winrm.Protocol')566    def test__run_ps(self, mock_protocol_init):567        mock_protocol = mock.MagicMock()568        mock_protocol._raw_get_command_output.side_effect = [569            (b'output1', b'error1', 0, False),570            (b'output2', b'error2', 0, False),571            (b'output3', b'error3', 0, True)572        ]573        mock_protocol_init.return_value = mock_protocol574        self._init_runner()575        result = self._runner._run_ps("Get-Location")576        self.assertEquals(result, ('succeeded',577                                   {'failed': False,578                                    'succeeded': True,579                                    'return_code': 0,580                                    'stdout': 'output1output2output3',581                                    'stderr': 'error1error2error3'},582                                   None))583    @mock.patch('winrm.Protocol')584    def test__run_ps_failed(self, mock_protocol_init):585        mock_protocol = mock.MagicMock()586        mock_protocol._raw_get_command_output.side_effect = [587            (b'output1', b'error1', 0, False),588            (b'output2', b'error2', 0, False),589            (b'output3', b'error3', 1, True)590        ]591        mock_protocol_init.return_value = mock_protocol592        self._init_runner()593        result = self._runner._run_ps("Get-Location")594        self.assertEquals(result, ('failed',595                                   {'failed': True,596                                    'succeeded': False,597                                    'return_code': 1,598                                    'stdout': 'output1output2output3',599                                    'stderr': 'error1error2error3'},600                                   None))601    @mock.patch('winrm.Protocol')602    def test__run_ps_timeout(self, mock_protocol_init):603        mock_protocol = mock.MagicMock()604        self._init_runner()605        self._runner._timeout = 0.1606        def sleep_for_timeout_then_raise(*args, **kwargs):607            time.sleep(0.2)608            return (b'output1', b'error1', 123, False)609        mock_protocol._raw_get_command_output.side_effect = sleep_for_timeout_then_raise610        mock_protocol_init.return_value = mock_protocol611        result = self._runner._run_ps("Get-Location")612        self.assertEquals(result, ('timeout',613                                   {'failed': True,614                                    'succeeded': False,615                                    'return_code': -1,616                                    'stdout': 'output1',617                                    'stderr': 'error1'},618                                   None))619    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._winrm_run_ps')620    def test__run_ps_b64_default(self, mock_winrm_run_ps):621        mock_winrm_run_ps.return_value = mock.MagicMock(status_code=0,622                                                        timeout=False,623                                                        std_out='output1',624                                                        std_err='error1')625        self._init_runner()626        result = self._runner._run_ps("$PSVersionTable")627        self.assertEquals(result, ('succeeded',628                                   {'failed': False,629                                    'succeeded': True,630                                    'return_code': 0,631                                    'stdout': 'output1',632                                    'stderr': 'error1'},633                                   None))634        mock_winrm_run_ps.assert_called_with(self._runner._session,635                                             '$PSVersionTable',636                                             env={},637                                             cwd=None,638                                             is_b64=False)639    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._winrm_run_ps')640    def test__run_ps_b64_true(self, mock_winrm_run_ps):641        mock_winrm_run_ps.return_value = mock.MagicMock(status_code=0,642                                                        timeout=False,643                                                        std_out='output1',644                                                        std_err='error1')645        self._init_runner()646        result = self._runner._run_ps("xyz123", is_b64=True)647        self.assertEquals(result, ('succeeded',648                                   {'failed': False,649                                    'succeeded': True,650                                    'return_code': 0,651                                    'stdout': 'output1',652                                    'stderr': 'error1'},653                                   None))654        mock_winrm_run_ps.assert_called_with(self._runner._session,655                                             'xyz123',656                                             env={},657                                             cwd=None,658                                             is_b64=True)659    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._run_ps')660    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._tmp_script')661    def test__run_ps_script(self, mock_tmp_script, mock_run_ps):662        mock_tmp_script.return_value.__enter__.return_value = 'C:\\tmpscript.ps1'663        mock_run_ps.return_value = 'expected'664        self._init_runner()665        result = self._runner._run_ps_script("$PSVersionTable")666        self.assertEquals(result, 'expected')667        mock_tmp_script.assert_called_with('[System.IO.Path]::GetTempPath()',668                                           '$PSVersionTable')669        mock_run_ps.assert_called_with('& {C:\\tmpscript.ps1}')670    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._run_ps')671    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._tmp_script')672    def test__run_ps_script_with_params(self, mock_tmp_script, mock_run_ps):673        mock_tmp_script.return_value.__enter__.return_value = 'C:\\tmpscript.ps1'674        mock_run_ps.return_value = 'expected'675        self._init_runner()676        result = self._runner._run_ps_script("Get-ChildItem", '-param1 value1 arg1')677        self.assertEquals(result, 'expected')678        mock_tmp_script.assert_called_with('[System.IO.Path]::GetTempPath()',679                                           'Get-ChildItem')680        mock_run_ps.assert_called_with('& {C:\\tmpscript.ps1} -param1 value1 arg1')681    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._run_ps')682    def test__run_ps_or_raise(self, mock_run_ps):683        mock_run_ps.return_value = ('success',684                                    {685                                        'failed': False,686                                        'succeeded': True,687                                        'return_code': 0,688                                        'stdout': 'output',689                                        'stderr': 'error',690                                    },691                                    None)692        self._init_runner()693        result = self._runner._run_ps_or_raise('Get-ChildItem', 'my error message')694        self.assertEquals(result, {695            'failed': False,696            'succeeded': True,697            'return_code': 0,698            'stdout': 'output',699            'stderr': 'error',700        })701    @mock.patch('winrm_runner.winrm_base.WinRmBaseRunner._run_ps')702    def test__run_ps_or_raise_raises_on_failure(self, mock_run_ps):703        mock_run_ps.return_value = ('success',704                                    {705                                        'failed': True,706                                        'succeeded': False,707                                        'return_code': 1,708                                        'stdout': 'output',709                                        'stderr': 'error',710                                    },711                                    None)712        self._init_runner()713        with self.assertRaises(RuntimeError):714            self._runner._run_ps_or_raise('Get-ChildItem', 'my error message')715    def test_multireplace(self):716        multireplace_map = {'a': 'x',717                            'c': 'y',718                            'aaa': 'z'}719        result = self._runner._multireplace('aaaccaa', multireplace_map)720        self.assertEquals(result, 'zyyxx')721    def test_multireplace_powershell(self):722        param_str = (723            '\n'724            '\r'725            '\t'726            '\a'...

test_montecarlo_cpp.py

Source:test_montecarlo_cpp.py

...7#8# calculator = cppimport.imp("tools.montecarlo_cpp.pymontecarlo")9#10#11# def _runner(my_cards, cards_on_table, players, expected_result, iterations=5000):12#     """Montecarlo test"""13#     if len(cards_on_table) < 3:14#         cards_on_table = {'null'}15#     equity = calculator.montecarlo(my_cards, cards_on_table, players, iterations) * 10016#     assert equity == pytest.approx(expected_result, abs=3)17#18#19# @pytest.mark.skipif(ON_CI)20# def test_montecarlo1():21#     """Montecarlo test"""22#     my_cards = {'3H', '3S'}23#     cards_on_table = {'8S', '4S', 'QH', '8C', '4H'}24#     expected_results = 40.225#     players = 226#     _runner(my_cards, cards_on_table, players, expected_results)27#28#29# @pytest.mark.skipif(ON_CI)30# def test_montecarlo2():31#     """Montecarlo test"""32#     my_cards = {'8H', '8D'}33#     cards_on_table = {'QH', '7H', '9H', 'JH', 'TH'}34#     expected_results = 95.635#     players = 236#     _runner(my_cards, cards_on_table, players, expected_results)37#38#39# @pytest.mark.skipif(ON_CI)40# def test_montecarlo3():41#     """Montecarlo test"""42#     my_cards = {'AS', 'KS'}43#     cards_on_table = {}44#     expected_results = 49.9 + 1.945#     players = 346#     _runner(my_cards, cards_on_table, players, expected_results)47#48#49# @pytest.mark.skipif(ON_CI)50# def test_montecarlo4():51#     """Montecarlo test"""52#     my_cards = {'AS', 'KS'}53#     cards_on_table = {}54#     expected_results = 66.1 + 1.655#     players = 256#     _runner(my_cards, cards_on_table, players, expected_results)57#58#59# @pytest.mark.skipif(ON_CI)60# def test_montecarlo5():61#     """Montecarlo test"""62#     my_cards = {'8S', 'TS'}63#     cards_on_table = {'8H', 'KS', '9S', 'TH', 'KH'}64#     expected_results = 71.5 + 5.965#     players = 266#     _runner(my_cards, cards_on_table, players, expected_results)67#68#69# @pytest.mark.skipif(ON_CI)70# def test_montecarlo6():71#     """Montecarlo test"""72#     my_cards = {'8S', 'TS'}73#     cards_on_table = {'2S', '3S', '4S', 'KS', 'AS'}74#     expected_results = 8775#     players = 276#     _runner(my_cards, cards_on_table, players, expected_results)77#78#79# @pytest.mark.skipif(ON_CI)80# def test_montecarlo7():81#     """Montecarlo test"""82#     my_cards = {'8S', '2S'}83#     cards_on_table = {'5S', '3S', '4S', 'KS', 'AS'}84#     expected_results = 10085#     players = 286#     _runner(my_cards, cards_on_table, players, expected_results)87#88#89# @pytest.mark.skipif(ON_CI)90# def test_montecarlo8():91#     """Montecarlo test"""92#     my_cards = {'8S', 'TS'}93#     cards_on_table = {}94#     expected_results = 22.6 + 2.995#     players = 596#     _runner(my_cards, cards_on_table, players, expected_results)97#98#99# @pytest.mark.skipif(ON_CI)100# def test_montecarlo8b():101#     """Montecarlo test"""102#     my_cards = {'2C', 'QS'}103#     cards_on_table = {}104#     expected_results = 49.6  # 45 win and 4 tie105#     players = 2106#     _runner(my_cards, cards_on_table, players, expected_results)107#108#109# @pytest.mark.skipif(ON_CI)110# def test_montecarlo9():111#     """Montecarlo test"""112#     my_cards = {'7H', '7S'}113#     cards_on_table = {'7C', '8C', '8S', 'AC', 'AH'}114#     expected_results = 83115#     players = 2116#     _runner(my_cards, cards_on_table, players, expected_results)117#118#119# @pytest.mark.skipif(ON_CI)120# def test_montecarlo10():121#     """Montecarlo test"""122#     my_cards = {'3S', 'QH'}123#     cards_on_table = {'2C', '5H', '7C'}124#     expected_results = 30.9 + 2.2125#     players = 2126#     _runner(my_cards, cards_on_table, players, expected_results)127#128#129# @pytest.mark.skipif(ON_CI)130# def test_montecarlo11():131#     """Montecarlo test"""132#     my_cards = {'5C', 'JS'}133#     cards_on_table = {}134#     expected_results = 23135#     players = 4136#     _runner(my_cards, cards_on_table, players, expected_results)137#138#139# @pytest.mark.skipif(ON_CI)140# def test_montecarlo12():141#     """Montecarlo test"""142#     my_cards = {'TC', 'TH'}143#     cards_on_table = {'4D', 'QD', 'KC'}144#     expected_results = 66.7 + 0.38145#     players = 2146#     _runner(my_cards, cards_on_table, players, expected_results)147#148#149# @pytest.mark.skipif(ON_CI)150# def test_montecarlo13():151#     """Montecarlo test"""152#     my_cards = {'JH', 'QS'}153#     cards_on_table = {'5C', 'JD', 'AS', 'KS', 'QD'}154#     expected_results = 77155#     players = 2156#     _runner(my_cards, cards_on_table, players, expected_results)157#158#159# @pytest.mark.skipif(ON_CI)160# def test_montecarlo14():161#     """Montecarlo test"""162#     my_cards = {'2H', '8S'}163#     cards_on_table = {'AC', 'AD', 'AS', 'KS', 'KD'}164#     expected_results = 95165#     players = 2166#     _runner(my_cards, cards_on_table, players, expected_results)167#168#169# @pytest.mark.skipif(ON_CI)170# def test_montecarlo15():171#     """Montecarlo test"""172#     my_cards = {'KD', 'KS'}173#     cards_on_table = {'4D', '6S', '9C', '9S', 'TC'}174#     expected_results = 88175#     players = 2176#     _runner(my_cards, cards_on_table, players, expected_results)177#178#179# @pytest.mark.skipif(ON_CI)180# def test_montecarlo16():181#     """Montecarlo test"""182#     my_cards = {'5H', 'KD'}183#     cards_on_table = {'KH', 'JS', '2C', 'QS'}184#     expected_results = 75.6 + 3.6185#     players = 2186#     _runner(my_cards, cards_on_table, players, expected_results)187#188#189# @pytest.mark.skipif(ON_CI)190# def test_montecarlo17():191#     """Montecarlo test"""192#     my_cards = {'JD', 'JS'}193#     cards_on_table = {'8C', 'TC', 'JC', '5H', 'QC'}194#     expected_results = 26.1195#     players = 3196#     _runner(my_cards, cards_on_table, players, expected_results)197#198#199# @pytest.mark.skipif(ON_CI)200# def test_montecarlo19():201#     """Montecarlo test"""202#     my_cards = {'TD', '7D'}203#     cards_on_table = {'8D', 'QD', '7C', '5D', '6D'}204#     expected_results = 87205#     players = 2...

test_montecarlo_numpy.py

Source:test_montecarlo_numpy.py

1"""Test numpy based equity calculator"""2import pytest3from tools.montecarlo_numpy2 import numpy_montecarlo4def _runner(my_cards, cards_on_table, players, expected_result):5    """Montecarlo test"""6    equity = numpy_montecarlo(my_cards, cards_on_table, 10000, players)7    assert equity == pytest.approx(expected_result, abs=1)8@pytest.mark.skip("Failing")9def test_montecarlo1():10    """Montecarlo test"""11    my_cards = [['3H', '3S']]12    cards_on_table = ['8S', '4S', 'QH', '8C', '4H']13    expected_results = 40.214    players = 215    _runner(my_cards, cards_on_table, players, expected_results)16@pytest.mark.skip("Failing")17def test_montecarlo2():18    """Montecarlo test"""19    my_cards = [['8H', '8D']]20    cards_on_table = ['QH', '7H', '9H', 'JH', 'TH']21    expected_results = 95.622    players = 223    _runner(my_cards, cards_on_table, players, expected_results)24@pytest.mark.skip("Failing")25def test_montecarlo3():26    """Montecarlo test"""27    my_cards = [['AS', 'KS']]28    cards_on_table = []29    expected_results = 49.9 + 1.930    players = 331    _runner(my_cards, cards_on_table, players, expected_results)32@pytest.mark.skip("Failing")33def test_montecarlo4():34    """Montecarlo test"""35    my_cards = [['AS', 'KS']]36    cards_on_table = []37    expected_results = 66.1 + 1.638    players = 239    _runner(my_cards, cards_on_table, players, expected_results)40@pytest.mark.skip("Failing")41def test_montecarlo5():42    """Montecarlo test"""43    my_cards = [['8S', 'TS']]44    cards_on_table = ['8H', 'KS', '9S', 'TH', 'KH']45    expected_results = 71.5 + 5.946    players = 247    _runner(my_cards, cards_on_table, players, expected_results)48@pytest.mark.skip("Failing")49def test_montecarlo6():50    """Montecarlo test"""51    my_cards = [['8S', 'TS']]52    cards_on_table = ['2S', '3S', '4S', 'KS', 'AS']53    expected_results = 8754    players = 255    _runner(my_cards, cards_on_table, players, expected_results)56@pytest.mark.skip("Failing")57def test_montecarlo7():58    """Montecarlo test"""59    my_cards = [['8S', '2S']]60    cards_on_table = ['5S', '3S', '4S', 'KS', 'AS']61    expected_results = 10062    players = 263    _runner(my_cards, cards_on_table, players, expected_results)64@pytest.mark.skip("Failing")65def test_montecarlo8():66    """Montecarlo test"""67    my_cards = [['8S', 'TS']]68    cards_on_table = []69    expected_results = 22.6 + 2.970    players = 571    _runner(my_cards, cards_on_table, players, expected_results)72@pytest.mark.skip("Failing")73def test_montecarlo8b():74    """Montecarlo test"""75    my_cards = [['2C', 'QS']]76    cards_on_table = []77    expected_results = 49.6  # 45 win and 4 tie78    players = 279    _runner(my_cards, cards_on_table, players, expected_results)80@pytest.mark.skip("Failing")81def test_montecarlo9():82    """Montecarlo test"""83    my_cards = [['7H', '7S']]84    cards_on_table = ['7C', '8C', '8S', 'AC', 'AH']85    expected_results = 8386    players = 287    _runner(my_cards, cards_on_table, players, expected_results)88@pytest.mark.skip("Failing")89def test_montecarlo10():90    """Montecarlo test"""91    my_cards = [['3S', 'QH']]92    cards_on_table = ['2C', '5H', '7C']93    expected_results = 30.9 + 2.294    players = 295    _runner(my_cards, cards_on_table, players, expected_results)96@pytest.mark.skip("Failing")97def test_montecarlo11():98    """Montecarlo test"""99    my_cards = [['5C', 'JS']]100    cards_on_table = []101    expected_results = 23102    players = 4103    _runner(my_cards, cards_on_table, players, expected_results)104@pytest.mark.skip("Failing")105def test_montecarlo12():106    """Montecarlo test"""107    my_cards = [['TC', 'TH']]108    cards_on_table = ['4D', 'QD', 'KC']109    expected_results = 66.7 + 0.38110    players = 2111    _runner(my_cards, cards_on_table, players, expected_results)112@pytest.mark.skip("Failing")113def test_montecarlo13():114    """Montecarlo test"""115    my_cards = [['JH', 'QS']]116    cards_on_table = ['5C', 'JD', 'AS', 'KS', 'QD']117    expected_results = 77118    players = 2119    _runner(my_cards, cards_on_table, players, expected_results)120@pytest.mark.skip("Failing")121def test_montecarlo14():122    """Montecarlo test"""123    my_cards = [['2H', '8S']]124    cards_on_table = ['AC', 'AD', 'AS', 'KS', 'KD']125    expected_results = 95126    players = 2127    _runner(my_cards, cards_on_table, players, expected_results)128@pytest.mark.skip("Failing")129def test_montecarlo15():130    """Montecarlo test"""131    my_cards = [['KD', 'KS']]132    cards_on_table = ['4D', '6S', '9C', '9S', 'TC']133    expected_results = 88134    players = 2135    _runner(my_cards, cards_on_table, players, expected_results)136@pytest.mark.skip("Failing")137def test_montecarlo16():138    """Montecarlo test"""139    my_cards = [['5H', 'KD']]140    cards_on_table = ['KH', 'JS', '2C', 'QS']141    expected_results = 75.6 + 3.6142    players = 2143    _runner(my_cards, cards_on_table, players, expected_results)144@pytest.mark.skip("Failing")145def test_montecarlo17():146    """Montecarlo test"""147    my_cards = [['JD', 'JS']]148    cards_on_table = ['8C', 'TC', 'JC', '5H', 'QC']149    expected_results = 26.1150    players = 3151    _runner(my_cards, cards_on_table, players, expected_results)152@pytest.mark.skip("Failing")153def test_montecarlo19():154    """Montecarlo test"""155    my_cards = [['TD', '7D']]156    cards_on_table = ['8D', 'QD', '7C', '5D', '6D']157    expected_results = 87158    players = 2...

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.