Best Python code snippet using autotest_python
ssh_executor.py
Source:ssh_executor.py
...14 self.ftp_client = None15 self.logger = logging.getLogger('logger')16 self.remote_host = config['submission_host']17 self.credentials = config['ssh_credentials']18 def setup_ssh(self):19 """20 Initiate SSH connection and save it as self.ssh_client21 """22 self.logger.info('Will set up ssh')23 if self.ssh_client:24 self.close_connections()25 if ':' not in self.credentials:26 with open(self.credentials) as json_file:27 credentials = json.load(json_file)28 else:29 credentials = {}30 credentials['username'] = self.credentials.split(':')[0]31 credentials['password'] = self.credentials.split(':')[1]32 self.logger.info('Credentials loaded successfully: %s', credentials['username'])33 self.ssh_client = paramiko.SSHClient()34 self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())35 self.ssh_client.connect(self.remote_host,36 username=credentials['username'],37 password=credentials['password'],38 timeout=30)39 self.logger.info('Done setting up ssh')40 def setup_ftp(self):41 """42 Initiate SFTP connection and save it as self.ftp_client43 If needed, SSH connection will be automatically set up44 """45 self.logger.info('Will set up ftp')46 if self.ftp_client:47 self.close_connections()48 if not self.ssh_client:49 self.setup_ssh()50 self.ftp_client = self.ssh_client.open_sftp()51 self.logger.info('Done setting up ftp')52 def execute_command(self, command):53 """54 Execute command over SSH55 """56 if not self.ssh_client:57 self.setup_ssh()58 if isinstance(command, list):59 command = '; '.join(command)60 self.logger.info('Executing %s', command)61 (_, stdout, stderr) = self.ssh_client.exec_command(command)62 self.logger.info('Executed %s. Reading response', command)63 # Close channel after minute of waiting for EOF64 # This timeouts and closes channel if nothing was received65 stdout_timeout = time.time() + 6066 while not stdout.channel.eof_received:67 time.sleep(1)68 if time.time() > stdout_timeout:69 stdout.channel.close()70 break71 stdout = stdout.read().decode('utf-8').strip()...
test_ssh.py
Source:test_ssh.py
...33 'port': 4900034 },35 scope='module'36)37def test_setup_ssh(setup_ssh):38 # Rudimentary smoke test for setup_ssh so we have39 # multiple uses for the setup_ssh40 assert 'port' in setup_ssh['custom']41 assert setup_ssh['custom']['host'] == 'localhost'42def test_ssh_class(setup_ssh, resource_test_dir, resman):43 with swallow_logs(new_level=logging.DEBUG) as log:44 # Test connecting to test SSH server.45 # TODO: Add a test using a SSH key pair.46 config = dict(47 name='ssh-test-resource',48 type='ssh',49 **setup_ssh['custom']50 )51 resource = resman.factory(config)...
test_flows_setup_ssh.py
Source:test_flows_setup_ssh.py
1import __builtin__2import maps3import mock4from tendrl.commons.flows.setup_ssh import SetupSsh5'''Unit Test Cases'''6def test_constructor():7 setup_ssh = SetupSsh()8 assert setup_ssh._defs is not None9@mock.patch('tendrl.commons.event.Event.__init__',10 mock.Mock(return_value=None))11@mock.patch('tendrl.commons.message.Message.__init__',12 mock.Mock(return_value=None))13def test_run():14 param = maps.NamedDict(ssh_setup_script='Test ssh_setup_script')15 setattr(__builtin__, "NS", maps.NamedDict())16 NS.publisher_id = "node_agent"17 setup_ssh = SetupSsh(parameters=param)...
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!!