KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use is_linux method in localstack
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...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
