KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use ssh_command method in autotest
Best Python code snippet using autotest_python
GUIServerControlv2.py
Source:GUIServerControlv2.py 
1from tkinter import *2import paramiko as pk3import subprocess4import time5import logging6from itertools import count7import sys8from tkinter import filedialog9from tkinter import Toplevel10from fabric import Connection as connection11from fabric import config as conf12from fabric import transfer as trans13class GUIServerControl:14 def __init__(self, master, donemessage=None):15 print("APPLICATION STARTED")16 LOG_FILENAME = "gflapp.log"17 logging.basicConfig(filename=LOG_FILENAME, level=logging.INFO)18 pk.util.log_to_file("paramiko.log", level="INFO")19 self.master = master20 #root.configure(background='grey')21 master.title("GFL Docker Management")22 23 # window = Toplevel()24 # window.wm_title("New Window")25 #master.geometry('1000x1000')26 self.entryversion = StringVar()27 self.sshaddressn1 = StringVar()28 self.sshaddressn2 = StringVar()29 self.sshaddressn3 = StringVar()30 self.sshuser = StringVar()31 self.sshpass = StringVar()32 self.commandoutput = StringVar()33 self.tranfilename = StringVar()34 self.dockeruser = StringVar()35 self.dockerpass = StringVar()36 self.entrydirname = StringVar()37 labelversion = Label(master, text="Version : ").grid(row=1, column=0)38 labelsshaddressn1 = Label(master, text="Node 1 Address : ").grid(row=2, column=0)39 labelsshaddressn2 = Label(master, text="Node 2 Address : ").grid(row=3, column=0)40 labelsshaddressn3 = Label(master, text="Node 3 Address : ").grid(row=4, column=0)41 labelsshuser = Label(master, text="SSH User : ").grid(row=5, column=0)42 labelsshpass = Label(master, text="SSH Password : ").grid(row=6, column=0)43 self.entryversionbox = Entry(root, width=25, textvariable=self.entryversion)44 self.entryversionbox.grid(row=1, column=1)45 self.entrysshaddressboxn1 = Entry(root, width=25, textvariable=self.sshaddressn1)46 self.entrysshaddressboxn1.grid(row=2, column=1)47 self.entrysshaddressboxn2 = Entry(root, width=25, textvariable=self.sshaddressn2)48 self.entrysshaddressboxn2.grid(row=3, column=1)49 self.entrysshaddressboxn3 = Entry(root, width=25, textvariable=self.sshaddressn3)50 self.entrysshaddressboxn3.grid(row=4, column=1)51 self.entrysshuserbox = Entry(root, width=25, textvariable=self.sshuser)52 self.entrysshuserbox.grid(row=5, column=1)53 self.entrysshpassbox = Entry(root, width=25, textvariable=self.sshpass, show="*")54 self.entrysshpassbox.grid(row=6, column=1)55 # GFL CONTROLS #56 self.label = Label (master, text="Useful Docker Commands")57 self.label.grid(row=1, column=2)58 self.gflnet_button = Button(master, text="Create gflnet", command= lambda: self.ssh_connect(self.sshuser.get(), self.sshpass.get(), self.sshaddressn1.get(), self.gflnet()))59 self.gflnet_button.grid(row=2, column=2)60 self.gflnetls_button = Button(master, text="List GFL Networks", command= lambda: self.ssh_connect(self.sshuser.get(), self.sshpass.get(), self.sshaddressn1.get(), self.gflnetls()))61 self.gflnetls_button.grid(row=3, column=2)62 self.changeversion = Button(master, text="Change Version", command= lambda: self.ssh_connect(self.sshuser.get(), self.sshpass.get(), self.sshaddressn1.get(), self.gflchangeversion()))63 self.changeversion.grid(row=4, column=2)64 self.dockerimages = Button (master, text="View Docker Images", command= lambda: self.ssh_connect(self.sshuser.get(), self.sshpass.get(), self.sshaddressn1.get(), self.dockerimages))65 self.dockerimages.grid(row=5, column=2)66 self.pullimages = Button (master, text="Pull Docker Images", command= lambda: self.ssh_connect(self.sshuser.get(), self.sshpass.get(), self.sshaddressn1.get(), self.pullimages))67 self.pullimages.grid(row=6, column=2)68 self.startgfl_button = Button (master, text="Start GFL", command=lambda: self.ssh_connect(self.sshuser.get(), self.sshpass.get(), self.sshaddressn1.get(), self.startgfl()))69 self.startgfl_button.grid(row=7, column=2)70 self.stopgfl_button = Button (master, text="Stop GFL", command=lambda: self.ssh_connect(self.sshuser.get(), self.sshpass.get(), self.sshaddressn1.get(), self.stopgfl()))71 self.stopgfl_button.grid(row=8, column=2)72 self.volumeprune = Button (master, text="Prune Docker Volumes", command=self.volumeprune).grid(row=9, column=2)73 self.yscrollbar = Scrollbar(master)74 self.yscrollbar.grid(row=12, column=3, sticky=N + S)75 self.output_box = Text(master, textvariable=self.newmesssage(donemessage), width=80, yscrollcommand=self.yscrollbar.set)76 self.output_box.grid(row=12, column=0, columnspan=5, sticky=W+E, padx=10)77 self.output_box.see("end")78 self.message_bar = Message(master, text=donemessage, width=1000)79 self.message_bar.grid(row=13, column=0, columnspan=5, sticky=W+E)80 self.close_button = Button(master, text="Close", command=master.quit)81 self.close_button.grid(row=14, column=0, columnspan=5)82 # Server Controls83 self.server_label = Label (master, text="Server Control/Setup")84 self.server_label.grid(row=1, column=4)85 self.OSupdate_button = Button(master, text="OS Update/Upgrade", command=lambda: self.sudo_ssh_connect(self.sshuser.get(), self.sshpass.get(), self.sshaddressn1.get(), self.OSUpdate()))86 self.OSupdate_button.grid(row=2, column=4)87 self.installdocker_button = Button(master, text="Install Docker/Compose", command=lambda: self.sudo_ssh_connect(self.sshuser.get(), self.sshpass.get(), self.sshaddressn1.get(), self.install_docker()))88 self.installdocker_button.grid(row=3, column=4)89 self.dockerlogin_button = Button(master, text="Docker Login", command=lambda: self.docker_login())90 self.dockerlogin_button.grid(row=6, column=4)91 # File Upload92 self.upload_label = Label(master, text="File Uploads To Server: ")93 self.upload_label.grid(row=8, column=0)94 self.fileselect_button = Button(master, text="Upload File", command= lambda: self.selectfile(self.sshaddressn1.get(), self.sshuser.get(), self.sshpass.get()))95 self.fileselect_button.grid(row=9, column=0)96 self.entryfilename = Message(root, width=25, textvariable=self.tranfilename)97 self.entryfilename.grid(row=9, column=1)98 self.logdownload_button = Button(master, text="Log Download", command= lambda: self.log_download(self.sshaddressn1.get(), self.sshuser.get(), self.sshpass.get()))99 self.logdownload_button.grid(row=7, column=1)100 # self.newtext = Text(window, width=80)101 # self.newtext.grid(row=0, column=0)102 def ssh_connect(self, ssh_user, ssh_password, ssh_host, ssh_command):103 self.connect_accept()104 if len(self.sshaddressn1.get()) < 1:105 self.output_box.insert(END, "NO ADDRESSS ENTERED")106 self.message_bar.configure(text="NO ADDRESS ENTERED!!!")107 else:108 output = ""109 self.output_box.insert(END, "***** COMMAND STARTED ******\n")110 c = connection(host=ssh_host,user=ssh_user, port=22, connect_kwargs={'password': ssh_password})111 for cmdlist in ssh_command:112 self.output_box.insert(END, "RUNNING COMMAND: " + cmdlist)113 output = c.run(cmdlist, pty=True)114 output_stdout = output.stdout115 for line in output_stdout:116 sys.stdout.flush()117 self.output_box.insert(END, line)118 self.message_bar.configure(text="Done Command: \n" + cmdlist)119 print(output_stdout)120 logging.info(output)121 def sudo_ssh_connect(self, ssh_user, ssh_password, ssh_host, ssh_command):122 self.connect_accept()123 if len(self.sshaddressn1.get()) < 1:124 self.output_box.insert(END, "NO ADDRESSS ENTERED")125 self.message_bar.configure(text="NO ADDRESS ENTERED!!!")126 else:127 self.output_box.insert(END, "***** COMMAND STARTED ******\n")128 output = ""129 config = conf.Config(overrides={'sudo': {'password': ssh_password}})130 c = connection(host=ssh_host, user=ssh_user, port=22, config=config, connect_kwargs={'password': ssh_password})131 for cmdlist in ssh_command:132 self.output_box.insert(END, "RUNNING COMMAND: " + cmdlist)133 output = c.sudo(cmdlist, hide='stderr', pty=True)134 output_stdout = output.stdout135 for line in output_stdout:136 sys.stdout.flush()137 self.output_box.insert(END, line)138 self.message_bar.configure(text="Done Command: \n" + cmdlist)139 print(output_stdout)140 logging.info(output)141 def walltest(self):142 ssh_command = []143 ssh_command.append("wall 'I am Listening'")144 print("The function command has ran")145 return ssh_command146 def gflnet(self): #Might not be needed147 print("Creating gflnet")148 ssh_command = []149 ssh_command.append("docker network rm gflnet; docker network create gflnet")150 #ssh_command.append("docker network create gflnet")151 print("Complete Recreating GFLNet")152 return ssh_command153 def gflnetls(self):154 ssh_command = []155 ssh_command.append("docker network ls")156 return ssh_command157 def gflchangeversion(self):158 print("CHANGEVERSIONNUMBER: " + self.entryversion.get())159 if len(self.entryversion.get()) == 0:160 self.message_bar.configure(text="NO VERSION SELECTED")161 else:162 ssh_command = []163 ssh_command.append("cp /home/gfl/compositions/gfl/gfl.yml /home/gfl/compositions/gfl/" + self.entryversion.get() + ".yml")164 ssh_command.append("sed -i s/GFLVERSION/" + self.entryversion.get() + "/g /home/gfl/compositions/gfl/" + self.entryversion.get() + ".yml")165 print("Created new composition file for: ", (self.entryversion.get()))166 print("Changed Composition version to Version: ", (self.entryversion.get()))167 return ssh_command168 def pullimages(self):169 ssh_command = []170 print("PULIMAGESVERSION: " + self.entryversion.get())171 if len(self.entryversion.get()) == 0:172 self.message_bar.configure(text="NO VERSION SELECTED")173 else:174 ssh_command.append("docker-compose -f /home/gfl/compositions/databases/databases.yml pull")175 ssh_command.append("COMPOSE_HTTP_TIMEOUT=1000 docker-compose -f /home/gfl/compositions/gfl/" + self.entryversion.get() + ".yml pull")176 return ssh_command177 def startgfl(self):178 print("STARTGFLVERSION: " + self.entryversion.get())179 if len(self.entryversion.get()) == 0:180 self.message_bar.configure(text="NO VERSION SELECTED")181 else:182 ssh_command = []183 ssh_command.append('COMPOSE_HTTP_TIMEOUT=1000 docker-compose -f /home/gfl/compositions/databases/databases.yml up -d')184 ssh_command.append('COMPOSE_HTTP_TIMEOUT=1000 docker-compose -f /home/gfl/compositions/gfl/' + self.entryversion.get() + '.yml up -d')185 ssh_command.append('docker ps')186 print(ssh_command)187 self.message_bar.configure(text="Starting GFL Version: \n" + self.entryversion.get())188 return ssh_command189 def stopgfl(self):190 print("STOPGFLVERSION: " + self.entryversion.get())191 if len(self.entryversion.get()) == 0:192 self.entryversionbox.configure(bg="red")193 print("NO VERSION SELECTED")194 self.message_bar.configure(text="NO VERSION SELECTED")195 elif len(self.sshaddressn1.get()) < 1:196 self.connect_warning()197 else:198 ssh_command = []199 ssh_command.append('COMPOSE_HTTP_TIMEOUT=1000 docker-compose -f /home/gfl/compositions/gfl/' + self.entryversion.get() + '.yml down')200 ssh_command.append('COMPOSE_HTTP_TIMEOUT=1000 docker-compose -f /home/gfl/compositions/databases/databases.yml down')201 ssh_command.append('docker ps')202 print(ssh_command)203 return ssh_command204 def dockerimages(self):205 ssh_command = []206 ssh_command.append("docker images")207 return ssh_command208 def volumeprune(self):209 ssh_command = []210 ssh_command.append("docker volume prune -f")211 return ssh_command212 # Here I want to take a 'done' message from a function and return it to the message field at the bottom of the window213 #MIGHT NOT BE NEEDED214 def newmesssage(self, donemessage):215 return donemessage216 def selectfile(self,ssh_host, ssh_user, ssh_password):217 if len(self.sshaddressn1.get()) < 1:218 self.output_box.insert(END, "NO ADDRESSS ENTERED")219 self.message_bar.configure(text="NO ADDRESS ENTERED!!!")220 else:221 try:222 print("Selecting File")223 senddest = "/home/gfl/APPLUPLOAD"224 filename = filedialog.askopenfilename()225 self.upload_label.configure(text=filename)226 config = conf.Config(overrides={'sudo': {'password': ssh_password}})227 c = connection(host=ssh_host, user=ssh_user, port=22, connect_kwargs={'password': ssh_password}).run("mkdir -p /home/gfl/APPLUPLOAD", hide='stderr', pty=True)228 c = connection(host=ssh_host, user=ssh_user, port=22, connect_kwargs={'password': ssh_password}).put(filename, remote=senddest)229 print("Uploaded {} to {}".format(filename, senddest))230 self.output_box.insert(END, "Files Transfered: " + filename + " To: " + senddest)231 return filename232 except PermissionError:233 self.output_box.insert(END, "Permission Error")234 self.message_bar.configure(text="Permission Error")235 def log_download(self, ssh_host, ssh_user, ssh_password):236 if len(self.sshaddressn1.get()) < 1:237 self.output_box.insert(END, "NO ADDRESSS ENTERED")238 self.message_bar.configure(text="NO ADDRESS ENTERED!!!")239 else:240 print("Downloading Logs")241 config = conf.Config(overrides={'sudo': {'password': ssh_password}})242 c = connection(host=ssh_host, user=ssh_user, port=22, connect_kwargs={'password': ssh_password}).run("tar -cvpf ~/GFLserverlogs.tar /home/gfl/data/logs/gfl/", hide='stderr', pty=True)243 c = connection(host=ssh_host, user=ssh_user, port=22, connect_kwargs={'password': ssh_password}).get("/home/" + ssh_user +"/GFLserverlogs.tar", "")244 def OSUpdate(self):245 print("Updating OS")246 ssh_command = []247 ssh_command.append("apt update")248 ssh_command.append("DEBIAN_FRONTEND=noninteractive apt-get -y upgrade")249 ssh_command.append("apt install -y openssh-server zookeeperd curl build-essential luajit")250 print("DONE!!!")251 return ssh_command252 def install_docker(self):253 ssh_command = []254 ssh_command.append("curl -ssL https://get.docker.io | sh")255 ssh_command.append("usermod -aG docker " + self.sshuser.get())256 ssh_command.append("curl -L 'https://github.com/docker/compose/releases/download/1.25.0/docker-compose-$(uname -s)-$(uname -m)' -o /usr/local/bin/docker-compose")257 ssh_command.append("chmod +x /usr/local/bin/docker-compose")258 return ssh_command259 def docker_login(self):260 if len(self.sshaddressn1.get()) < 1:261 self.connect_warning()262 else:263 top=self.top=Toplevel(self.master)264 self.dockeruserbox = Entry(self.top, width=25, textvariable=self.dockeruser)265 self.dockeruserbox.grid(row=1, column=1)266 self.dockerpassbox = Entry(self.top, width=25, textvariable=self.dockerpass, show="*")267 self.dockerpassbox.grid(row=2, column=1)268 self.dockerok_button = Button(self.top, text="Login", command=lambda: self.sudo_ssh_connect(self.sshuser.get(), self.sshpass.get(), self.sshaddressn1.get(), self.docker_login_execute()))269 self.dockerok_button.grid(row=3, column=0)270 self.docker_message = Message(self.top, text="", width=200)271 self.docker_message.grid(row=3, column=1)272 273 def docker_login_execute(self):274 ssh_command = []275 ssh_command.append("docker login tcsjh-dev.co.uk:5000 --username " + self.dockeruser.get() + " --password " + self.dockerpass.get())276 ssh_command.append("docker login tcsjh-dev.co.uk:5002 --username " + self.dockeruser.get() + " --password " + self.dockerpass.get())277 self.docker_message.configure(text="DOCKER LOGIN OK")278 return ssh_command279 def connect_warning(self):280 self.entrysshaddressboxn1.configure(bg="red")281 self.entrysshuserbox.configure(bg="red")282 self.entrysshpassbox.configure(bg="red")283 self.message_bar.configure(text="PLEASE ENTER CONNECTION DETAILS!!!")284 def connect_accept(self):285 boxes = [self.entryversionbox, self.entrysshpassbox, self.entrysshuserbox, self.entrysshaddressboxn1, self.entrysshaddressboxn2, self.entrysshaddressboxn3]286 for line in boxes:287 if len(line.get()) < 1:288 line.configure({"background": "red"})289 else:290 line.configure({"background": "white"})291root = Tk()292my_gui = GUIServerControl(root)...
check_env_remote.py
Source:check_env_remote.py
...14 status, check_result = function(software,value,check_item)15 print(check_result)16 return status, check_result17 @staticmethod18 def deal_ssh_command(ssh_cmd,ssh_bash,flag):19 if flag == 'remote':20 ssh_cmd_ = ssh_bash.split(' ')21 ssh_cmd_.extend(ssh_cmd)22 else:23 ssh_cmd_ = ssh_cmd24 return ssh_cmd_25 def check_os(self):26 check_item = ['openEuler release 20.03 (LTS-SP1)']27 software = "os_system"28 ssh_command = ['cat', '/etc/system-release']29 ssh_command = self.deal_ssh_command(ssh_command,self.ssh_command,self.flag)30 status, check_result = self.check_result_contain(check_item, software,ssh_command,31 common_result_check_contain)32 if status:33 print(" The os os openEuler release 20.03 (LTS-SP1)") 34 return status,check_result35 36 def check_cat_fstab(self):37 """huge mount"""38 check_item = ["/mnt/huge"]39 software = "huge"40 ssh_command = ['cat', '/etc/fstab']41 ssh_command = self.deal_ssh_command(ssh_command,self.ssh_command,self.flag)42 status, check_result = self.check_result_contain(check_item, software,ssh_command,43 common_result_check_contain)44 if status:45 print(" The hugepage is config success") 46 return status,check_result47 def check_status_firewalld(self):48 check_item = ["inactive"]49 software = "firewalld"50 ssh_command = ['systemctl', 'status', 'firewalld']51 ssh_command = self.deal_ssh_command(ssh_command,self.ssh_command,self.flag)52 status, check_result = self.check_result_contain(check_item, software,ssh_command,53 common_result_check_contain)54 if status:55 print(" The firewalld status is incatiive") 56 return status,check_result57 def check_status_hinic3(self):58 check_item = ["hisdk3","hiovs3","hinic3"]59 software = "hinic3"60 ssh_command = ['lsmod']61 ssh_command = self.deal_ssh_command(ssh_command,self.ssh_command,self.flag)62 status, check_result = self.check_result_contain(check_item, software,ssh_command,63 common_result_check_contain)64 if status:65 print(' The SP680 OVS and live migration drivers are installed')66 return status,check_result67 def ovs_vsctl_show(self):68 check_item = ["dpdk","vf"]69 software = "ovs_vsctl_show"70 ssh_command = ['ovs-vsctl','show']71 ssh_command = self.deal_ssh_command(ssh_command,self.ssh_command,self.flag)72 status, check_result = self.check_result_contain(check_item, software,ssh_command,73 common_result_check_contain)74 if status:75 print(' The ovs configuration is normal')76 return status,check_result77 def check_vf_status(self):78 check_item = ["VirtIO"]79 software = "VF device"80 ssh_command = ['hinicadm3','info']81 ssh_command = self.deal_ssh_command(ssh_command,self.ssh_command,self.flag)82 status, check_result = self.check_result_contain(check_item, software,ssh_command,83 common_result_check_contain)84 if status:85 print(' The VF device has been created.')86 return status,check_result87 88 def check_openv_switch_status(self):89 """æ£æ¥openvswitchæå¡ç¶æ"""90 check_item = ['running']91 software = 'The openswitch service'92 ssh_command = ['service','openvswitch','status']93 ssh_command = self.deal_ssh_command(ssh_command,self.ssh_command,self.flag)94 status, check_result = self.check_result_contain(check_item, software,ssh_command, common_result_check_contain)95 if status:96 print(' The openswitch server is running')97 return status,check_result98 99 def check_vm_ping(self, ip, show_message):100 """èææºä¹é´äºpingè¿æ¥"""101 check_item = ['1 received']102 software = 'VM ping'103 ssh_command = ['ping','-c 1', ip]104 ssh_command = self.deal_ssh_command(ssh_command,self.ssh_command,self.flag)105 status, check_result = self.check_result_contain(check_item, software,ssh_command, common_result_check_contain)106 if status:107 print(" {}".format(show_message))...
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 autotest 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?
