Refactor all python test libraries to conform to python coding guidelines
Refactor robot Test Case Suite to conform to Robot code coding guidelines
Now that I have pycharm installed in a cloned VM that I resized to allow for
development tools, I see coding issues with the original code that I had
This is mostly a cosmetic change as no fundamental changes were made
to the original code
Rebased and fixed conflicts
Change-Id: I5dc0534e92fa708b45399944994101afd7efed63
diff --git a/tests/atests/common/volthaMngr.py b/tests/atests/common/volthaMngr.py
index c164989..9b314c2 100755
--- a/tests/atests/common/volthaMngr.py
+++ b/tests/atests/common/volthaMngr.py
@@ -20,48 +20,48 @@
import os
-import time
import subprocess
import testCaseUtils
import logging
+
class VolthaMngr(object):
"""
This class implements voltha startup/shutdown callable helper functions
"""
def __init__(self):
- self.dirs = {}
- self.dirs ['root'] = None
- self.dirs ['voltha'] = None
- self.dirs ['log'] = None
+ self.dirs = dict()
+ self.dirs['root'] = None
+ self.dirs['voltha'] = None
+ self.dirs['log'] = None
- def vSetLogDirs(self, rootDir, volthaDir, logDir):
- testCaseUtils.configDirs(self, logDir, rootDir, volthaDir)
+ def v_set_log_dirs(self, root_dir, voltha_dir, log_dir):
+ testCaseUtils.config_dirs(self, log_dir, root_dir, voltha_dir)
- def startAllPods(self):
- proc1 = subprocess.Popen([testCaseUtils.getDir(self, 'root') + '/build.sh', 'start'],
+ def start_all_pods(self):
+ proc1 = subprocess.Popen([testCaseUtils.get_dir(self, 'root') + '/build.sh', 'start'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
output = proc1.communicate()[0]
print(output)
- proc1.stdout.close
+ proc1.stdout.close()
- def stopAllPods(self):
- proc1 = subprocess.Popen([testCaseUtils.getDir(self, 'root') + '/build.sh', 'stop'],
+ def stop_all_pods(self):
+ proc1 = subprocess.Popen([testCaseUtils.get_dir(self, 'root') + '/build.sh', 'stop'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
output = proc1.communicate()[0]
print(output)
- proc1.stdout.close
+ proc1.stdout.close()
- def resetKubeAdm(self):
- proc1 = subprocess.Popen([testCaseUtils.getDir(self, 'root') + '/build.sh', 'clear'],
+ def reset_kube_adm(self):
+ proc1 = subprocess.Popen([testCaseUtils.get_dir(self, 'root') + '/build.sh', 'clear'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
output = proc1.communicate()[0]
print(output)
- proc1.stdout.close
+ proc1.stdout.close()
"""
Because we are not deploying SEBA with XOS and NEM, and that a standalone Voltha
@@ -69,63 +69,62 @@
NetCfg in two fashion.
One is to add to device list and the other is to add the missing Sadis section
"""
- def alterOnosNetCfg(self):
- logging.info ('Altering the Onos NetCfg to suit Voltha\'s needs')
- logging.debug ('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
- 'http://localhost:30120/onos/v1/network/configuration/devices/ -d @%s/tests/atests/build/devices_json'
- % testCaseUtils.getDir(self, 'voltha'))
+ def alter_onos_net_cfg(self):
+ logging.info('Altering the Onos NetCfg to suit Voltha\'s needs')
+ logging.debug('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
+ 'http://localhost:30120/onos/v1/network/configuration/devices/ -d @%s/tests/atests/build/devices_json'
+ % testCaseUtils.get_dir(self, 'voltha'))
os.system('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
- 'http://localhost:30120/onos/v1/network/configuration/devices/ -d @%s/tests/atests/build/devices_json'
- % testCaseUtils.getDir(self, 'voltha'))
- logging.debug ('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
- 'http://localhost:30120/onos/v1/network/configuration/apps/ -d @%s/tests/atests/build/sadis_json'
- % testCaseUtils.getDir(self, 'voltha'))
+ 'http://localhost:30120/onos/v1/network/configuration/devices/ -d @%s/tests/atests/build/devices_json'
+ % testCaseUtils.get_dir(self, 'voltha'))
+ logging.debug('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
+ 'http://localhost:30120/onos/v1/network/configuration/apps/ -d @%s/tests/atests/build/sadis_json'
+ % testCaseUtils.get_dir(self, 'voltha'))
os.system('curl --user karaf:karaf -X POST -H "Content-Type: application/json" '
- 'http://localhost:30120/onos/v1/network/configuration/apps/ -d @%s/tests/atests/build/sadis_json'
- % testCaseUtils.getDir(self, 'voltha'))
-
- def getAllRunningPods(self):
- allRunningPods = []
- proc1 = subprocess.Popen(['/usr/bin/kubectl', 'get', 'pods', '--all-namespaces'],
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE)
- proc2 = subprocess.Popen(['grep', '-v', 'NAMESPACE'], stdin=proc1.stdout,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE)
- proc1.stdout.close
- out, err = proc2.communicate()
- print(out)
- if out:
- for line in out.split('\n'):
- items = line.split()
- nsName = {}
- if len(items) > 2:
- nsName = {}
- nsName['NS'] = items[0]
- nsName['Name'] = items[1]
- allRunningPods.append(nsName)
- return allRunningPods
-
- def collectPodLogs(self):
+ 'http://localhost:30120/onos/v1/network/configuration/apps/ -d @%s/tests/atests/build/sadis_json'
+ % testCaseUtils.get_dir(self, 'voltha'))
+
+ def collect_pod_logs(self):
logging.info('Collect logs from all Pods')
- allRunningPods = self.getAllRunningPods()
+ allRunningPods = get_all_running_pods()
for nsName in allRunningPods:
Namespace = nsName.get('NS')
- podName = nsName.get('Name')
+ podName = nsName.get('Name')
if 'onos' not in podName:
os.system('/usr/bin/kubectl logs -n %s -f %s > %s/%s.log 2>&1 &' %
- (Namespace, podName, testCaseUtils.getDir(self, 'log'), podName))
+ (Namespace, podName, testCaseUtils.get_dir(self, 'log'), podName))
else:
os.system('/usr/bin/kubectl logs -n %s -f %s onos > %s/%s.log 2>&1 &' %
- (Namespace, podName, testCaseUtils.getDir(self, 'log'), podName))
+ (Namespace, podName, testCaseUtils.get_dir(self, 'log'), podName))
-def voltha_Initialize(rootDir, volthaDir, logDir):
- voltha = VolthaMngr()
- voltha.vSetLogDirs(rootDir, volthaDir, logDir)
- voltha.stopAllPods()
- voltha.resetKubeAdm()
- voltha.startAllPods()
- voltha.alterOnosNetCfg()
- voltha.collectPodLogs()
+def get_all_running_pods():
+ allRunningPods = []
+ proc1 = subprocess.Popen(['/usr/bin/kubectl', 'get', 'pods', '--all-namespaces'],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE)
+ proc2 = subprocess.Popen(['grep', '-v', 'NAMESPACE'], stdin=proc1.stdout,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE)
+ proc1.stdout.close()
+ out, err = proc2.communicate()
+ print(out)
+ if out:
+ for line in out.split('\n'):
+ items = line.split()
+ if len(items) > 2:
+ nsName = dict()
+ nsName['NS'] = items[0]
+ nsName['Name'] = items[1]
+ allRunningPods.append(nsName)
+ return allRunningPods
+
+def voltha_initialize(root_dir, voltha_dir, log_dir):
+ voltha = VolthaMngr()
+ voltha.v_set_log_dirs(root_dir, voltha_dir, log_dir)
+ voltha.stop_all_pods()
+ voltha.reset_kube_adm()
+ voltha.start_all_pods()
+ voltha.alter_onos_net_cfg()
+ voltha.collect_pod_logs()