New cord-tester cross connect app to configure the leaf switch for cord-tester.
This is required when cord-tester is running on the head node on the PODD.
And compute node access is needed to test vcpes from inside the podd head node.
For now, we don't enable the configuration of the app as one cannot overwrite default xconnect flows.
Like eg: 222 which would be already configured.
In this case, we would just need new entries in olt config or use a new subscriber vcpe list
to open up for cord-tester vsg testing.
Change-Id: Ie96c3eba79aded235e67f05d806722abe6024ffa
diff --git a/src/test/vsg/vsgTest.py b/src/test/vsg/vsgTest.py
index 125397b..82eb147 100644
--- a/src/test/vsg/vsgTest.py
+++ b/src/test/vsg/vsgTest.py
@@ -50,6 +50,10 @@
subscriber_map = {}
restore_methods = []
TIMEOUT=120
+ FABRIC_PORT_HEAD_NODE = 1
+ FABRIC_PORT_COMPUTE_NODE = 2
+ APP_NAME = 'org.ciena.xconnect'
+ APP_FILE = os.path.join(test_path, '..', 'apps/xconnect-1.0-SNAPSHOT.oar')
@classmethod
def getSubscriberCredentials(cls, subId):
@@ -133,6 +137,35 @@
cls.restApiXos = restApiXos
@classmethod
+ def closeVCPEAccess(cls, vcpes):
+ return
+ OnosCtrl.uninstall_app(cls.APP_NAME, onos_ip = cls.HEAD_NODE)
+
+ @classmethod
+ def openVCPEAccess(cls, vcpes):
+ """
+ This code works below to configure the leaf switch.
+ But it needs the olt_config.json to be modified to not overlap with existing/default vcpes.
+ That is to avoid overwriting the flows already provisioned for eg: for 222 vcpe.
+ (default and created on CiaB).
+ So returning for now with a no-op
+ """
+ return
+ OnosCtrl.install_app(cls.APP_FILE, onos_ip = cls.HEAD_NODE)
+ time.sleep(2)
+ s_tags = map(lambda vcpe: int(vcpe['s_tag']), vcpes)
+ devices = OnosCtrl.get_device_ids(controller = cls.HEAD_NODE)
+ device_config = {}
+ for device in devices:
+ device_config[device] = []
+ for s_tag in s_tags:
+ xconnect_config = {'vlan': s_tag, 'ports' : [ cls.FABRIC_PORT_HEAD_NODE, cls.FABRIC_PORT_COMPUTE_NODE ] }
+ device_config[device].append(xconnect_config)
+
+ cfg = { 'apps' : { 'org.ciena.xconnect' : { 'xconnectTestConfig' : device_config } } }
+ OnosCtrl.config(cfg, controller = cls.HEAD_NODE)
+
+ @classmethod
def setUpClass(cls):
cls.controllers = get_controllers()
cls.controller = cls.controllers[0]
@@ -156,10 +189,14 @@
cls.vcpe_dhcp_stag = vcpe_dhcp_stag
VSGAccess.setUp()
cls.setUpCordApi()
+ if cls.on_podd is True:
+ cls.openVCPEAccess(cls.vcpes_dhcp)
@classmethod
def tearDownClass(cls):
VSGAccess.tearDown()
+ if cls.on_podd is True:
+ cls.closeVCPEAccess(cls.vcpes_dhcp)
def cliEnter(self, controller = None):
retries = 0