self.test_plc=test_plc
self.test_node=test_node
self.test_slice=test_slice
- self.test_ssh =TestSsh(self.test_plc.test_ssh)
+ self.test_ssh = self.create_test_ssh()
- def is_local(self):
- return self.test_ssh.is_local()
-
- def host_to_guest(self,command):
- return self.test_plc.host_to_guest(self.test_plc.vserver,self.test_plc.vservername,command)
-
- def get_privateKey(self,slice_spec):
+ def get_privateKey(self):
+ slice_spec=self.test_slice.slice_spec
try:
- (found,remote_privatekey)=self.test_slice.locate_key(slice_spec)
- return (found,remote_privatekey)
+ (found,privatekey)=self.test_slice.locate_key()
+ return (found,privatekey)
except Exception,e:
print str(e)
- def get_initscript(self,slice_spec):
- (found,remote_privatekey)=self.get_privateKey(slice_spec)
- if not found :
+ def create_test_ssh(self):
+ (found,privatekey) = self.get_privateKey()
+ if not found:
raise Exception,"Cannot find a valid key for slice %s"%self.test_slice.name()
- for hostname in slice_spec['nodenames']:
- utils.header("Checking initiscript %s on the slice %s@%s"
- %(slice_spec['initscriptname'],self.test_slice.name(),hostname))
- init_file=self.test_plc.run_in_guest('ssh -i %s %s@%s ls -l /tmp/init* '
- %(remote_privatekey,self.test_slice.name(),hostname))
- if ( init_file):
- return False
-
- return True
+ return TestSsh (self.test_node.name(),key=privatekey,username=self.test_slice.name(),
+ # so that copies end up in the home dir
+ buildname=".")
+
+ def name (self):
+ return "%s@%s"%(self.test_slice.name(),self.test_node.name())
+
+ def check_initscript(self,initscript):
+ slice_spec=self.test_slice.slice_spec
+ initscript = slice_spec['initscriptname']
+ utils.header("Checking initscript %s on sliver %s"%(initscript,self.name()))
+ return self.test_ssh.run("ls -l /tmp/%s.stamp"%initscript)==0
- def run_tcp_server (self,port):
- print ("XXX run_tcp_server not implemented")
- return True
- def run_tcp_client (self,hostname,port):
- print ("XXX run_tcp_client not implemented")
- return True
-
- def run_tcpcheck(self,peer_spec,remote_privatekey):
- if peer_spec['peer_name']=="server":
- slice_ssh = TestSsh (peer_spec['server_name'],
- key=remote_privatekey,
- username=peer_spec['slice_name'])
- remote_server_command=slice_ssh.actual_command("./tcptest.py server -t 10")
- return self.test_plc.run_in_guest(remote_server_command)
-
- else:
- slice_ssh = TestSsh (peer_spec['client_name'],
- key=remote_privatekey,
- username=peer_spec['slice_name'])
- client_command="/tcptest.py client -a %s -p %d"%(peer_spec['peer_server'],
- peer_spec['server_port'])
- remote_client_command=slice_ssh.actual_command(client_command)
- return self.test_plc.run_in_guest(remote_client_command)
-
- def do_check_tcp(self,tcp_param,options):
- for tcp_spec in tcp_param:
- #copy the tcptest file under the chroot
- localfile=remotefile="tcptest.py"
- self.test_plc.copy_in_guest(localfile, remotefile, False)
- peer_param=tcp_spec['tcp_fields']
- if (tcp_spec['tcp_fields']['peer_name']=='server'):
- #server instruction
- utils.header("Transfert the tcp script to the server at %s@%s"%(peer_param['slice_name'],
- peer_param['server_name']))
- slice_spec=self.test_slice.get_slice(peer_param['slice_name'])
- (found,remote_privatekey)=self.get_privateKey(slice_spec)
- cp_server_command="scp -i %s ./tcptest.py %s@%s:"%(remote_privatekey,peer_param['slice_name'],
- peer_param['server_name'])
- self.test_plc.run_in_guest(cp_server_command)
- serv_status=self.run_tcpcheck(peer_param,remote_privatekey)
- if (serv_status):
- utils.header("FAILED to check loop Connexion on the %s server side"%peer_param['server_name'])
- return False
- else:
- #Client instruction
- utils.header("Transfert the tcp script to the client at %s@%s" %(peer_param['slice_name'],
- peer_param['client_name']))
- slice_spec=self.test_slice.get_slice(peer_param['slice_name'])
- (found,remote_privatekey)=self.get_privateKey(slice_spec)
- cp_client_command="scp -i %s ./tcptest.py %s@%s:"%(remote_privatekey, peer_param['slice_name'],
- peer_param['client_name'])
- self.test_plc.run_in_guest(cp_client_command)
- client_status=self.run_tcpcheck(peer_param,remote_privatekey)
- if ( serv_status):
- utils.header("FAILED to Contact the server %s from the client side %s"%(peer_param['peer_server'],
- peer_param['client_name']))
- return False
-
-
- self.test_plc.run_in_guest("rm -rf tcptest.py")
- return True
+ def run_tcp_server (self,port,timeout=10):
+ server_command = "tcptest.py server -p %d -t %d"%(port,timeout)
+ return self.test_ssh.copy("tcptest.py")==0 and self.test_ssh.run(server_command)==0
+ def run_tcp_client (self,servername,port):
+ client_command="tcptest.py client -a %s -p %d"%(servername,port)
+ return self.test_ssh.copy("tcptest.py")==0 and self.test_ssh.run(client_command)==0
+ def tar_var_logs (self):
+ return self.test_ssh.actual_command("sudo tar -C /var/log -cf - .")
+
git://git.onelab.eu / tests.git / blobdiff