blob: f17fcf8f20dc21d1a4796acca2ba745d2d55a9b9 [file] [log] [blame]
import inspect
import unittest
import json
import os
from nose.tools import assert_not_equal
from nose.plugins import Plugin
from CordTestUtils import log_test as log
from CordTestUtils import running_on_pod
from VolthaCtrl import voltha_setup, voltha_teardown
from SSHTestAgent import SSHTestAgent
log.setLevel('INFO')
class CordTestConfigRestore(Plugin):
name = 'cordTestConfigRestore'
context = None
restore_methods = ('configRestore', 'config_restore',)
def options(self, parser, env=os.environ):
super(CordTestConfigRestore, self).options(parser, env = env)
def configure(self, options, conf):
self.enabled = True
#just save the test case context on start
def startContext(self, context):
if inspect.isclass(context) and issubclass(context, unittest.TestCase):
if context.__name__.endswith('exchange'):
self.context = context
#reset the context on exit
def stopContext(self, context):
if inspect.isclass(context) and issubclass(context, unittest.TestCase):
if context.__name__.endswith('exchange'):
self.context = None
def doFailure(self, test, exception):
if self.context:
log.info('Inside test case failure for test: %s' %self.context.__name__)
for restore_method in self.restore_methods:
if hasattr(self.context, restore_method):
method = getattr(self.context, restore_method)
#check only for class/static methods
if method.__self__ is self.context:
method()
break
def addError(self, test, exception):
self.doFailure(test, exception)
def addFailure(self, test, exception):
self.doFailure(test, exception)
def get_test_class(module):
class_test = None
for name, obj in inspect.getmembers(module):
if inspect.isclass(obj) and issubclass(obj, unittest.TestCase):
if obj.__name__.endswith('exchange'):
class_test = obj
break
else:
class_test = obj
return class_test
def setup_module(module):
class_test = get_test_class(module)
assert_not_equal(class_test, None)
module_name = module.__name__.split('.')[-1]
cfg = '{}.json'.format(module_name)
module_config = os.path.join(os.path.dirname(module.__file__), cfg)
if os.access(module_config, os.F_OK):
with open(module_config) as f:
json_data = json.load(f)
for k, v in json_data.iteritems():
setattr(class_test, k, v)
#check for voltha and configure as appropriate
voltha_attrs = dict(host='172.17.0.1',
rest_port = 8881,
config_fake = False,
olt_type = 'ponsim_olt',
olt_mac = '00:0c:e2:31:12:00',
uplink_vlan_map = { 'of:0000000000000001' : '222' },
uplink_vlan_start = 333
)
voltha_enabled = bool(int(os.getenv('VOLTHA_ENABLED', 0)))
voltha_configure = True
if hasattr(class_test, 'VOLTHA_AUTO_CONFIGURE'):
voltha_configure = getattr(class_test, 'VOLTHA_AUTO_CONFIGURE')
if voltha_enabled and voltha_configure:
for k,v in voltha_attrs.iteritems():
voltha_attr = 'VOLTHA_{}'.format(k.upper())
if hasattr(class_test, voltha_attr):
v = getattr(class_test, voltha_attr)
voltha_attrs[k] = v
else:
setattr(class_test, voltha_attr, v)
ret = voltha_setup(**voltha_attrs)
if ret is not None:
#setup the stage to drop voltha on the way out
setattr(class_test, 'voltha_ctrl', ret[0])
setattr(class_test, 'voltha_device', ret[1])
setattr(class_test, 'voltha_switch_map', ret[2])
def teardown_module(module):
class_test = get_test_class(module)
if class_test is None:
return
if not hasattr(class_test, 'voltha_ctrl') or \
not hasattr(class_test, 'voltha_device') or \
not hasattr(class_test, 'voltha_switch_map'):
return
voltha_ctrl = getattr(class_test, 'voltha_ctrl')
voltha_device = getattr(class_test, 'voltha_device')
voltha_switch_map = getattr(class_test, 'voltha_switch_map')
voltha_teardown(voltha_ctrl, voltha_device, voltha_switch_map)
def running_on_ciab():
if running_on_pod() is False:
return False
head_node = os.getenv('HEAD_NODE', 'prod')
HEAD_NODE = head_node + '.cord.lab' if len(head_node.split('.')) == 1 else head_node
agent = SSHTestAgent(host = HEAD_NODE, user = 'ubuntu', password = 'ubuntu')
#see if user ubuntu works
st, output = agent.run_cmd('sudo virsh list')
if st is False and output is not None:
#we are on real pod
return False
#try vagrant
agent = SSHTestAgent(host = HEAD_NODE, user = 'vagrant', password = 'vagrant')
st, output = agent.run_cmd('sudo virsh list')
if st is True and output is not None:
return True
return False