Best Python code snippet using localstack_python
test_get_cloudify.py
Source:test_get_cloudify.py  
1########2# Copyright (c) 2014 GigaSpaces Technologies Ltd. All rights reserved3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain 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,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15############16from copy import copy17import importlib18import logging19import mock20import os21import shutil22from StringIO import StringIO23import sys24import tarfile25import tempfile26import testtools27import urllib28sys.path.append("../")29get_cloudify = importlib.import_module('get-cloudify')30class CliBuilderUnitTests(testtools.TestCase):31    """Unit tests for functions in get_cloudify.py"""32    def setUp(self):33        super(CliBuilderUnitTests, self).setUp()34        self.get_cloudify = get_cloudify35        self.get_cloudify.IS_VIRTUALENV = False36    def _create_dummy_requirements_tar(self, url, destination):37        tempdir = os.path.dirname(destination)38        fpath = self._generate_requirements_file(tempdir)39        try:40            tar = tarfile.open(name=destination, mode='w:gz')41            tar.add(name=tempdir, arcname='maindir')42            tar.close()43        finally:44            os.remove(fpath)45        return destination46    def _generate_requirements_file(self, path):47        fpath = os.path.join(path, 'dev-requirements.txt')48        with open(fpath, 'w') as f:49            f.write('sh==1.11\n')50        return fpath51    def test_validate_urls(self):52        self._validate_url(self.get_cloudify.PIP_URL)53        self._validate_url(self.get_cloudify.PYCR64_URL)54        self._validate_url(self.get_cloudify.PYCR32_URL)55    @staticmethod56    def _validate_url(url):57        try:58            status = urllib.urlopen(url).getcode()59            if not status == 200:60                raise AssertionError('url {} is not valid.'.format(url))61        except:62            raise AssertionError('url {} is not valid.'.format(url))63    def test_run_valid_command(self):64        proc = self.get_cloudify._run('echo Hi!')65        self.assertEqual(proc.returncode, 0, 'process execution failed')66    def test_run_invalid_command(self):67        builder_stdout = StringIO()68        # replacing builder stdout69        self.get_cloudify.sys.stdout = builder_stdout70        cmd = 'this is not a valid command'71        proc = self.get_cloudify._run(cmd)72        self.assertIsNot(proc.returncode, 0, 'command \'{}\' execution was '73                                             'expected to fail'.format(cmd))74    def test_installer_init_unexpected_argument(self):75        """Make sure typos in parse_args don't go un-noticed with **kwargs."""76        self.assertRaises(77            TypeError,78            self.get_cloudify.CloudifyInstaller,79            unknown_argument='unknown',80        )81    def test_installer_init_no_args(self):82        """Installer should initialise with no args."""83        self.get_cloudify.CloudifyInstaller()84    @mock.patch('get-cloudify.IS_WIN')85    def test_installer_init_pycrypto_not_windows(self, mock_win):86        """Installer init should complain with pycrypto not on windows."""87        # Original values will be restored by mock patch88        self.get_cloudify.IS_WIN = False89        self.assertRaises(90            self.get_cloudify.ArgumentNotValidForOS,91            self.get_cloudify.CloudifyInstaller,92            install_pycrypto=True,93        )94    @mock.patch('get-cloudify.IS_WIN')95    def test_installer_init_pycrypto_windows(self, mock_win):96        """Installer init should work with pycrypto on windows."""97        # Original values will be restored by mock patch98        self.get_cloudify.IS_WIN = True99        self.get_cloudify.CloudifyInstaller(100            install_pycrypto=True,101        )102    def test_installer_init_version_set(self):103        """Installer init should work with version set."""104        self.get_cloudify.CloudifyInstaller(105            version='1',106        )107    def test_installer_init_pre_set(self):108        """Installer init should work with pre set."""109        self.get_cloudify.CloudifyInstaller(110            pre=True,111        )112    def test_installer_init_source_set(self):113        """Installer init should work with source set."""114        self.get_cloudify.CloudifyInstaller(115            source='http://www.example.com/example.tar.gz',116        )117    def test_installer_init_version_and_pre_failure(self):118        """Installer init should fail with version and pre."""119        self.assertRaises(120            self.get_cloudify.ArgumentCombinationInvalid,121            self.get_cloudify.CloudifyInstaller,122            version='1',123            pre=True,124        )125    def test_installer_init_version_and_source_failure(self):126        """Installer init should fail with version and source."""127        self.assertRaises(128            self.get_cloudify.ArgumentCombinationInvalid,129            self.get_cloudify.CloudifyInstaller,130            version='1',131            source='http://www.example.com/example.tar.gz',132        )133    def test_installer_init_pre_and_source_failure(self):134        """Installer init should fail with pre and source."""135        self.assertRaises(136            self.get_cloudify.ArgumentCombinationInvalid,137            self.get_cloudify.CloudifyInstaller,138            pre=True,139            source='http://www.example.com/example.tar.gz',140        )141    def test_installer_init_pre_and_version_and_source_failure(self):142        """Installer init should fail with pre, version and source."""143        self.assertRaises(144            self.get_cloudify.ArgumentCombinationInvalid,145            self.get_cloudify.CloudifyInstaller,146            pre=True,147            version='1',148            source='http://www.example.com/example.tar.gz',149        )150    def test_is_module_installed(self):151        installer = self.get_cloudify.CloudifyInstaller()152        self.assertTrue(installer.is_installed('os'))153    def test_is_module_not_installed(self):154        installer = self.get_cloudify.CloudifyInstaller()155        self.assertFalse(installer.is_installed('not_real_module'))156    @mock.patch('get-cloudify.IS_LINUX')157    @mock.patch('get-cloudify.IS_WIN')158    @mock.patch('get-cloudify.IS_DARWIN')159    @mock.patch('get-cloudify._exit',160                side_effect=SystemExit)161    @mock.patch('get-cloudify.PLATFORM')162    def test_main_unsupported_os(self,163                                 mock_platform,164                                 mock_exit,165                                 mock_darwin,166                                 mock_win,167                                 mock_linux):168        # Original values will be restored by mock patch169        self.get_cloudify.IS_LINUX = False170        self.get_cloudify.IS_WIN = False171        self.get_cloudify.IS_DARWIN = False172        self.get_cloudify.PLATFORM = 'fake'173        self.assertRaises(174            SystemExit,175            self.get_cloudify.main,176        )177        mock_exit.assert_called_once_with(178            message='Platform fake not supported.',179            status='unsupported_platform',180        )181    @mock.patch('get-cloudify.IS_LINUX')182    @mock.patch('get-cloudify.IS_WIN')183    @mock.patch('get-cloudify.IS_DARWIN')184    @mock.patch('get-cloudify.logger')185    @mock.patch('get-cloudify.CloudifyInstaller')186    @mock.patch(187        'get-cloudify.parse_args',188        return_value={189            'quiet': False,190            'verbose': False,191        },192    )193    def test_main_linux(self,194                        mock_parse_args,195                        mock_installer,196                        mock_log,197                        mock_darwin,198                        mock_win,199                        mock_linux):200        # Original values will be restored by mock patch201        self.get_cloudify.IS_LINUX = True202        self.get_cloudify.IS_WIN = False203        self.get_cloudify.IS_DARWIN = False204        self.get_cloudify.main()205        mock_parse_args.assert_called_once_with()206        mock_log.setLevel.assert_called_once_with(logging.INFO)207        mock_installer.assert_called_once_with()208        mock_installer().execute.assert_called_once_with()209    @mock.patch('get-cloudify.IS_LINUX')210    @mock.patch('get-cloudify.IS_WIN')211    @mock.patch('get-cloudify.IS_DARWIN')212    @mock.patch('get-cloudify.logger')213    @mock.patch('get-cloudify.CloudifyInstaller')214    @mock.patch(215        'get-cloudify.parse_args',216        return_value={217            'quiet': False,218            'verbose': False,219        },220    )221    def test_main_windows(self,222                          mock_parse_args,223                          mock_installer,224                          mock_log,225                          mock_darwin,226                          mock_win,227                          mock_linux):228        # Original values will be restored by mock patch229        self.get_cloudify.IS_LINUX = False230        self.get_cloudify.IS_WIN = True231        self.get_cloudify.IS_DARWIN = False232        self.get_cloudify.main()233        mock_parse_args.assert_called_once_with()234        mock_log.setLevel.assert_called_once_with(logging.INFO)235        mock_installer.assert_called_once_with()236        mock_installer().execute.assert_called_once_with()237    @mock.patch('get-cloudify.IS_LINUX')238    @mock.patch('get-cloudify.IS_WIN')239    @mock.patch('get-cloudify.IS_DARWIN')240    @mock.patch('get-cloudify.logger')241    @mock.patch('get-cloudify.CloudifyInstaller')242    @mock.patch(243        'get-cloudify.parse_args',244        return_value={245            'quiet': False,246            'verbose': False,247        },248    )249    def test_main_darwin(self,250                         mock_parse_args,251                         mock_installer,252                         mock_log,253                         mock_darwin,254                         mock_win,255                         mock_linux):256        # Original values will be restored by mock patch257        self.get_cloudify.IS_LINUX = False258        self.get_cloudify.IS_WIN = False259        self.get_cloudify.IS_DARWIN = True260        self.get_cloudify.main()261        mock_parse_args.assert_called_once_with()262        mock_log.setLevel.assert_called_once_with(logging.INFO)263        mock_installer.assert_called_once_with()264        mock_installer().execute.assert_called_once_with()265    @mock.patch('get-cloudify.IS_LINUX')266    @mock.patch('get-cloudify.IS_WIN')267    @mock.patch('get-cloudify.IS_DARWIN')268    @mock.patch('get-cloudify.logger')269    @mock.patch('get-cloudify.CloudifyInstaller')270    @mock.patch(271        'get-cloudify.parse_args',272        return_value={273            'quiet': True,274            'verbose': False,275        },276    )277    def test_main_quiet_logging(self,278                                mock_parse_args,279                                mock_installer,280                                mock_log,281                                mock_darwin,282                                mock_win,283                                mock_linux):284        # Original values will be restored by mock patch285        self.get_cloudify.IS_LINUX = False286        self.get_cloudify.IS_WIN = False287        self.get_cloudify.IS_DARWIN = True288        self.get_cloudify.main()289        mock_parse_args.assert_called_once_with()290        mock_log.setLevel.assert_called_once_with(logging.ERROR)291        mock_installer.assert_called_once_with()292        mock_installer().execute.assert_called_once_with()293    @mock.patch('get-cloudify.IS_LINUX')294    @mock.patch('get-cloudify.IS_WIN')295    @mock.patch('get-cloudify.IS_DARWIN')296    @mock.patch('get-cloudify.logger')297    @mock.patch('get-cloudify.CloudifyInstaller')298    @mock.patch(299        'get-cloudify.parse_args',300        return_value={301            'quiet': False,302            'verbose': True,303        },304    )305    def test_main_verbose_logging(self,306                                  mock_parse_args,307                                  mock_installer,308                                  mock_log,309                                  mock_darwin,310                                  mock_win,311                                  mock_linux):312        # Original values will be restored by mock patch313        self.get_cloudify.IS_LINUX = False314        self.get_cloudify.IS_WIN = False315        self.get_cloudify.IS_DARWIN = True316        self.get_cloudify.main()317        mock_parse_args.assert_called_once_with()318        mock_log.setLevel.assert_called_once_with(logging.DEBUG)319        mock_installer.assert_called_once_with()320        mock_installer().execute.assert_called_once_with()321    @mock.patch('get-cloudify.IS_LINUX')322    @mock.patch('get-cloudify.IS_WIN')323    @mock.patch('get-cloudify.IS_DARWIN')324    @mock.patch('get-cloudify.logger')325    @mock.patch('get-cloudify.CloudifyInstaller')326    @mock.patch(327        'get-cloudify.parse_args',328        return_value={329            'quiet': False,330            'verbose': False,331            'fake_arg': 'this',332        },333    )334    def test_main_parsed_args_used(self,335                                   mock_parse_args,336                                   mock_installer,337                                   mock_log,338                                   mock_darwin,339                                   mock_win,340                                   mock_linux):341        # Original values will be restored by mock patch342        self.get_cloudify.IS_LINUX = False343        self.get_cloudify.IS_WIN = False344        self.get_cloudify.IS_DARWIN = True345        self.get_cloudify.main()346        mock_parse_args.assert_called_once_with()347        mock_log.setLevel.assert_called_once_with(logging.INFO)348        mock_installer.assert_called_once_with(349            fake_arg='this',350        )351        mock_installer().execute.assert_called_once_with()352    @mock.patch('get-cloudify.IS_WIN')353    def test_get_env_bin_path_windows(self, mock_win):354        # Original values will be restored by mock patch355        self.get_cloudify.IS_WIN = True356        result = self.get_cloudify._get_env_bin_path('path')357        self.assertEquals(os.path.join('path', 'scripts'), result)358    @mock.patch('get-cloudify.IS_WIN')359    def test_get_env_bin_path_not_windows(self, mock_win):360        # Original values will be restored by mock patch361        self.get_cloudify.IS_WIN = False362        result = self.get_cloudify._get_env_bin_path('path')363        self.assertEquals(os.path.join('path', 'bin'), result)364    @mock.patch('get-cloudify.os')365    def test_is_root(self, mock_os):366        mock_os.getuid.return_value = 0367        self.assertTrue(self.get_cloudify._is_root())368    @mock.patch('get-cloudify.os')369    def test_not_is_root(self, mock_os):370        mock_os.getuid.return_value = 1371        self.assertFalse(self.get_cloudify._is_root())372    @mock.patch('get-cloudify._is_root')373    @mock.patch('get-cloudify.logger')374    @mock.patch('get-cloudify.os')375    def test_drop_root_privileges_from_sudo(self,376                                            mock_os,377                                            mock_log,378                                            mock_root_check):379        mock_root_check.return_value = True380        type(mock_os).environ = mock.PropertyMock(381            return_value={382                'SUDO_GID': 12345,383                'SUDO_UID': 54321,384            },385        )386        self.get_cloudify._drop_root_privileges()387        mock_log.info.assert_called_once_with('Dropping root permissions...')388        mock_os.setegid.assert_called_once_with(12345)389        mock_os.seteuid.assert_called_once_with(54321)390    @mock.patch('get-cloudify._is_root')391    @mock.patch('get-cloudify.logger')392    @mock.patch('get-cloudify.os')393    def test_drop_root_privileges_without_sudo(self,394                                               mock_os,395                                               mock_log,396                                               mock_root_check):397        mock_root_check.return_value = True398        type(mock_os).environ = mock.PropertyMock(399            return_value={},400        )401        self.get_cloudify._drop_root_privileges()402        mock_log.info.assert_called_once_with('Dropping root permissions...')403        mock_os.setegid.assert_called_once_with(0)404        mock_os.seteuid.assert_called_once_with(0)405    @mock.patch('get-cloudify._is_root')406    @mock.patch('get-cloudify.logger')407    @mock.patch('get-cloudify.os')408    def test_no_drop_root_privileges(self,409                                     mock_os,410                                     mock_log,411                                     mock_root_check):412        mock_root_check.return_value = False413        self.get_cloudify._drop_root_privileges()414        self.assertFalse(mock_log.info.called)415        self.assertFalse(mock_os.setegid.called)416        self.assertFalse(mock_os.seteuid.called)417    @mock.patch('get-cloudify.sys.exit')418    @mock.patch('get-cloudify.logger')419    def test_exit_unsupported_os(self,420                                 mock_log,421                                 mock_exit):422        self.get_cloudify._exit(423            message='Unsupported OS',424            status='unsupported_platform',425        )426        mock_log.error.assert_called_once_with('Unsupported OS')427        mock_exit.assert_called_once_with(200)428    @mock.patch('get-cloudify.sys.exit')429    @mock.patch('get-cloudify.logger')430    def test_exit_venv_create_fail(self,431                                   mock_log,432                                   mock_exit):433        self.get_cloudify._exit(434            message='Venv creation failure',435            status='virtualenv_creation_failure',436        )437        mock_log.error.assert_called_once_with('Venv creation failure')438        mock_exit.assert_called_once_with(210)439    @mock.patch('get-cloudify.sys.exit')440    @mock.patch('get-cloudify.logger')441    def test_exit_dep_download_fail(self,442                                    mock_log,443                                    mock_exit):444        self.get_cloudify._exit(445            message='Download failure',446            status='dependency_download_failure',447        )448        mock_log.error.assert_called_once_with('Download failure')449        mock_exit.assert_called_once_with(220)450    @mock.patch('get-cloudify.sys.exit')451    @mock.patch('get-cloudify.logger')452    def test_exit_dep_extract_fail(self,453                                   mock_log,454                                   mock_exit):455        self.get_cloudify._exit(456            message='Extraction failure',457            status='dependency_extraction_failure',458        )459        mock_log.error.assert_called_once_with('Extraction failure')460        mock_exit.assert_called_once_with(221)461    @mock.patch('get-cloudify.sys.exit')462    @mock.patch('get-cloudify.logger')463    def test_exit_dep_install_fail(self,464                                   mock_log,465                                   mock_exit):466        self.get_cloudify._exit(467            message='Install failure',468            status='dependency_installation_failure',469        )470        mock_log.error.assert_called_once_with('Install failure')471        mock_exit.assert_called_once_with(222)472    @mock.patch('get-cloudify.sys.exit')473    @mock.patch('get-cloudify.logger')474    def test_exit_dep_unsupported_distro(self,475                                         mock_log,476                                         mock_exit):477        self.get_cloudify._exit(478            message='Wrong distro',479            status='dependency_unsupported_on_distribution',480        )481        mock_log.error.assert_called_once_with('Wrong distro')482        mock_exit.assert_called_once_with(223)483    @mock.patch('get-cloudify.sys.exit')484    @mock.patch('get-cloudify.logger')485    def test_exit_cloudify_already_uninstalled(self,486                                               mock_log,487                                               mock_exit):488        self.get_cloudify._exit(489            message='Cloudify already here',490            status='cloudify_already_installed',491        )492        mock_log.error.assert_called_once_with('Cloudify already here')493        mock_exit.assert_called_once_with(230)494    @mock.patch('get-cloudify._exit',495                side_effect=SystemExit)496    @mock.patch('get-cloudify._download_file',497                side_effect=StandardError('Boom!'))498    @mock.patch('get-cloudify.CloudifyInstaller.is_installed',499                return_value=False)500    def test_install_pip_failed_download(self,501                                         mock_find_pip,502                                         mock_download,503                                         mock_exit):504        installer = self.get_cloudify.CloudifyInstaller()505        self.assertRaises(506            SystemExit,507            installer.get_pip,508        )509        mock_exit.assert_called_once_with(510            message='Failed pip download from {0}. (Boom!)'.format(511                get_cloudify.PIP_URL,512            ),513            status='dependency_download_failure',514        )515    @mock.patch('get-cloudify._exit',516                side_effect=SystemExit)517    @mock.patch('get-cloudify._download_file',518                return_value=None)519    @mock.patch('get-cloudify.CloudifyInstaller.is_installed',520                return_value=False)521    def test_install_pip_fail(self,522                              mock_find_pip,523                              mock_download,524                              mock_exit):525        python_path = 'non_existing_path'526        installer = self.get_cloudify.CloudifyInstaller(527            python_path=python_path,528        )529        self.assertRaises(530            SystemExit,531            installer.get_pip,532        )533        mock_exit.assert_called_once_with(534            message='Could not install pip',535            status='dependency_installation_failure',536        )537    @mock.patch('get-cloudify._exit',538                side_effect=SystemExit)539    def test_make_virtualenv_fail(self, mock_exit):540        self.assertRaises(541            SystemExit,542            self.get_cloudify._make_virtualenv,543            '/path/to/dir',544            'non_existing_path',545        )546        mock_exit.assert_called_once_with(547            message='Could not create virtualenv: /path/to/dir',548            status='virtualenv_creation_failure',549        )550    @mock.patch('get-cloudify._exit',551                side_effect=SystemExit)552    def test_install_non_existing_module(self, mock_exit):553        self.assertRaises(554            SystemExit,555            self.get_cloudify._install_package,556            'nonexisting_module',557        )558        mock_exit.assert_called_once_with(559            message='Could not install package: nonexisting_module.',560            status='dependency_installation_failure',561        )562    @mock.patch('get-cloudify.logger')563    @mock.patch('get-cloudify._run')564    @mock.patch('get-cloudify.IS_VIRTUALENV')565    def test_install_in_existing_venv_no_path(self,566                                              mock_is_venv,567                                              mock_run,568                                              mock_log):569        # Original value will be restored by mock patch570        self.get_cloudify.IS_VIRTUALENV = True571        type(mock_run.return_value).returncode = mock.PropertyMock(572            return_value=0,573        )574        self.get_cloudify._install_package('test-package')575        expected_info_log_calls = [576            mock.call('Installing test-package...'),577            mock.call('Installing within current virtualenv.'),578        ]579        mock_log.info.assert_has_calls(expected_info_log_calls)580        self.assertEquals(2, mock_log.info.call_count)581        mock_run.assert_called_once_with(582            'pip install test-package',583        )584    @mock.patch('get-cloudify.logger')585    @mock.patch('get-cloudify._run')586    @mock.patch('get-cloudify.IS_VIRTUALENV')587    @mock.patch('get-cloudify._get_env_bin_path')588    def test_install_with_venv_path(self,589                                    mock_bin_path,590                                    mock_is_venv,591                                    mock_run,592                                    mock_log):593        # Original value will be restored by mock patch594        self.get_cloudify.IS_VIRTUALENV = False595        type(mock_run.return_value).returncode = mock.PropertyMock(596            return_value=0,597        )598        test_package = 'test-package'599        if self.get_cloudify.IS_WIN:600            venv_path = 'C:\\my\\venv'601            venv_bin_path = 'C:\\my\\venv\\scripts'602            expected_call = '{path}\\scripts\\pip install {package}'603        else:604            venv_path = '/my/venv'605            venv_bin_path = '/my/venv/bin'606            expected_call = '{path}/bin/pip install {package}'607        expected_call = expected_call.format(608            path=venv_path,609            package=test_package,610        )611        mock_bin_path.return_value = venv_bin_path612        self.get_cloudify._install_package('test-package',613                                           virtualenv_path=venv_path)614        mock_log.info.assert_called_once_with(615            'Installing test-package...',616        )617        mock_run.assert_called_once_with(expected_call)618    @mock.patch('get-cloudify.logger')619    @mock.patch('get-cloudify._run')620    @mock.patch('get-cloudify.IS_VIRTUALENV')621    @mock.patch('get-cloudify._get_env_bin_path')622    def test_install_with_venv_path_ignores_current_venv(self,623                                                         mock_bin_path,624                                                         mock_is_venv,625                                                         mock_run,626                                                         mock_log):627        # Original value will be restored by mock patch628        self.get_cloudify.IS_VIRTUALENV = True629        type(mock_run.return_value).returncode = mock.PropertyMock(630            return_value=0,631        )632        test_package = 'test-package'633        if self.get_cloudify.IS_WIN:634            venv_path = 'C:\\my\\venv'635            venv_bin_path = 'C:\\my\\venv\\scripts'636            expected_call = '{path}\\scripts\\pip install {package}'637        else:638            venv_path = '/my/venv'639            venv_bin_path = '/my/venv/bin'640            expected_call = '{path}/bin/pip install {package}'641        expected_call = expected_call.format(642            path=venv_path,643            package=test_package,644        )645        mock_bin_path.return_value = venv_bin_path646        self.get_cloudify._install_package('test-package',647                                           virtualenv_path=venv_path)648        mock_log.info.assert_called_once_with(649            'Installing test-package...',650        )651        mock_run.assert_called_once_with(expected_call)652    def test_get_os_props(self):653        if not self.get_cloudify.IS_WIN:654            distro = self.get_cloudify._get_os_props()[0]655            distros = ('ubuntu', 'redhat', 'debian', 'fedora', 'centos',656                       'archlinux')657            if distro.lower() not in distros:658                self.fail('distro prop \'{0}\' should be equal to one of: '659                          '{1}'.format(distro, distros))660    def test_download_file(self):661        self.get_cloudify.VERBOSE = True662        # We cannot use NamedTemporaryFile here as it will fail on Windows663        # where a NamedTemporaryFile can only be written to using the file664        # handle returned from the NamedTemporaryFile call665        tmpdir = tempfile.mkdtemp()666        tmp_file_name = os.path.join(tmpdir, 'testfile')667        self.get_cloudify._download_file('http://www.google.com',668                                         tmp_file_name)669        with open(tmp_file_name) as f:670            content = f.readlines()671            self.assertIsNotNone(content)672        shutil.rmtree(tmpdir)673    def test_check_cloudify_not_installed_in_venv(self):674        tmp_venv = tempfile.mkdtemp()675        # Handle windows and linux by using correct python path676        # sys.executable would be nicer, but appveyor stopped showing logs677        # when using that678        if self.get_cloudify.IS_WIN:679            python_path = 'c:\\python27\python.exe'680        else:681            python_path = 'python'682        installer = self.get_cloudify.CloudifyInstaller(virtualenv=tmp_venv)683        try:684            self.get_cloudify._make_virtualenv(tmp_venv, python_path)685            self.assertFalse(686                installer.check_cloudify_installed()687            )688        finally:689            shutil.rmtree(tmp_venv)690    def test_check_cloudify_installed_in_venv(self):691        tmp_venv = tempfile.mkdtemp()692        # Handle windows and linux by using correct python path693        # sys.executable would be nicer, but appveyor stopped showing logs694        # when using that695        if self.get_cloudify.IS_WIN:696            python_path = 'c:\\python27\python.exe'697        else:698            python_path = 'python'699        try:700            self.get_cloudify._make_virtualenv(tmp_venv, python_path)701            installer = get_cloudify.CloudifyInstaller(virtualenv=tmp_venv)702            installer.execute()703            self.assertTrue(704                installer.check_cloudify_installed()705            )706        finally:707            shutil.rmtree(tmp_venv)708    def test_get_requirements_from_source_url(self):709        def get(url, destination):710            return self._create_dummy_requirements_tar(url, destination)711        self.get_cloudify._download_file = get712        try:713            installer = self.get_cloudify.CloudifyInstaller()714            req_list = installer._get_default_requirement_files('null')715            self.assertEquals(len(req_list), 1)716            self.assertIn('dev-requirements.txt', req_list[0])717        finally:718            self.get_cloudify._download_file = get_cloudify._download_file719    def test_get_requirements_from_source_path(self):720        tempdir = tempfile.mkdtemp()721        self._generate_requirements_file(tempdir)722        try:723            installer = self.get_cloudify.CloudifyInstaller()724            req_list = installer._get_default_requirement_files(tempdir)725            self.assertEquals(len(req_list), 1)726            self.assertIn('dev-requirements.txt', req_list[0])727        finally:728            shutil.rmtree(tempdir)729class TestArgParser(testtools.TestCase):730    """Unit tests for functions in get_cloudify.py"""731    def setUp(self):732        super(TestArgParser, self).setUp()733        self.get_cloudify = get_cloudify734        self.get_cloudify.IS_VIRTUALENV = False735        self.expected_args = {736            'verbose': False,737            'quiet': False,738            'version': None,739            'pre': False,740            'source': None,741            'force': False,742            'virtualenv': None,743            'install_pip': False,744            'install_virtualenv': False,745            'with_requirements': None,746            'upgrade': False,747            'pip_args': None,748        }749        self.expected_repo_url = \750            'https://github.com/{user}/cloudify-cli/archive/{branch}.tar.gz'751        self.expected_repo_default_user = 'cloudify-cosmo'752    @mock.patch('get-cloudify.IS_LINUX')753    @mock.patch('get-cloudify.IS_WIN')754    @mock.patch('get-cloudify.IS_DARWIN')755    def test_expected_argsr_linux(self,756                                  mock_linux,757                                  mock_win,758                                  mock_darwin):759        # Original values will be restored by mock patch760        self.get_cloudify.IS_LINUX = True761        self.get_cloudify.IS_WIN = False762        self.get_cloudify.IS_DARWIN = False763        args = self.get_cloudify.parse_args([])764        expected_args = copy(self.expected_args)765        expected_args['install_pythondev'] = False766        self.assertEquals(expected_args, args)767    @mock.patch('get-cloudify.IS_LINUX')768    @mock.patch('get-cloudify.IS_WIN')769    @mock.patch('get-cloudify.IS_DARWIN')770    def test_expected_args_windows(self,771                                   mock_linux,772                                   mock_win,773                                   mock_darwin):774        # Original values will be restored by mock patch775        self.get_cloudify.IS_LINUX = False776        self.get_cloudify.IS_WIN = True777        self.get_cloudify.IS_DARWIN = False778        args = self.get_cloudify.parse_args([])779        expected_args = copy(self.expected_args)780        expected_args['install_pycrypto'] = False781        self.assertEquals(expected_args, args)782    @mock.patch('get-cloudify.IS_LINUX')783    @mock.patch('get-cloudify.IS_WIN')784    @mock.patch('get-cloudify.IS_DARWIN')785    def test_expected_args_darwin(self,786                                  mock_linux,787                                  mock_win,788                                  mock_darwin):789        # Original values will be restored by mock patch790        self.get_cloudify.IS_LINUX = False791        self.get_cloudify.IS_WIN = False792        self.get_cloudify.IS_DARWIN = True793        args = self.get_cloudify.parse_args([])794        self.assertEquals(self.expected_args, args)795    @mock.patch('get-cloudify.IS_LINUX')796    @mock.patch('get-cloudify.IS_WIN')797    @mock.patch('get-cloudify.IS_DARWIN')798    @mock.patch('get-cloudify.logger')799    def test_deprecated_withrequirements(self,800                                         mock_log,801                                         mock_linux,802                                         mock_win,803                                         mock_darwin):804        # Original values will be restored by mock patch805        self.get_cloudify.IS_LINUX = True806        self.get_cloudify.IS_WIN = False807        self.get_cloudify.IS_DARWIN = False808        # Source required when using with-requirements809        args = self.get_cloudify.parse_args([810            '--source=notreal',811            '--withrequirements=test',812        ])813        self.assertEquals(['test'], args['with_requirements'])814        mock_log.warning.assert_called_once_with(815            '--withrequirements is deprecated. Use --with-requirements. '816            '--withrequirements will be removed in a future release.'817        )818    @mock.patch('get-cloudify.IS_LINUX')819    @mock.patch('get-cloudify.IS_WIN')820    @mock.patch('get-cloudify.IS_DARWIN')821    @mock.patch('get-cloudify.logger')822    def test_deprecated_installvirtualenv(self,823                                          mock_log,824                                          mock_linux,825                                          mock_win,826                                          mock_darwin):827        # Original values will be restored by mock patch828        self.get_cloudify.IS_LINUX = True829        self.get_cloudify.IS_WIN = False830        self.get_cloudify.IS_DARWIN = False831        args = self.get_cloudify.parse_args(['--installvirtualenv'])832        self.assertTrue(args['install_virtualenv'])833        mock_log.warning.assert_called_once_with(834            '--installvirtualenv is deprecated. Use --install-virtualenv. '835            '--installvirtualenv will be removed in a future release.'836        )837    @mock.patch('get-cloudify.IS_LINUX')838    @mock.patch('get-cloudify.IS_WIN')839    @mock.patch('get-cloudify.IS_DARWIN')840    @mock.patch('get-cloudify.logger')841    def test_deprecated_installpip(self,842                                   mock_log,843                                   mock_linux,844                                   mock_win,845                                   mock_darwin):846        # Original values will be restored by mock patch847        self.get_cloudify.IS_LINUX = True848        self.get_cloudify.IS_WIN = False849        self.get_cloudify.IS_DARWIN = False850        args = self.get_cloudify.parse_args(['--installpip'])851        self.assertTrue(args['install_pip'])852        mock_log.warning.assert_called_once_with(853            '--installpip is deprecated. Use --install-pip. '854            '--installpip will be removed in a future release.'855        )856    @mock.patch('get-cloudify.IS_LINUX')857    @mock.patch('get-cloudify.IS_WIN')858    @mock.patch('get-cloudify.IS_DARWIN')859    @mock.patch('get-cloudify.logger')860    def test_deprecated_installpycrypto(self,861                                        mock_log,862                                        mock_linux,863                                        mock_win,864                                        mock_darwin):865        # Original values will be restored by mock patch866        self.get_cloudify.IS_LINUX = False867        self.get_cloudify.IS_WIN = True868        self.get_cloudify.IS_DARWIN = False869        args = self.get_cloudify.parse_args(['--installpycrypto'])870        self.assertTrue(args['install_pycrypto'])871        mock_log.warning.assert_called_once_with(872            '--installpycrypto is deprecated. Use --install-pycrypto. '873            '--installpycrypto will be removed in a future release.'874        )875    @mock.patch('get-cloudify.IS_LINUX')876    @mock.patch('get-cloudify.IS_WIN')877    @mock.patch('get-cloudify.IS_DARWIN')878    @mock.patch('get-cloudify.logger')879    def test_deprecated_installpythondev(self,880                                         mock_log,881                                         mock_linux,882                                         mock_win,883                                         mock_darwin):884        # Original values will be restored by mock patch885        self.get_cloudify.IS_LINUX = True886        self.get_cloudify.IS_WIN = False887        self.get_cloudify.IS_DARWIN = False888        args = self.get_cloudify.parse_args(['--installpythondev'])889        self.assertTrue(args['install_pythondev'])890        mock_log.warning.assert_called_once_with(891            '--installpythondev is deprecated. Use --install-pythondev. '892            '--installpythondev will be removed in a future release.'893        )894    @mock.patch('get-cloudify.IS_LINUX')895    @mock.patch('get-cloudify.IS_WIN')896    @mock.patch('get-cloudify.IS_DARWIN')897    @mock.patch('get-cloudify.logger')898    def test_deprecated_pythonpath(self,899                                   mock_log,900                                   mock_linux,901                                   mock_win,902                                   mock_darwin):903        # Original values will be restored by mock patch904        self.get_cloudify.IS_LINUX = True905        self.get_cloudify.IS_WIN = False906        self.get_cloudify.IS_DARWIN = False907        args = self.get_cloudify.parse_args(['--pythonpath=test'])908        self.assertEquals('test', args['python_path'])909        mock_log.warning.assert_called_once_with(910            '--pythonpath is deprecated. '911            'To use a different interpreter, run this script with '912            'your preferred interpreter and that interpreter will be '913            'used.'914        )915    @mock.patch('get-cloudify.IS_LINUX')916    @mock.patch('get-cloudify.IS_WIN')917    @mock.patch('get-cloudify.IS_DARWIN')918    @mock.patch('get-cloudify.logger')919    def test_deprecated_forceonline(self,920                                    mock_log,921                                    mock_linux,922                                    mock_win,923                                    mock_darwin):924        # Original values will be restored by mock patch925        self.get_cloudify.IS_LINUX = True926        self.get_cloudify.IS_WIN = False927        self.get_cloudify.IS_DARWIN = False928        args = self.get_cloudify.parse_args(['--forceonline'])929        # This should currently be completely ignored930        self.assertNotIn('force_online', args)931        self.assertNotIn('forceonline', args)932        mock_log.warning.assert_called_once_with(933            '--forceonline is deprecated. '934            'Online install is currently the only option, so this '935            'argument will be ignored.'936        )937    @mock.patch('get-cloudify.IS_LINUX')938    @mock.patch('get-cloudify.IS_WIN')939    @mock.patch('get-cloudify.IS_DARWIN')940    def test_args_chosen(self,941                         mock_linux,942                         mock_win,943                         mock_darwin):944        """Check that parse_args actually sets arguments."""945        # Original values will be restored by mock patch946        self.get_cloudify.IS_LINUX = True947        self.get_cloudify.IS_WIN = False948        self.get_cloudify.IS_DARWIN = False949        set_args = self.get_cloudify.parse_args(['-f',950                                                 '--virtualenv=venv_path',951                                                 '--quiet',952                                                 '--version=3.2',953                                                 '--install-pip',954                                                 '--install-pythondev'])955        self.assertTrue(set_args['force'])956        self.assertTrue(set_args['install_pip'])957        self.assertTrue(set_args['quiet'])958        self.assertEqual(set_args['version'], '3.2')959        self.assertEqual(set_args['virtualenv'], 'venv_path')960    @mock.patch('get-cloudify.IS_LINUX')961    @mock.patch('get-cloudify.IS_WIN')962    @mock.patch('get-cloudify.IS_DARWIN')963    @mock.patch('get-cloudify.argparse.ArgumentParser.error',964                side_effect=SystemExit)965    def test_with_requirements_alone_fails(self,966                                           mock_parse_error,967                                           mock_linux,968                                           mock_win,969                                           mock_darwin):970        """with_requirements should fail when used alone"""971        # Original values will be restored by mock patch972        self.get_cloudify.IS_LINUX = True973        self.get_cloudify.IS_WIN = False974        self.get_cloudify.IS_DARWIN = False975        self.assertRaises(976            SystemExit,977            self.get_cloudify.parse_args,978            ['--with-requirements=test'],979        )980        mock_parse_error.assert_called_once_with(981            '--source or --use-branch is required when '982            'calling with --with-requirements.'983        )984    @mock.patch('get-cloudify.IS_LINUX')985    @mock.patch('get-cloudify.IS_WIN')986    @mock.patch('get-cloudify.IS_DARWIN')987    def test_with_requirements_and_use_branch(self,988                                              mock_linux,989                                              mock_win,990                                              mock_darwin):991        """with_requirements with use_branch should not fail"""992        # Original values will be restored by mock patch993        self.get_cloudify.IS_LINUX = True994        self.get_cloudify.IS_WIN = False995        self.get_cloudify.IS_DARWIN = False996        # No need to assert, we're just happy not to have errors997        self.get_cloudify.parse_args(['--use-branch=test',998                                      '--with-requirements=test'])999    @mock.patch('get-cloudify.IS_LINUX')1000    @mock.patch('get-cloudify.IS_WIN')1001    @mock.patch('get-cloudify.IS_DARWIN')1002    def test_with_requirements_and_source(self,1003                                          mock_linux,1004                                          mock_win,1005                                          mock_darwin):1006        """with_requirements with source should not fail"""1007        # Original values will be restored by mock patch1008        self.get_cloudify.IS_LINUX = True1009        self.get_cloudify.IS_WIN = False1010        self.get_cloudify.IS_DARWIN = False1011        # No need to assert, we're just happy not to have errors1012        self.get_cloudify.parse_args(['--source=test',1013                                      '--with-requirements=test'])1014    @mock.patch('get-cloudify.IS_LINUX')1015    @mock.patch('get-cloudify.IS_WIN')1016    @mock.patch('get-cloudify.IS_DARWIN')1017    @mock.patch('get-cloudify.logger')1018    def test_source_with_no_requirements(self,1019                                         mock_log,1020                                         mock_linux,1021                                         mock_win,1022                                         mock_darwin):1023        """source should warn if called without requirements"""1024        # It's quite likely to be an error for anything but the most trivial1025        # changes1026        # Original values will be restored by mock patch1027        self.get_cloudify.IS_LINUX = True1028        self.get_cloudify.IS_WIN = False1029        self.get_cloudify.IS_DARWIN = False1030        self.get_cloudify.parse_args([1031            '--source=test',1032        ])1033        mock_log.warning.assert_called_once_with(1034            'A source URL or branch was specified, but '1035            '--with-requirements was omitted. You may need to retry using '1036            '--with-requirements if the installation fails.'1037        )1038    @mock.patch('get-cloudify.IS_LINUX')1039    @mock.patch('get-cloudify.IS_WIN')1040    @mock.patch('get-cloudify.IS_DARWIN')1041    @mock.patch('get-cloudify.logger')1042    def test_branch_with_no_requirements(self,1043                                         mock_log,1044                                         mock_linux,1045                                         mock_win,1046                                         mock_darwin):1047        """use_branch should warn if called without requirements"""1048        # It's quite likely to be an error for anything but the most trivial1049        # changes1050        # Original values will be restored by mock patch1051        self.get_cloudify.IS_LINUX = True1052        self.get_cloudify.IS_WIN = False1053        self.get_cloudify.IS_DARWIN = False1054        self.get_cloudify.parse_args([1055            '--use-branch=test',1056        ])1057        mock_log.warning.assert_called_once_with(1058            'A source URL or branch was specified, but '1059            '--with-requirements was omitted. You may need to retry using '1060            '--with-requirements if the installation fails.'1061        )1062    @mock.patch('get-cloudify.IS_LINUX')1063    @mock.patch('get-cloudify.IS_WIN')1064    @mock.patch('get-cloudify.IS_DARWIN')1065    @mock.patch('get-cloudify.argparse.ArgumentParser.error',1066                side_effect=SystemExit)1067    def test_use_branch_invalid_format(self,1068                                       mock_parse_error,1069                                       mock_linux,1070                                       mock_win,1071                                       mock_darwin):1072        """use_branch should not work with more than one slash"""1073        # Original values will be restored by mock patch1074        self.get_cloudify.IS_LINUX = True1075        self.get_cloudify.IS_WIN = False1076        self.get_cloudify.IS_DARWIN = False1077        self.assertRaises(1078            SystemExit,1079            self.get_cloudify.parse_args,1080            ['--use-branch=test/this/fails'],1081        )1082        mock_parse_error.assert_called_once_with(1083            '--use-branch should be specified either as '1084            '<branch> or as <user>/<branch>. '1085            'Too many "/" found in arguments.'1086        )1087    @mock.patch('get-cloudify.IS_LINUX')1088    @mock.patch('get-cloudify.IS_WIN')1089    @mock.patch('get-cloudify.IS_DARWIN')1090    @mock.patch('get-cloudify.logger')1091    def test_use_branch_just_branch(self,1092                                    mock_log,1093                                    mock_linux,1094                                    mock_win,1095                                    mock_darwin):1096        """use_branch without slash should just set branch"""1097        # Original values will be restored by mock patch1098        self.get_cloudify.IS_LINUX = True1099        self.get_cloudify.IS_WIN = False1100        self.get_cloudify.IS_DARWIN = False1101        args = self.get_cloudify.parse_args([1102            '--use-branch=test',1103        ])1104        expected_repo_url = self.expected_repo_url.format(1105            user=self.expected_repo_default_user,1106            branch='test',1107        )1108        self.assertEquals(expected_repo_url, args['source'])1109    @mock.patch('get-cloudify.IS_LINUX')1110    @mock.patch('get-cloudify.IS_WIN')1111    @mock.patch('get-cloudify.IS_DARWIN')1112    def test_use_branch_user_and_branch(self,1113                                        mock_linux,1114                                        mock_win,1115                                        mock_darwin):1116        """use_branch with slash should set user and branch"""1117        # Original values will be restored by mock patch1118        self.get_cloudify.IS_LINUX = True1119        self.get_cloudify.IS_WIN = False1120        self.get_cloudify.IS_DARWIN = False1121        args = self.get_cloudify.parse_args([1122            '--use-branch=user/branch',1123        ])1124        expected_repo_url = self.expected_repo_url.format(1125            user='user',1126            branch='branch',1127        )1128        self.assertEquals(expected_repo_url, args['source'])1129    def test_mutually_exclusive_verbosity(self):1130        ex = self.assertRaises(1131            SystemExit, self.get_cloudify.parse_args, ['--verbose', '--quiet'])1132        exit_code = ex.message1133        self.assertEqual(2, exit_code)1134    def test_mutually_exclusive_pre_version(self):1135        ex = self.assertRaises(1136            SystemExit, self.get_cloudify.parse_args, ['--version', '--pre'])1137        exit_code = ex.message1138        self.assertEqual(2, exit_code)1139    def test_mutually_exclusive_version_source(self):1140        ex = self.assertRaises(1141            SystemExit,1142            self.get_cloudify.parse_args,1143            ['--version', '--source=test'],1144        )1145        exit_code = ex.message1146        self.assertEqual(2, exit_code)1147    def test_mutually_exclusive_pre_source(self):1148        ex = self.assertRaises(1149            SystemExit,1150            self.get_cloudify.parse_args,1151            ['--pre', '--source=test'],1152        )1153        exit_code = ex.message1154        self.assertEqual(2, exit_code)1155    @mock.patch('get-cloudify.IS_LINUX')1156    @mock.patch('get-cloudify.IS_WIN')1157    @mock.patch('get-cloudify.IS_DARWIN')1158    def test_listvar_with_requirements(self,1159                                       mock_linux,1160                                       mock_win,1161                                       mock_darwin):1162        """We are expecting to be able to provide multiple requirements"""1163        # Original values will be restored by mock patch1164        self.get_cloudify.IS_LINUX = True1165        self.get_cloudify.IS_WIN = False1166        self.get_cloudify.IS_DARWIN = False1167        args = self.get_cloudify.parse_args([1168            '--source=requiredwithargs',1169            '--with-requirements', 'test.txt', 'test2.txt',1170        ])1171        self.assertEquals(1172            ['test.txt', 'test2.txt'],1173            args['with_requirements'],1174        )1175class ArgsObject(object):...passGen.py
Source:passGen.py  
1import pickle2import random3import socket4import time5import threading6import errno7import os8import fcntl9import paramiko10import getpass11IP = ["94.106.244.227", "127.0.0.1"]12PORT = 765413BUFFERSIZE = 10014class bcolors:15    HEADER = '\033[95m'16    OKBLUE = '\033[94m'17    OKCYAN = '\033[96m'18    OKGREEN = '\033[92m'19    WARNING = '\033[93m'20    FAIL = '\033[91m'21    ENDC = '\033[0m'22    BOLD = '\033[1m'23    UNDERLINE = '\033[4m'24def passwordEncoder(password_holder):25    encoded_password_holder = []26    password_encoded = ""27    for password in password_holder:28        for caracter in password:29            caractere_holder = ord(caracter) + 4730            password_encoded += chr(caractere_holder)31        encoded_password_holder.append(password_encoded)32        password_encoded = ""33    return encoded_password_holder34def check_linux():35    platform = os.name36    if platform == "posix" or platform == "Linux" or platform == "Darwin":37        return True38    else:39        return False40def commandInput():41    global is_linux42    if is_linux == True:43        command = input(bcolors.OKGREEN+ "passGen >> " + bcolors.ENDC)44        return command45    else:46        command = input("passGen >> ")47        return command48def commandWorker():49    try:50        while True:51            global server, random_IP52            command = commandInput()53            args = command.split(' ')54            erreur_syntax = False55            if args[0] == "gen" or args[0] == "Gen":56                crashed = connexionStatus(server_status)57                if not crashed:58                    if len(args) == 1:59                        length = random.randint(1, 10)60                        password_counter = random.randint(1,20)61                        erreur_syntax = False62                        passData = [args[0], password_counter, length]63                        passData = pickle.dumps(passData)64                        server.send(passData)65                    else:66                        try:67                            password_counter = int(args[1])68                            length = int(args[2])69                            if password_counter >= 30:70                                if is_linux:71                                    print(bcolors.FAIL + "Nombre de mot de passe trop grand !(MAX 50)" + bcolors.ENDC)72                                else:73                                    print("Nombre de mot de passe trop grand !(MAX 50)")74                                erreur_syntax = True75                            elif length >= 30:76                                if is_linux:77                                    print(bcolors.FAIL + "Longueur du mot de passe trop grand !(MAX 50)" + bcolors.ENDC)78                                else:79                                    print("Longueur du mot de passe trop grand !(MAX 50)")80                                erreur_syntax = True81                            else:82                                erreur_syntax = False83                                passData = [args[0], password_counter, length]84                                passData = pickle.dumps(passData)85                                server.send(passData)86                        except :87                            if is_linux:88                                print(bcolors.WARNING + "Erreur dans la syntax. Utilisation :")89                                print("\t Gen : Génerer des mot de passe. Utilisation : Gen <Nombre de mot de passe> <Longueur du mot de passe>")90                            else:91                                print("Erreur dans la syntax. Utilisation :")92                                print(93                                    "\t Gen : Génerer des mot de passe. Utilisation : Gen <Nombre de mot de passe> <Longueur du mot de passe>")94                            erreur_syntax = True95                            pass96                    if not erreur_syntax :97                        try:98                            result = server.recv(99999999)99                            password_holder = pickle.loads(result)100                        except KeyboardInterrupt:101                            continue102                        validchoice = False103                        while not validchoice:104                            print("Voulez-vous affichés les mots de passe ? (O/N)")105                            choice = commandInput()106                            if choice == "O" or choice == "o":107                                validchoice = True108                                for index, decoded_password in enumerate(password_holder):109                                    if is_linux:110                                        print("MDP {} : ".format(index + 1), bcolors.OKBLUE + decoded_password + bcolors.ENDC)111                                    else:112                                        print("MDP {} : ".format(index + 1) + decoded_password)113                            elif choice == "N" or choice == "n":114                                validchoice = True115                            else:116                                validchoice = False117                        validchoice = False118                        while not validchoice:119                            print("Voulez vous sauvegarder les mots de passe dans un fichier ?(O/N)")120                            save = commandInput()121                            if save == "O" or save == "o":122                                print("Sauvegarde en cours...")123                                encoded_password_holder = passwordEncoder(password_holder)124                                with open('Passwords', 'wb') as file:125                                    pickle.dump(encoded_password_holder, file)126                                if is_linux:127                                    print("Le fichier " + bcolors.WARNING + "\"Passwords\"" + bcolors.ENDC + " contient vos mot de passe")128                                else:129                                    print("Le fichier \"Passwords\" contient vos mot de passe")130                                validchoice = True131                            elif save == "N" or save == "n":132                                validchoice = True133                            else:134                                validchoice = False135                else:136                    if is_linux:137                        print(bcolors.FAIL + "Vous n'etes connecté à aucun serveur !" + bcolors.ENDC)138                    else:139                        print("Vous n'etes connecté à aucun serveur !")140            elif args[0] == "Info" or args[0] == "info" or args[0] == "i" or args[0] == "stat" or args[0] == "s" or args[0] == "I" or args[0] == "S":141                try:142                    if len(args) == 1:143                        crash = connexionStatus(server_status)144                        if crash == True:145                            if is_linux:146                                print(bcolors.FAIL + "Déconnecté !" + bcolors.ENDC)147                            else:148                                print("Déconnecté !")149                        elif crash == False:150                            if is_linux:151                                print(bcolors.OKGREEN + "Connecté au serveur : " + bcolors.ENDC+ bcolors.OKBLUE + random_IP + bcolors.ENDC)152                            else:153                                print("Connecté au serveur : {}".format(random_IP))154                    if len(args) == 2:155                        crash = connexionStatus(server_status)156                        if crash == False:157                            if args[1].rstrip() == "-a":158                                if is_linux:159                                    print(bcolors.OKGREEN + "Connecté au serveur : " + bcolors.ENDC + bcolors.OKBLUE + random_IP + bcolors.ENDC)160                                else:161                                    print("Connecté au serveur : {}".format(random_IP))162                                server.send(pickle.dumps([args[0]]))163                                sys_info = server.recv(99999999)164                                sys_info = pickle.loads(sys_info)165                                if is_linux:166                                    print(bcolors.WARNING + "Platform : {}".format(sys_info[0]) + bcolors.ENDC)167                                    print(bcolors.WARNING + "Platform-Release : {}".format(sys_info[1]) + bcolors.ENDC)168                                    print(bcolors.WARNING + "Platform-Version : {}".format(sys_info[2]) + bcolors.ENDC)169                                    print(bcolors.FAIL + "Architecture : {}".format(sys_info[3]) + bcolors.ENDC)170                                    print(bcolors.OKBLUE + "Hostname : {}".format(sys_info[4]) + bcolors.ENDC)171                                    print(bcolors.OKBLUE + "Ip Address : {}".format(sys_info[5]) + bcolors.ENDC)172                                    print(bcolors.FAIL + "Processor : {}".format(sys_info[6]) + bcolors.ENDC)173                                    print(bcolors.FAIL + "CPU Cores : {}".format(sys_info[7]) + bcolors.ENDC)174                                    print(bcolors.OKGREEN + "Ram : {}".format(sys_info[8]) + bcolors.ENDC)175                                    print(bcolors.OKGREEN + "Ram Usage : {}%".format(sys_info[9]) + bcolors.ENDC)176                                    print(bcolors.FAIL + "CPU Usage : {}%".format(sys_info[10]) + bcolors.ENDC)177                                    start_ms = time.time()178                                    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)179                                    s.connect((random_IP, PORT))180                                    s.close()181                                    print(bcolors.OKBLUE + "Latence : {} ms".format(round(time.time() - start_ms, 2)) + bcolors.ENDC)182                                else:183                                    print("Platform : {}".format(sys_info[0]))184                                    print("Platform-Release : {}".format(sys_info[1]))185                                    print("Platform-Version : {}".format(sys_info[2]))186                                    print("Architecture : {}".format(sys_info[3]))187                                    print("Hostname : {}".format(sys_info[4]))188                                    print("Ip Address : {}".format(sys_info[5]))189                                    print("Processor : {}".format(sys_info[6]))190                                    print("CPU Cores : {}".format(sys_info[7]))191                                    print("Ram : {}".format(sys_info[8]))192                                    print("Ram Usage : {}%".format(sys_info[9]))193                                    print("CPU Usage : {}%".format(sys_info[10]))194                                    start_ms = time.time()195                                    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)196                                    s.connect((random_IP, PORT))197                                    s.close()198                                    print("Latence : {} ms".format(round(time.time() - start_ms, 2)))199                        elif crash == True:200                            if is_linux:201                                print(bcolors.FAIL + "Déconnecté !" + bcolors.ENDC)202                            else:203                                print("Déconnecté !")204                except KeyboardInterrupt:205                    pass206            elif args[0] == "Serveur" or args[0] == "SERVEUR" or args[0] == "serveur" or args[0] == "serv":207                serveur_up = []208                if len(args) == 1:209                    for serveur in IP:210                        try:211                            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)212                            s.settimeout(1)213                            s.connect((serveur, PORT))214                            s.close()215                            serveur_up.append(serveur)216                        except:217                            pass218                    for index, serveur in enumerate(serveur_up):219                        if is_linux:220                            print(bcolors.OKGREEN + "Serveur {} [{}] : est fonctionnel !".format(index+1, serveur) + bcolors.ENDC)221                        else:222                            print("Serveur {} [{}] : est fonctionnel !".format(index + 1,serveur))223                    #faire une boucle et afficher les serveur disponibles224                elif args[1] == "-a":225                    serveur_up = []226                    serveur_down = []227                    for serveur in IP:228                        try:229                            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)230                            s.settimeout(1)231                            s.connect((serveur, PORT))232                            s.close()233                            serveur_up.append(serveur)234                        except:235                            serveur_down.append(serveur)236                            pass237                    for index, serveur in enumerate(serveur_up):238                        if is_linux:239                            print(bcolors.OKGREEN + "Serveur {} [{}] : est fonctionnel !".format(index+1, serveur) + bcolors.ENDC)240                        else:241                            print("Serveur {} [{}] : est fonctionnel !".format(index + 1,serveur))242                    for index, serveur in enumerate(serveur_down):243                        if is_linux:244                            print(bcolors.FAIL + "Serveur {} [{}] : est down !".format(index + 1, serveur) + bcolors.ENDC)245                        else:246                            print("Serveur {} [{}] : est down !".format(index + 1, serveur))247                    #Faire une boucle et afficher serveurs disponible et non disponibles248            elif args[0] == "exit" or args[0] == "Exit" or args[0] == "ex" or args[0] == "quit" or args[0] == "QUIT" or args[0] == "Quit":249                os._exit(0)250            elif args[0] == "Help" or args[0] == "help" or args[0] == "HELP" or args[0] == "aide" or args[0] == "Aide" or args[0] == "AIDE":251                if is_linux:252                    print(bcolors.WARNING)253                    print("Liste de commande disponible : ")254                    print("\t Gen : Génerer des mot de passe. Utilisation : Gen <Nombre de mot de passe> <Longueur du mot de passe>.")255                    print("\t Info ou stat : Afficher le status de connexion. Utilisation : -a Pour plus d'inforamtions sur le serveur.")256                    print("\t Serveur : Afficher les serveurs disponibles. Utilisation : -a pour afficher tout les serveurs.")257                    print("\t Changer : Changer de serveur.")258                    print("\t Remote : Controler un serveur à distance par ssh.")259                    print("\t Exit : Quitter le programme.")260                    print("\t Help ou Aide : afficher ce menu.")261                    print("Plus de fonctionne viendront quand les idées seront la.")262                    print(bcolors.ENDC)263                else:264                    print("Liste de commande disponible : ")265                    print("\t Gen : Génerer des mot de passe. Utilisation : Gen <Nombre de mot de passe> <Longueur du mot de passe>.")266                    print("\t Info ou stat : Afficher le status de connexion. Utilisation : -a Pour plus d'inforamtions sur le serveur.")267                    print("\t Serveur : Afficher les serveurs disponibles. Utilisation : -a pour afficher tout les serveurs.")268                    print("\t Changer : Changer de serveur.")269                    print("\t Remote : Controler un serveur à distance par ssh.")270                    print("\t Exit : Quitter le programme.")271                    print("\t Help ou Aide : afficher ce menu.")272                    print("Plus de fonctionne viendront quand les idées seront la.")273            elif args[0] == "changer" or args[0] == "Changer" or args[0] == "CHANGER":274                validchoice = False275                while not validchoice:276                    serveur_up = []277                    crashed = connexionStatus(server_status)278                    if not crashed:279                        if is_linux:280                            print(bcolors.WARNING + "Vous etes connecté au serveur " + bcolors.OKGREEN + random_IP + bcolors.ENDC)281                        else:282                            print("Vous etes connecté au serveur {}".format(random_IP))283                    for serveur in IP:284                        try:285                            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)286                            s.connect((serveur, PORT))287                            s.close()288                            serveur_up.append(serveur)289                        except:290                            pass291                    for index, serveur in enumerate(serveur_up):292                        try:293                            start_ms = time.time()294                            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)295                            s.connect((serveur, PORT))296                            s.close()297                            if is_linux:298                                print(bcolors.OKCYAN + "Serveur ({}) [{}] : {} ms".format(index+1, serveur, round(time.time() - start_ms, 3)) + bcolors.ENDC)299                            else:300                                print("Serveur ({}) [{}] : {} ms".format(index + 1, serveur,round(time.time() - start_ms,3)))301                        except:302                            pass303                    if len(serveur_up) == 0:304                        if is_linux:305                            print(bcolors.FAIL + "Aucun serveur disponible !" + bcolors.ENDC)306                        else:307                            print("Aucun serveur disponible !")308                        validchoice = True309                    else:310                        try:311                            serveur_choix = input("Choisissez un serveur dans la liste ci-dessus (Enter pour ne rien faire) : ")312                            if not serveur_choix:313                                validchoice = True314                            else:315                                serveur_choix = int(serveur_choix)316                                if serveur_up[serveur_choix -1] == random_IP:317                                    if is_linux:318                                        print(bcolors.WARNING + "Vous êtes deja connecté à ce serveur !. Serveur non changé !" + bcolors.ENDC)319                                    else:320                                        print("Vous êtes deja connecté à ce serveur !. Serveur non changé !")321                                    validchoice = True322                                else:323                                    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)324                                    server.connect((IP[serveur_choix-1],PORT))325                                    random_IP = IP[serveur_choix - 1]326                                    if is_linux:327                                        print(bcolors.WARNING + "Connecté au serveur {}".format(IP[serveur_choix-1])+ bcolors.ENDC)328                                    else:329                                        print("Connecté au serveur {}".format(IP[serveur_choix - 1]))330                                    validchoice = True331                        except ValueError:332                            validchoice = False333                            pass334                        except IndexError:335                            if is_linux:336                                print( bcolors.WARNING + "Serveur non compris dans la liste..." + bcolors.ENDC)337                            else:338                                print("Serveur non compris dans la liste...")339                            validchoice = False340                        except ConnectionError:341                            if is_linux:342                                print(bcolors.FAIL + "Serveur Down" + bcolors.ENDC)343                            else:344                                print("Serveur Down")345                            validchoice = False346            elif args[0] == "remote" or args[0] == "Remote":347                attempt = 3348                while attempt != 0:349                    try:350                        if is_linux:351                            username = input(bcolors.OKBLUE + "Utilisateur >> {}".format(bcolors.ENDC))352                        else:353                            username = input("Utilisateur >> ")354                        if is_linux:355                            pasword = getpass.getpass(bcolors.WARNING + "Mot de passe >> {}".format(bcolors.ENDC))356                        else:357                            pasword = getpass.getpass("Mot de passe >> ")358                        ssh = paramiko.SSHClient()359                        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())360                        ssh.connect(random_IP, username=username, password=pasword)361                        attempt = 0362                        crashed = connexionStatus(server_status)363                        current_server = random_IP364                        while not crashed and current_server == random_IP:365                            crashed = connexionStatus(server_status)366                            command_ssh = input("{} >> ".format(random_IP))367                            if command_ssh == "Exit" or command_ssh == "exit":368                                break369                            stdin, stdout, stderr = ssh.exec_command(command_ssh)370                            print(stdout.read().decode())371                            err = stderr.read().decode()372                            if err:373                                print(err)374                    except paramiko.AuthenticationException:375                        if is_linux:376                            print(bcolors.FAIL + "Nom d'utilisateur ou mot de passe incorrect !" + bcolors.ENDC)377                        else:378                            print("Nom d'utilisateur ou mot de passe incorrect !")379                        attempt -=1380                    except paramiko.ssh_exception.NoValidConnectionsError:381                        print("Impossible de se connecter au serveur {}. Veuillez check le firewall !".format(random_IP))382                        attempt = 0383                    except getpass.GetPassWarning:384                        pass385            elif args[0] == "Hash" or args[0] == "hash":386                crashed = connexionStatus(server_status)387                erreur_syntax = False388                if crashed == False:389                    if len(args) == 1:390                        if is_linux:391                            print(bcolors.WARNING + "Utilisation : Hash <HashType> <hash>" + bcolors.ENDC)392                        else:393                            print("Utilisation : Hash <HashType> <hash>")394                        erreur_syntax = True395                    elif len(args) == 2:396                        erreur_syntax = False397                        hash = [args[0],"Default" ,args[1]]398                        hash = pickle.dumps(hash)399                        server.send(hash)400                    elif len(args) == 3:401                        erreur_syntax = False402                        hash = [args[0], args[1], args[2]]403                        hash = pickle.dumps(hash)404                        server.send(hash)405                    else:406                        if is_linux:407                            print(bcolors.WARNING + "Utilisation : Hash <HashType> <hash>" + bcolors.ENDC)408                        else:409                            print("Utilisation : Hash <HashType> <hash>")410                        erreur_syntax = True411                    if not erreur_syntax:412                        try:413                            hash = server.recv(99999999)414                            hash = pickle.loads(hash)415                            print(hash)416                        except KeyboardInterrupt:417                            pass418                else:419                    if is_linux:420                        print(bcolors.FAIL + "Vous n'etes connecté à aucun serveur !" + bcolors.ENDC)421                    else:422                        print("Vous n'etes connecté à aucun serveur !")423    except AttributeError:424        os._exit(0)425    except KeyboardInterrupt:426        os._exit(0)427def connexionStatus(server_status):428    try:429        server_status.recv(4)430    except socket.error as e:431        err = e.args[0]432        if err == errno.EAGAIN or err == errno.EWOULDBLOCK:433            return False434        else:435            return True436    else:437        return True438def connexionHolder():439    connected = False440    while not connected:441        try:442            global server, server_status443            server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)444            server_status = socket.socket(socket.AF_INET, socket.SOCK_STREAM)445            serveur_up = []446            while len(serveur_up) == 0:447                for serveur in IP:448                    try:449                        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)450                        s.connect((serveur, PORT))451                        s.close()452                        serveur_up.append(serveur)453                    except:454                        pass455            global random_IP456            random_IP = serveur_up[random.randint(0,len(serveur_up)-1)]457            server.connect((random_IP, PORT))458            server_status.connect((random_IP, PORT))459            server_status.setblocking(False)460            connected = True461            crashed = False462            while not crashed:463                crashed = connexionStatus(server_status)464                if crashed == True:465                    crashed = True466                    connected = False467                time.sleep(0.1)468        except ConnectionError:469            print("La connexion à échouer.. Connexion à un autre serveur en cours...")470            connected = False471            time.sleep(1)472        except socket.timeout:473            print("Le serveur est down ou alors le port n'est pas forwarded...")474is_linux = check_linux()475connexion_thread = threading.Thread(target=connexionHolder)476connexion_thread.start()...remote_run_aws.py
Source:remote_run_aws.py  
...38        instances_bitness = RemoteRunAwsService.get_bitness(instances)39        return CmdRunner.run_multiple_commands(40            instances,41            lambda instance: RemoteRunAwsService.run_aws_monkey_cmd_async(42                instance['instance_id'], RemoteRunAwsService._is_linux(instance['os']), island_ip,43                instances_bitness[instance['instance_id']]),44            lambda _, result: result.is_success)45    @staticmethod46    def is_running_on_aws():47        return RemoteRunAwsService.aws_instance.is_instance()48    @staticmethod49    def update_aws_region_authless():50        """51        Updates the AWS region without auth params (via IAM role)52        """53        AwsService.set_region(RemoteRunAwsService.aws_instance.region)54    @staticmethod55    def get_bitness(instances):56        """57        For all given instances, checks whether they're 32 or 64 bit.58        :param instances: List of instances to check59        :return: Dictionary with instance ids as keys, and True/False as values. True if 64bit, False otherwise60        """61        return CmdRunner.run_multiple_commands(62            instances,63            lambda instance: RemoteRunAwsService.run_aws_bitness_cmd_async(64                instance['instance_id'], RemoteRunAwsService._is_linux(instance['os'])),65            lambda instance, result: RemoteRunAwsService._get_bitness_by_result(66                RemoteRunAwsService._is_linux(instance['os']), result))67    @staticmethod68    def _get_bitness_by_result(is_linux, result):69        if not result.is_success:70            return None71        elif is_linux:72            return result.stdout.find('i686') == -1  # i686 means 32bit73        else:74            return result.stdout.lower().find('programfiles(x86)') != -1  # if not found it means 32bit75    @staticmethod76    def run_aws_bitness_cmd_async(instance_id, is_linux):77        """78        Runs an AWS command to check bitness79        :param instance_id: Instance ID of target80        :param is_linux:    Whether target is linux81        :return:            Cmd82        """83        cmd_text = 'uname -m' if is_linux else 'Get-ChildItem Env:'84        return RemoteRunAwsService.run_aws_cmd_async(instance_id, is_linux, cmd_text)85    @staticmethod86    def run_aws_monkey_cmd_async(instance_id, is_linux, island_ip, is_64bit):87        """88        Runs a monkey remotely using AWS89        :param instance_id: Instance ID of target90        :param is_linux:    Whether target is linux91        :param island_ip:   IP of the island which the instance will try to connect to92        :param is_64bit:    Whether the instance is 64bit93        :return:            Cmd94        """95        cmd_text = RemoteRunAwsService._get_run_monkey_cmd_line(is_linux, is_64bit, island_ip)96        return RemoteRunAwsService.run_aws_cmd_async(instance_id, is_linux, cmd_text)97    @staticmethod98    def run_aws_cmd_async(instance_id, is_linux, cmd_line):99        cmd_runner = AwsCmdRunner(is_linux, instance_id)100        return Cmd(cmd_runner, cmd_runner.run_command_async(cmd_line))101    @staticmethod102    def _is_linux(os):103        return 'linux' == os104    @staticmethod105    def _get_run_monkey_cmd_linux_line(bit_text, island_ip):106        return r'wget --no-check-certificate https://' + island_ip + r':5000/api/monkey/download/monkey-linux-' + \107               bit_text + r'; chmod +x monkey-linux-' + bit_text + r'; ./monkey-linux-' + bit_text + r' m0nk3y -s ' + \108               island_ip + r':5000'109    @staticmethod110    def _get_run_monkey_cmd_windows_line(bit_text, island_ip):111        return r"[System.Net.ServicePointManager]::ServerCertificateValidationCallback = {" \112               r"$true}; (New-Object System.Net.WebClient).DownloadFile('https://" + island_ip + \113               r":5000/api/monkey/download/monkey-windows-" + bit_text + r".exe','.\\monkey.exe'); " \114                                                                         r";Start-Process -FilePath '.\\monkey.exe' " \115                                                                         r"-ArgumentList 'm0nk3y -s " + island_ip + r":5000'; "116    @staticmethod...configure.py
Source:configure.py  
1#!/usr/bin/env python2import confu3parser = confu.standard_parser("cpuinfo configuration script")4parser.add_argument("--log", dest="log_level",5    choices=("none", "fatal", "error", "warning", "info", "debug"), default="error")6parser.add_argument("--mock", dest="mock", action="store_true")7def main(args):8    options = parser.parse_args(args)9    build = confu.Build.from_options(options)10    macros = {11        "CPUINFO_LOG_LEVEL": {"none": 0, "fatal": 1, "error": 2, "warning": 3, "info": 4, "debug": 5}[options.log_level],12        "CLOG_LOG_TO_STDIO": int(not options.mock),13        "CPUINFO_MOCK": int(options.mock),14    }15    if build.target.is_linux or build.target.is_android:16        macros["_GNU_SOURCE"] = 117    build.export_cpath("include", ["cpuinfo.h"])18    with build.options(source_dir="src", macros=macros, extra_include_dirs="src", deps=build.deps.clog):19        sources = ["api.c", "init.c", "cache.c"]20        if build.target.is_x86 or build.target.is_x86_64:21            sources += [22                "x86/init.c", "x86/info.c", "x86/isa.c", "x86/vendor.c",23                "x86/uarch.c", "x86/name.c", "x86/topology.c",24                "x86/cache/init.c", "x86/cache/descriptor.c", "x86/cache/deterministic.c",25            ]26            if build.target.is_macos:27                sources += ["x86/mach/init.c"]28            elif build.target.is_linux or build.target.is_android:29                sources += [30                    "x86/linux/init.c",31                    "x86/linux/cpuinfo.c",32                ]33        if build.target.is_arm or build.target.is_arm64:34            sources += ["arm/uarch.c", "arm/cache.c"]35            if build.target.is_linux or build.target.is_android:36                sources += [37                    "arm/linux/init.c",38                    "arm/linux/cpuinfo.c",39                    "arm/linux/clusters.c",40                    "arm/linux/midr.c",41                    "arm/linux/chipset.c",42                    "arm/linux/hwcap.c",43                ]44                if build.target.is_arm:45                    sources.append("arm/linux/aarch32-isa.c")46                elif build.target.is_arm64:47                    sources.append("arm/linux/aarch64-isa.c")48                if build.target.is_android:49                    sources += [50                        "arm/android/properties.c",51                    ]52        if build.target.is_macos:53            sources += ["mach/topology.c"]54        if build.target.is_linux or build.target.is_android:55            sources += [56                "linux/cpulist.c",57                "linux/smallfile.c",58                "linux/multiline.c",59                "linux/processors.c",60            ]61            if options.mock:62                sources += ["linux/mockfile.c"]63        build.static_library("cpuinfo", map(build.cc, sources))64    with build.options(source_dir="tools", deps=[build, build.deps.clog]):65        build.executable("cpu-info", build.cc("cpu-info.c"))66        build.executable("isa-info", build.cc("isa-info.c"))67        build.executable("cache-info", build.cc("cache-info.c"))68    if build.target.is_x86_64:69        with build.options(source_dir="tools", include_dirs=["src", "include"]):70            build.executable("cpuid-dump", build.cc("cpuid-dump.c"))71    with build.options(source_dir="test", deps=[build, build.deps.clog, build.deps.googletest]):72        build.smoketest("init-test", build.cxx("init.cc"))73        if build.target.is_linux:74            build.smoketest("get-current-test", build.cxx("get-current.cc"))75        if build.target.is_x86_64:76            build.smoketest("brand-string-test", build.cxx("name/brand-string.cc"))77    if options.mock:78        with build.options(source_dir="test", include_dirs="test", macros="CPUINFO_MOCK", deps=[build, build.deps.googletest]):79            if build.target.is_arm64 and build.target.is_linux:80                build.unittest("scaleway-test", build.cxx("scaleway.cc"))81    if not options.mock:82        with build.options(source_dir="bench", deps=[build, build.deps.clog, build.deps.googlebenchmark]):83            build.benchmark("init-bench", build.cxx("init.cc"))84            if not build.target.is_macos:85                build.benchmark("get-current-bench", build.cxx("get-current.cc"))86    return build87if __name__ == "__main__":88    import sys...Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
