sys.path.append('transportpce_tests/common/')
import test_utils
+
class TransportPCERendererTesting(unittest.TestCase):
processes = None
res['errors']['error'])
def test_17_service_path_delete_rdm_check(self):
- response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
+ response = test_utils.check_netconf_node_request(
+ "ROADMA01", "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
"dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"},
{"node-id": "XPDRC01",
"dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}],
- 196.1, 40, 196.075, 196.125, 761,
- 768)
+ 196.1, 40, 196.075, 196.125, 761,
+ 768)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
"src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
{"node-id": "XPDRA01",
"src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}],
- 196.1, 40, 196.075, 196.125, 761,
- 768)
+ 196.1, 40, 196.075, 196.125, 761,
+ 768)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
def test_21_get_roadmconnection_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
self.assertEqual(-3.3, res['roadm-connections'][0]['target-output-power'])
def test_22_get_roadmconnection_ROADMC(self):
- response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode'])
self.assertEqual(1, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
def test_25_get_roadmconnection_ROADMC(self):
- response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
self.assertIn('Success', res["output"]["result"])
def test_27_get_roadmconnection_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
self.assertEqual(-60, res['roadm-connections'][0]['target-output-power'])
def test_28_get_roadmconnection_ROADMC(self):
- response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADMC01", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP1-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
"dest-tp": "SRG1-PP1-TXRX", "src-tp": "DEG2-TTP-TXRX"},
{"node-id": "XPDRC01",
"dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}],
- 196.1, 40, 196.075, 196.125, 761,
- 768)
+ 196.1, 40, 196.075, 196.125, 761,
+ 768)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
"src-tp": "DEG1-TTP-TXRX", "dest-tp": "SRG1-PP1-TXRX"},
{"node-id": "XPDRA01",
"src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}],
- 196.1, 40, 196.075, 196.125, 761,
- 768)
+ 196.1, 40, 196.075, 196.125, 761,
+ 768)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
"dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
{"node-id": "ROADMA01",
"dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}],
- 196.05, 40, 196.025, 196.075, 753,
- 760)
+ 196.05, 40, 196.025, 196.075, 753,
+ 760)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
"dest-tp": "XPDR1-NETWORK2", "src-tp": "XPDR1-CLIENT2"},
{"node-id": "ROADMA01",
"dest-tp": "DEG1-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}],
- 196.1, 40, 196.075, 196.125, 761,
- 768)
+ 196.1, 40, 196.075, 196.125, 761,
+ 768)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
sys.path.append('transportpce_tests/common/')
import test_utils
+
class TransportPCEFulltesting(unittest.TestCase):
cr_serv_sample_data = {"input": {
"sdnc-request-header": {
time.sleep(2)
def test_13_check_xc1_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(5)
def test_14_check_xc1_ROADMC(self):
- response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
if ele['tp-id'] == 'DEG2-TTP-TXRX':
- freq_map = base64.b64decode(
- ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
- freq_map_array = [int(x) for x in freq_map]
- self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
+ freq_map = base64.b64decode(
+ ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
+ freq_map_array = [int(x) for x in freq_map]
+ self.assertEqual(freq_map_array[95], 0, "Lambda 1 should not be available")
time.sleep(3)
def test_18_connect_xprdA_N2_to_roadmA_PP2(self):
time.sleep(1)
def test_24_check_xc2_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP2-TXRX-753:760")
+ response = test_utils.check_netconf_node_request(
+ "ROADMA01", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP2-TXRX-753:760")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(1)
def test_38_check_xc1_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADMA01", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(7)
def test_39_check_xc1_ROADMC(self):
- response = test_utils.check_netconf_node_request("ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADMC01", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(2)
def test_42_check_xc2_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP2-TXRX-DEG1-TTP-TXRX-753:760")
+ response = test_utils.check_netconf_node_request(
+ "ROADMA01", "roadm-connections/SRG1-PP2-TXRX-DEG1-TTP-TXRX-753:760")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
if tpType == 'XPONDER-CLIENT':
client += 1
print("tpId = {}".format(tpId))
- print("tp= {}".format(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]))
+ print("tp= {}".format(res['network'][0]['node'][i]
+ ['ietf-network-topology:termination-point'][j]))
nbIfCapType = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-otn-network-topology:tp-supported-interfaces']
- ['supported-interface-capability'][0])
+ ['org-openroadm-otn-network-topology:tp-supported-interfaces']
+ ['supported-interface-capability'][0])
for k in range(0, nbIfCapType):
self.assertIn((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-otn-network-topology:tp-supported-interfaces']
- ['supported-interface-capability'][0]['if-cap-type']),
- CHECK_LIST[nodeId]['port-types'])
+ ['org-openroadm-otn-network-topology:tp-supported-interfaces']
+ ['supported-interface-capability'][0]['if-cap-type']),
+ CHECK_LIST[nodeId]['port-types'])
elif tpType == 'XPONDER-NETWORK':
network += 1
self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
res = response.json()
self.assertEqual(
{
- "mc-capabilities": [
- {
- "mc-node-name": "DEG1-TTP",
- "center-freq-granularity": 6.25,
- "slot-width-granularity": 12.5
- }
- ]
+ "mc-capabilities": [
+ {
+ "mc-node-name": "DEG1-TTP",
+ "center-freq-granularity": 6.25,
+ "slot-width-granularity": 12.5
+ }
+ ]
}, res)
time.sleep(3)
res = response.json()
self.assertEqual(
{
- "mc-capabilities": [
- {
- "mc-node-name": "DEG2-TTP",
- "center-freq-granularity": 6.25,
- "slot-width-granularity": 12.5
- }
- ]
+ "mc-capabilities": [
+ {
+ "mc-node-name": "DEG2-TTP",
+ "center-freq-granularity": 6.25,
+ "slot-width-granularity": 12.5
+ }
+ ]
}, res)
time.sleep(3)
res = response.json()
self.assertEqual(
{
- "mc-capabilities": [
- {
- "mc-node-name": "SRG1-PP",
- "center-freq-granularity": 6.25,
- "slot-width-granularity": 12.5
- }
- ]
+ "mc-capabilities": [
+ {
+ "mc-node-name": "SRG1-PP",
+ "center-freq-granularity": 6.25,
+ "slot-width-granularity": 12.5
+ }
+ ]
}, res)
time.sleep(3)
def test_07_device_renderer(self):
data = {
"transportpce-device-renderer:input": {
- "transportpce-device-renderer:modulation-format": "dp-qpsk",
- "transportpce-device-renderer:operation": "create",
- "transportpce-device-renderer:service-name": "testNMC-MC",
- "transportpce-device-renderer:wave-number": "0",
- "transportpce-device-renderer:center-freq": "196.05",
- "transportpce-device-renderer:width": "80",
- "transportpce-device-renderer:nodes": [
- {
- "transportpce-device-renderer:node-id": "ROADM-D1",
- "transportpce-device-renderer:src-tp": "SRG1-PP1-TXRX",
- "transportpce-device-renderer:dest-tp": "DEG1-TTP-TXRX"
- }
- ],
- "transportpce-device-renderer:min-freq": 196.00625,
- "transportpce-device-renderer:max-freq": 196.09375,
- "transportpce-device-renderer:lower-spectral-slot-number": 749,
- "transportpce-device-renderer:higher-spectral-slot-number": 763
- }
+ "transportpce-device-renderer:modulation-format": "dp-qpsk",
+ "transportpce-device-renderer:operation": "create",
+ "transportpce-device-renderer:service-name": "testNMC-MC",
+ "transportpce-device-renderer:wave-number": "0",
+ "transportpce-device-renderer:center-freq": "196.05",
+ "transportpce-device-renderer:width": "80",
+ "transportpce-device-renderer:nodes": [
+ {
+ "transportpce-device-renderer:node-id": "ROADM-D1",
+ "transportpce-device-renderer:src-tp": "SRG1-PP1-TXRX",
+ "transportpce-device-renderer:dest-tp": "DEG1-TTP-TXRX"
+ }
+ ],
+ "transportpce-device-renderer:min-freq": 196.00625,
+ "transportpce-device-renderer:max-freq": 196.09375,
+ "transportpce-device-renderer:lower-spectral-slot-number": 749,
+ "transportpce-device-renderer:higher-spectral-slot-number": 763
+ }
}
url = test_utils.RESTCONF_BASE_URL + \
- "/operations/transportpce-device-renderer:service-path"
+ "/operations/transportpce-device-renderer:service-path"
response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
"description": "TBD",
"supporting-port": "L1",
"type": "org-openroadm-interfaces:mediaChannelTrailTerminationPoint"},
- **res['interface'][0]),
+ **res['interface'][0]),
res['interface'][0])
# Check the mc-ttp max and min-freq
self.assertEqual({
- "min-freq": 196.00625,
- "max-freq": 196.09375
- },
+ "min-freq": 196.00625,
+ "max-freq": 196.09375
+ },
res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'])
time.sleep(3)
# Check the mc-ttp max and min-freq
self.assertEqual({
- "frequency": 196.05,
- "width": 80
+ "frequency": 196.05,
+ "width": 80
},
- res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
+ res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
time.sleep(3)
# get SRG-NMC interface
# Create ROADM-connection
def test_11_roadm_connection(self):
response = test_utils.check_netconf_node_request("ROADM-D1",
- "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
+ "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763",
# delete ROADM connection
def test_12_delete_roadm_connection(self):
response = test_utils.delete_request(test_utils.URL_CONFIG_NETCONF_TOPO +
- "node/ROADM-D1/yang-ext:mount/" +
- "org-openroadm-device:org-openroadm-device/" +
- "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
+ "node/ROADM-D1/yang-ext:mount/" +
+ "org-openroadm-device:org-openroadm-device/" +
+ "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
# Delete NMC SRG interface
def test_13_delete_srg_interface(self):
response = test_utils.delete_request(test_utils.URL_CONFIG_NETCONF_TOPO +
- "node/ROADM-D1/yang-ext:mount/" +
- "org-openroadm-device:org-openroadm-device/" +
- "interface/SRG1-PP1-TXRX-nmc-749:763")
+ "node/ROADM-D1/yang-ext:mount/" +
+ "org-openroadm-device:org-openroadm-device/" +
+ "interface/SRG1-PP1-TXRX-nmc-749:763")
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
res['errors']['error'])
def test_10_service_path_create_rdm_check(self):
- response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-A1", "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
{"node-id": "XPDR-C1",
"dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}],
196.1, 40, 196.075, 196.125, 761,
- 768)
+ 768)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
{"node-id": "XPDR-A1",
"src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}],
196.1, 40, 196.075, 196.125, 761,
- 768)
+ 768)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
self.assertEqual(196.1, res['org-openroadm-optical-channel-interfaces:och']['frequency'])
def test_21_get_roadmconnection_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
self.assertEqual(2.0, res['roadm-connections'][0]['target-output-power'])
def test_22_get_roadmconnection_ROADMC(self):
- response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode'])
self.assertEqual(196.1, res['org-openroadm-optical-channel-interfaces:och']['frequency'])
def test_25_get_roadmconnection_ROADMC(self):
- response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
self.assertIn('Success', res["output"]["result"])
def test_27_get_roadmconnection_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
self.assertEqual(-60, res['roadm-connections'][0]['target-output-power'])
def test_28_get_roadmconnection_ROADMC(self):
- response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-C1", "roadm-connections/DEG1-TTP-TXRX-SRG1-PP1-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
{"node-id": "XPDR-C1",
"dest-tp": "XPDR1-CLIENT1", "src-tp": "XPDR1-NETWORK1"}],
196.1, 40, 196.075, 196.125, 761,
- 768)
+ 768)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
{"node-id": "XPDR-A1",
"src-tp": "XPDR1-NETWORK1", "dest-tp": "XPDR1-CLIENT1"}],
196.053125, 40, 196.025, 196.08125, 761,
- 768)
+ 768)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
{"node-id": "ROADM-A1",
"dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}],
196.1, 40, 196.075, 196.125, 753,
- 760)
+ 760)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
{"node-id": "ROADM-A1",
"dest-tp": "DEG2-TTP-TXRX", "src-tp": "SRG1-PP2-TXRX"}],
196.053125, 40, 196.025, 196.08125, 761,
- 768)
+ 768)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
+
class TransportTapitesting(unittest.TestCase):
processes = None
self.assertEqual("Tpdr100g over WDM node", res["output"]["topology"]["node"][0]["name"][0]["value"],
'node name should be: Tpdr100g over WDM node')
self.assertIn("ETH", res["output"]["topology"]["node"][0]["layer-protocol-name"],
- 'Node layer protocol should contain ETH')
+ 'Node layer protocol should contain ETH')
self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
'node should contain 1 node rule group')
self.assertEqual("ROADM-infra", res["output"]["topology"]["node"][0]["name"][0]["value"],
'node name should be: ROADM-infra')
self.assertIn("PHOTONIC_MEDIA", res["output"]["topology"]["node"][0]["layer-protocol-name"],
- 'Node layer protocol should contain PHOTONIC_MEDIA')
+ 'Node layer protocol should contain PHOTONIC_MEDIA')
self.assertEqual(1, len(res["output"]["topology"]["node"][0]["node-rule-group"]),
'node should contain 1 node rule group')
for link in links:
if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
- 'OTU4 link should have an available capacity of 100 000 Mbps')
+ 'OTU4 link should have an available capacity of 100 000 Mbps')
elif link["name"][0]["value-name"] == "transitional link name":
self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
- 'link should have an available capacity of 100 Gbps')
+ 'link should have an available capacity of 100 Gbps')
self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
def test_33_create_ODU4_service(self):
for link in links:
if link["name"][0]["value"] == "OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
self.assertEqual(0, link["available-capacity"]["total-size"]["value"],
- 'OTU4 link should have an available capacity of 0 Mbps')
+ 'OTU4 link should have an available capacity of 0 Mbps')
elif link["name"][0]["value"] == "ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1":
self.assertEqual(100000, link["available-capacity"]["total-size"]["value"],
- 'ODU4 link should have an available capacity of 100 000 Mbps')
+ 'ODU4 link should have an available capacity of 100 000 Mbps')
elif link["name"][0]["value-name"] == "transitional link name":
self.assertEqual(100, link["available-capacity"]["total-size"]["value"],
- 'link should have an available capacity of 100 Gbps')
+ 'link should have an available capacity of 100 Gbps')
self.assertEqual(2, len(link["node-edge-point"]), 'link should have 2 neps')
def test_35_connect_sprda_2_n2_to_roadma_pp3(self):
links = res['network'][0]['ietf-network-topology:link']
for link in links:
if (link["org-openroadm-common-network:link-type"] == "XPONDER-OUTPUT" or
- link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
- link_name = link["link-id"]
- response = test_utils.delete_request(url+link_name)
- self.assertEqual(response.status_code, requests.codes.ok)
+ link["org-openroadm-common-network:link-type"] == "XPONDER-INPUT"):
+ link_name = link["link-id"]
+ response = test_utils.delete_request(url+link_name)
+ self.assertEqual(response.status_code, requests.codes.ok)
def test_42_check_tapi_topology_T0(self):
url = "{}/operations/tapi-topology:get-topology-details"
return dict_
return None
+
def count_object_with_double_key(list_dicts, key1, key2, value):
nb = 0
for dict in list_dicts:
if __name__ == "__main__":
- unittest.main(verbosity=2)
\ No newline at end of file
+ unittest.main(verbosity=2)
# pylint: disable=no-member
# pylint: disable=too-many-public-methods
+
import base64
import unittest
import time
sys.path.append('transportpce_tests/common/')
import test_utils
-
class TransportPCEFulltesting(unittest.TestCase):
processes = None
time.sleep(2)
def test_13_check_xc1_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(5)
def test_14_check_xc1_ROADMC(self):
- response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
if ele['tp-id'] == 'DEG2-TTP-TXRX':
- freq_map = base64.b64decode(
- ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
- freq_map_array = [int(x) for x in freq_map]
- self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
+ freq_map = base64.b64decode(
+ ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
+ freq_map_array = [int(x) for x in freq_map]
+ self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
time.sleep(3)
def test_18_connect_xprdA_N2_to_roadmA_PP2(self):
time.sleep(1)
def test_24_check_xc2_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP2-TXRX-753:760")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-A1", "roadm-connections/DEG2-TTP-TXRX-SRG1-PP2-TXRX-753:760")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(1)
def test_38_check_xc1_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(7)
def test_39_check_xc1_ROADMC(self):
- response = test_utils.check_netconf_node_request("ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-C1", "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-761:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(2)
def test_42_check_xc2_ROADMA(self):
- response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP2-TXRX-DEG2-TTP-TXRX-753:760")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-A1", "roadm-connections/SRG1-PP2-TXRX-DEG2-TTP-TXRX-753:760")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
res = response.json()
self.assertIn(
{'supported-interface-capability':
- ['org-openroadm-port-types:if-otsi-otsigroup'],
+ ['org-openroadm-port-types:if-otsi-otsigroup'],
'supporting-port': 'L1',
'supporting-circuit-pack-name': '1/1/2-PLUG-NET',
'logical-connection-point': 'XPDR1-NETWORK1',
res = response.json()
self.assertIn(
{'supported-interface-capability':
- ['org-openroadm-port-types:if-otsi-otsigroup'],
+ ['org-openroadm-port-types:if-otsi-otsigroup'],
'supporting-port': 'L1',
'supporting-circuit-pack-name': '1/2/2-PLUG-NET',
'logical-connection-point': 'XPDR2-NETWORK1',
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('org-openroadm-port-types:if-100GE-ODU4',
- res['mapping'][0]['supported-interface-capability'])
+ res['mapping'][0]['supported-interface-capability'])
self.assertIn('org-openroadm-port-types:if-OCH-OTU4-ODU4',
- res['mapping'][0]['supported-interface-capability'])
+ res['mapping'][0]['supported-interface-capability'])
self.assertEqual('C1', res['mapping'][0]['supporting-port'])
self.assertEqual('1/2/1/1-PLUG-CLIENT', res['mapping'][0]['supporting-circuit-pack-name'])
self.assertEqual('XPDR2-CLIENT1', res['mapping'][0]['logical-connection-point'])
self.assertEqual('InService', res['mapping'][0]['port-admin-state'])
self.assertEqual('InService', res['mapping'][0]['port-oper-state'])
self.assertEqual({
- "min-trib-slot": "1.1",
- "max-trib-slot": "1.20"
- }, res['mapping'][0]['mpdr-restrictions'])
+ "min-trib-slot": "1.1",
+ "max-trib-slot": "1.20"
+ }, res['mapping'][0]['mpdr-restrictions'])
# Added test to check mc-capability-profile for a transponder
def test_08_check_mccapprofile(self):
if __name__ == '__main__':
- unittest.main(verbosity=2)
+ unittest.main(verbosity=2)
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-import os
-
SIMS = {
('xpdra', '1.2.1'): {'port': '17830', 'configfile': 'oper-XPDRA.xml', 'logfile': 'xpdra-121.log'},
('roadma', '1.2.1'): {'port': '17831', 'configfile': 'oper-ROADMA.xml', 'logfile': 'roadma-121.log'},
('xpdrc', '7.1'): {'port': '17854', 'configfile': 'oper-XPDRC.xml', 'logfile': 'xpdrc-71.log'},
('xpdra2', '7.1'): {'port': '17857', 'configfile': 'oper-XPDRA2.xml', 'logfile': 'xpdra2-71.log'},
('xpdrc2', '7.1'): {'port': '17858', 'configfile': 'oper-XPDRC2.xml', 'logfile': 'xpdrc2-71.log'}
-}
\ No newline at end of file
+}
headers=TYPE_APPLICATION_XML,
auth=(ODL_LOGIN, ODL_PWD))
+
def put_jsonrequest(url, data):
return requests.request(
"PUT", url.format(RESTCONF_BASE_URL),
headers=TYPE_APPLICATION_JSON,
auth=(ODL_LOGIN, ODL_PWD))
+
def rawput_request(url, data):
return requests.request(
"PUT", url.format(RESTCONF_BASE_URL),
headers=TYPE_APPLICATION_JSON,
auth=(ODL_LOGIN, ODL_PWD))
+
def rawpost_request(url, data):
return requests.request(
"POST", url.format(RESTCONF_BASE_URL),
# pylint: disable=no-member
# pylint: disable=too-many-public-methods
import json
-import base64
import unittest
import time
import requests
sys.path.append('transportpce_tests/common/')
import test_utils
-
class TransportPCEFulltesting(unittest.TestCase):
processes = None
def test_13_change_status_line_port_xpdra(self):
url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
body = {"ports": [{
- "port-name": "1",
- "logical-connection-point": "XPDR1-NETWORK1",
- "port-type": "CFP2",
- "circuit-id": "XPDRA-NETWORK",
- "administrative-state": "outOfService",
- "port-qual": "xpdr-network"}]}
+ "port-name": "1",
+ "logical-connection-point": "XPDR1-NETWORK1",
+ "port-type": "CFP2",
+ "circuit-id": "XPDRA-NETWORK",
+ "administrative-state": "outOfService",
+ "port-qual": "xpdr-network"}]}
response = requests.request("PUT", url.format("http://127.0.0.1:8130/restconf"),
data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
def test_17_restore_status_line_port_xpdra(self):
url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
body = {"ports": [{
- "port-name": "1",
- "logical-connection-point": "XPDR1-NETWORK1",
- "port-type": "CFP2",
- "circuit-id": "XPDRA-NETWORK",
- "administrative-state": "inService",
- "port-qual": "xpdr-network"}]}
+ "port-name": "1",
+ "logical-connection-point": "XPDR1-NETWORK1",
+ "port-type": "CFP2",
+ "circuit-id": "XPDRA-NETWORK",
+ "administrative-state": "inService",
+ "port-qual": "xpdr-network"}]}
response = requests.request("PUT", url.format("http://127.0.0.1:8130/restconf"),
data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
def test_21_change_status_port_roadma_srg(self):
url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
body = {"ports": [{
- "port-name": "C1",
- "logical-connection-point": "SRG1-PP1",
- "port-type": "client",
- "circuit-id": "SRG1",
- "administrative-state": "outOfService",
- "port-qual": "roadm-external"}]}
+ "port-name": "C1",
+ "logical-connection-point": "SRG1-PP1",
+ "port-type": "client",
+ "circuit-id": "SRG1",
+ "administrative-state": "outOfService",
+ "port-qual": "roadm-external"}]}
response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
def test_24_restore_status_port_roadma_srg(self):
url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C1"
body = {"ports": [{
- "port-name": "C1",
- "logical-connection-point": "SRG1-PP1",
- "port-type": "client",
- "circuit-id": "SRG1",
- "administrative-state": "inService",
- "port-qual": "roadm-external"}]}
+ "port-name": "C1",
+ "logical-connection-point": "SRG1-PP1",
+ "port-type": "client",
+ "circuit-id": "SRG1",
+ "administrative-state": "inService",
+ "port-qual": "roadm-external"}]}
response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
def test_28_change_status_line_port_roadma_deg(self):
url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
body = {"ports": [{
- "port-name": "L1",
- "logical-connection-point": "DEG2-TTP-TXRX",
- "port-type": "LINE",
- "circuit-id": "1",
- "administrative-state": "outOfService",
- "port-qual": "roadm-external"}]}
+ "port-name": "L1",
+ "logical-connection-point": "DEG2-TTP-TXRX",
+ "port-type": "LINE",
+ "circuit-id": "1",
+ "administrative-state": "outOfService",
+ "port-qual": "roadm-external"}]}
response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
def test_31_restore_status_line_port_roadma_srg(self):
url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/2%2F0/ports/L1"
body = {"ports": [{
- "port-name": "L1",
- "logical-connection-point": "DEG2-TTP-TXRX",
- "port-type": "LINE",
- "circuit-id": "1",
- "administrative-state": "inService",
- "port-qual": "roadm-external"}]}
+ "port-name": "L1",
+ "logical-connection-point": "DEG2-TTP-TXRX",
+ "port-type": "LINE",
+ "circuit-id": "1",
+ "administrative-state": "inService",
+ "port-qual": "roadm-external"}]}
response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
def test_35_change_status_line_port_xpdrc(self):
url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
body = {"ports": [{
- "port-name": "1",
- "port-type": "CFP2",
- "administrative-state": "outOfService",
- "port-qual": "xpdr-network"}]}
+ "port-name": "1",
+ "port-type": "CFP2",
+ "administrative-state": "outOfService",
+ "port-qual": "xpdr-network"}]}
response = requests.request("PUT", url.format("http://127.0.0.1:8154/restconf"),
data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
def test_38_restore_status_line_port_xpdrc(self):
url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/1%2F0%2F1-PLUG-NET/ports/1"
body = {"ports": [{
- "port-name": "1",
- "port-type": "CFP2",
- "administrative-state": "inService",
- "port-qual": "xpdr-network"}]}
+ "port-name": "1",
+ "port-type": "CFP2",
+ "administrative-state": "inService",
+ "port-qual": "xpdr-network"}]}
response = requests.request("PUT", url.format("http://127.0.0.1:8154/restconf"),
data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
def test_42_change_status_port_roadma_srg(self):
url = "{}/config/org-openroadm-device:org-openroadm-device/circuit-packs/3%2F0/ports/C2"
body = {"ports": [{
- "port-name": "C2",
- "logical-connection-point": "SRG1-PP2",
- "port-type": "client",
- "circuit-id": "SRG1",
- "administrative-state": "outOfService",
- "port-qual": "roadm-external"}]}
+ "port-name": "C2",
+ "logical-connection-point": "SRG1-PP2",
+ "port-type": "client",
+ "circuit-id": "SRG1",
+ "administrative-state": "outOfService",
+ "port-qual": "roadm-external"}]}
response = requests.request("PUT", url.format("http://127.0.0.1:8141/restconf"),
data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
sys.path.append('transportpce_tests/common/')
import test_utils
-
class TransportPCEtesting(unittest.TestCase):
processes = None
self.assertDictEqual(
dict({u'frequency': 196.0812, u'otsi-rate': u'org-openroadm-common-optical-channel-types:R400G-otsi',
- u'transmit-power': -5, u'modulation-format': 'dp-qam16'},
- **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']),
+ u'transmit-power': -5, u'modulation-format': 'dp-qam16'},
+ **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']),
res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi'])
def test_15_check_interface_OTSI_GROUP_xpdra2(self):
self.assertDictEqual(
dict({u'frequency': 196.0812, u'otsi-rate': u'org-openroadm-common-optical-channel-types:R400G-otsi',
- u'transmit-power': -5, u'modulation-format': 'dp-qam16'},
- **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']),
+ u'transmit-power': -5, u'modulation-format': 'dp-qam16'},
+ **res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi']),
res['interface'][0]['org-openroadm-optical-tributary-signal-interfaces:otsi'])
def test_18_check_interface_OTSI_GROUP_xpdrc2(self):
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {'name': 'XPDR2-CLIENT1-ETHERNET100G',
- 'administrative-state': 'inService',
- 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT',
- 'type': 'org-openroadm-interfaces:ethernetCsmacd',
- 'supporting-port': 'C1'
- }
+ 'administrative-state': 'inService',
+ 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT',
+ 'type': 'org-openroadm-interfaces:ethernetCsmacd',
+ 'supporting-port': 'C1'
+ }
input_dict_2 = {u'speed': 100000}
self.assertDictEqual(dict(input_dict_1, **res['interface'][0]),
res['interface'][0])
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {'name': 'XPDR2-CLIENT1-ETHERNET100G',
- 'administrative-state': 'inService',
- 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT',
- 'type': 'org-openroadm-interfaces:ethernetCsmacd',
- 'supporting-port': 'C1'
- }
+ 'administrative-state': 'inService',
+ 'supporting-circuit-pack-name': '1/2/1/1-PLUG-CLIENT',
+ 'type': 'org-openroadm-interfaces:ethernetCsmacd',
+ 'supporting-port': 'C1'
+ }
input_dict_2 = {u'speed': 100000}
self.assertDictEqual(dict(input_dict_1, **res['interface'][0]),
res['interface'][0])
self.assertDictEqual(dict(input_dict_2, **res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']),
- res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
+ res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
def test_36_check_interface_ODU4_CLIENT_xpdrc2(self):
response = test_utils.check_netconf_node_request(
for tp in tpList:
if tp['tp-id'] == 'XPDR2-NETWORK1':
self.assertNotIn('org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes',
- dict.keys(tp))
+ dict.keys(tp))
def test_60_delete_OTUC4_service(self):
response = test_utils.service_delete_request("service1-OTUC4")
time.sleep(2)
def test_74_check_xc1_roadma(self):
- response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-755:768")
+ response = test_utils.check_netconf_node_request(
+ "ROADM-A1", "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-755:768")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
freq_map_array = [int(x) for x in freq_map]
self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
if ele['tp-id'] == 'DEG2-TTP-TXRX':
- freq_map = base64.b64decode(
- ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
- freq_map_array = [int(x) for x in freq_map]
- self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
+ freq_map = base64.b64decode(
+ ele['org-openroadm-network-topology:tx-ttp-attributes']['avail-freq-maps'][0]['freq-map'])
+ freq_map_array = [int(x) for x in freq_map]
+ self.assertEqual(freq_map_array[95], 0, "Index 1 should not be available")
time.sleep(3)
-
def test_78_check_interface_100GE_CLIENT_xpdra2(self):
response = test_utils.check_netconf_node_request(
"XPDR-A2", "interface/XPDR1-CLIENT1-ETHERNET")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {'name': 'XPDR1-CLIENT1-ETHERNET',
- 'administrative-state': 'inService',
- 'supporting-circuit-pack-name': '1/1/1-PLUG-CLIENT',
- 'type': 'org-openroadm-interfaces:ethernetCsmacd',
- 'supporting-port': 'C1'
- }
+ 'administrative-state': 'inService',
+ 'supporting-circuit-pack-name': '1/1/1-PLUG-CLIENT',
+ 'type': 'org-openroadm-interfaces:ethernetCsmacd',
+ 'supporting-port': 'C1'
+ }
input_dict_2 = {u'speed': 400000}
self.assertDictEqual(dict(input_dict_1, **res['interface'][0]),
res['interface'][0])
sys.path.append('transportpce_tests/common/')
import test_utils
-
class TransportPCEtesting(unittest.TestCase):
simple_topo_bi_dir_data = None
with open(TOPO_UNI_DIR_COMPLEX_FILE, 'r') as topo_uni_dir_complex:
cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read()
PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
- "..", "..", "sample_configs", "pce_portmapping_121.json")
+ "..", "..", "sample_configs", "pce_portmapping_121.json")
with open(PORT_MAPPING_FILE, 'r') as port_mapping:
cls.port_mapping_data = port_mapping.read()
sample_files_parsed = True
nbElmPath = len(res['output']['response-parameters']['path-description']
['aToZ-direction']['aToZ'])
self.assertEqual(31, nbElmPath)
- link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2", "state":"inService"}
+ link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2", "state": "inService"}
find = False
for i in range(0, nbElmPath):
resource_i = (res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
nbElmPath = len(res['output']['response-parameters']['path-description']
['aToZ-direction']['aToZ'])
self.assertEqual(47, nbElmPath)
- link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2", "state":"inService"}
+ link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2", "state": "inService"}
find = False
for i in range(0, nbElmPath):
resource_i = (res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
sys.path.append('transportpce_tests/common/')
import test_utils
-
class TransportPCE400Gtesting(unittest.TestCase):
simple_topo_bi_dir_data = None
cls.topo_bi_dir_data = topo_bi_dir.read()
OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
- "..", "..", "sample_configs", "honeynode-otntopo400G.json")
+ "..", "..", "sample_configs", "honeynode-otntopo400G.json")
with open(OTN_TOPO_BI_DIR_FILE, 'r') as otn_topo_bi_dir:
cls.otn_topo_bi_dir_data = otn_topo_bi_dir.read()
OTUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
- "..", "..", "sample_configs", "honeynode-otntopo400GwithOTUC4.json")
+ "..", "..", "sample_configs", "honeynode-otntopo400GwithOTUC4.json")
with open(OTUC4_OTN_TOPO_BI_DIR_FILE, 'r') as otuc4_otn_topo_bi_dir:
cls.otuc4_otn_topo_bi_dir_data = otuc4_otn_topo_bi_dir.read()
ODUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
- "..", "..", "sample_configs", "honeynode-otntopo400GwithODUC4.json")
+ "..", "..", "sample_configs", "honeynode-otntopo400GwithODUC4.json")
with open(ODUC4_OTN_TOPO_BI_DIR_FILE, 'r') as oduc4_otn_topo_bi_dir:
cls.oduc4_otn_topo_bi_dir_data = oduc4_otn_topo_bi_dir.read()
PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
- "..", "..", "sample_configs", "pce_portmapping_71.json")
+ "..", "..", "sample_configs", "pce_portmapping_71.json")
with open(PORT_MAPPING_FILE, 'r') as port_mapping:
cls.port_mapping_data = port_mapping.read()
sample_files_parsed = True
{"service-rate": "400", "clli": "NodeA",
"service-format": "OTU", "node-id": "XPDR-A2",
"rx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2"}}
- },
+ },
{"service-rate": "400", "clli": "NodeC",
"service-format": "OTU", "node-id": "XPDR-C2",
"rx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2"}}
- })
+ })
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
{"service-rate": "400", "clli": "NodeA", "service-format": "ODU",
"node-id": "XPDR-A2",
"tx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2"}}
- },
+ },
{"service-rate": "400", "clli": "NodeC", "service-format": "ODU",
"node-id": "XPDR-C2",
"tx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2"}}
- })
+ })
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
{"service-rate": "100", "clli": "NodeA", "service-format": "Ethernet",
"node-id": "XPDR-A2",
"tx-direction": {"port": {"port-device-name": "XPDR-A2-XPDR2",
- "port-name": "XPDR2-CLIENT1"}}},
+ "port-name": "XPDR2-CLIENT1"}}},
{"service-rate": "100", "clli": "NodeC", "service-format": "Ethernet",
"node-id": "XPDR-C2",
"tx-direction": {"port": {"port-device-name": "XPDR-C2-XPDR2",
- "port-name": "XPDR2-CLIENT1"}}})
+ "port-name": "XPDR2-CLIENT1"}}})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
['zToA-direction']['modulation-format'])
time.sleep(5)
+
if __name__ == "__main__":
unittest.main(verbosity=2)
sys.path.append('transportpce_tests/common/')
import test_utils
-
class TransportGNPYtesting(unittest.TestCase):
topo_cllinet_data = None
with open(TOPO_ORDTOPO_FILE, 'r') as topo_ordtopo:
cls.topo_ordtopo_data = topo_ordtopo.read()
PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
- "..", "..", "sample_configs", "gnpy", "gnpy_portmapping_121.json")
+ "..", "..", "sample_configs", "gnpy", "gnpy_portmapping_121.json")
with open(PORT_MAPPING_FILE, 'r') as port_mapping:
cls.port_mapping_data = port_mapping.read()
sample_files_parsed = True
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(2)
+
if __name__ == "__main__":
# logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
unittest.main(verbosity=2)
sys.path.append('transportpce_tests/common/')
import test_utils
-
class TransportNbiNotificationstesting(unittest.TestCase):
processes = None
cr_serv_sample_data = {"input": {