def test_46_get_no_oc_services(self):
print("start test")
response = test_utils.get_service_list_request("")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{
# Test deleted topology
def test_07_test_topology_simple_bi_deleted(self):
response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
- self.assertEqual(response.status_code, 404)
+ self.assertEqual(response.status_code, requests.codes.conflict)
time.sleep(1)
# Load simple bidirectional topology
# Test deleted topology
def test_14_test_topology_simple_deleted(self):
response = test_utils.get_ordm_topo_request("node/XPONDER-1-2")
- self.assertEqual(response.status_code, 404)
+ self.assertEqual(response.status_code, requests.codes.conflict)
time.sleep(1)
# Load complex topology
# Test deleted complex topology
def test_25_test_topology_complex_deleted(self):
response = test_utils.get_ordm_topo_request("node/XPONDER-3-2")
- self.assertEqual(response.status_code, 404)
+ self.assertEqual(response.status_code, requests.codes.conflict)
time.sleep(1)
'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLYy'},
res['mapping'])
+ def test_15_xpdr_portmapping_CLIENT4(self):
response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT4")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
def test_17_xpdr_device_disconnected(self):
response = test_utils.get_netconf_oper_request("XPDRA01")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_18_xpdr_device_not_connected(self):
response = test_utils.portmapping_request("XPDRA01")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_20_rdm_device_disconnected(self):
response = test_utils.get_netconf_oper_request("ROADMA01")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_21_rdm_device_not_connected(self):
response = test_utils.portmapping_request("ROADMA01")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_15_service_path_delete_rdm_check(self):
response = test_utils.check_netconf_node_request("ROADMA01", "interface/DEG1-TTP-TXRX-7")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_16_service_path_delete_rdm_check(self):
response = test_utils.check_netconf_node_request("ROADMA01", "interface/SRG1-PP7-TXRX-7")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
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-7")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_18_service_path_delete_xpdr_check(self):
response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-7")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_19_service_path_delete_xpdr_check(self):
response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-OTU")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_20_service_path_delete_xpdr_check(self):
response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-ODU")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_21_service_path_delete_xpdr_check(self):
response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-CLIENT1-ETHERNET")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_46_get_no_oc_services(self):
print("start test")
response = test_utils.get_service_list_request("")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_18_check_no_interface_ODU4_spdra(self):
response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU4")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_44_check_no_interface_ODU2E_NETWORK_spdra(self):
response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU2e-service1")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_45_check_no_interface_ODU2E_CLIENT_spdra(self):
response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ODU2e-service1")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_46_check_no_interface_10GE_CLIENT_spdra(self):
response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ETHERNET10G")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_47_check_otn_topo_links(self):
response = test_utils.get_otn_topo_request()
def test_51_check_no_interface_ODU4_spdra(self):
response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU4")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_52_check_otn_topo_links(self):
self.test_22_check_otn_topo_otu4_links()
def test_55_get_no_service(self):
response = test_utils.get_service_list_request("")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_56_check_no_interface_OTU4_spdra(self):
response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-OTU")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_57_check_no_interface_OCH_spdra(self):
response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-1")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_58_getLinks_OtnTopology(self):
response = test_utils.get_otn_topo_request()
res['node'][0]['netconf-node-topology:connection-status'],
'connected')
- def test_02_get_portmapping_CLIENT1(self):
- response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-CLIENT1")
+ def test_02_get_portmapping_CLIENT4(self):
+ response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-CLIENT4")
self.assertEqual(response.status_code, requests.codes.ok)
- res_mapping = (response.json())['mapping'][0]
- self.assertSetEqual({'org-openroadm-port-types:if-10GE-ODU2e',
- 'org-openroadm-port-types:if-10GE-ODU2',
- 'org-openroadm-port-types:if-10GE'},
- set(res_mapping['supported-interface-capability']))
- self.assertDictEqual(
- dict(
- {'supporting-port': 'CP1-SFP4-P1',
- 'supporting-circuit-pack-name': 'CP1-SFP4',
- 'logical-connection-point': 'XPDR1-CLIENT1',
- 'port-direction': 'bidirectional',
- 'port-qual': 'xpdr-client',
- 'lcp-hash-val': 'FqlcrxV7p30='},
- **res_mapping),
- res_mapping)
+ res = response.json()
+ self.assertEqual('CP1-SFP4-P1', res['mapping'][0]['supporting-port'])
+ self.assertEqual('CP1-SFP4', res['mapping'][0]['supporting-circuit-pack-name'])
+ self.assertEqual('XPDR1-CLIENT4', res['mapping'][0]['logical-connection-point'])
+ self.assertEqual('bidirectional', res['mapping'][0]['port-direction'])
+ self.assertEqual('xpdr-client', res['mapping'][0]['port-qual'])
+ self.assertEqual('FqlcrxV7p3g=', res['mapping'][0]['lcp-hash-val'])
+ self.assertIn('org-openroadm-port-types:if-10GE-ODU2e', res['mapping'][0]['supported-interface-capability'])
+ self.assertIn('org-openroadm-port-types:if-10GE-ODU2', res['mapping'][0]['supported-interface-capability'])
+ self.assertIn('org-openroadm-port-types:if-10GE', res['mapping'][0]['supported-interface-capability'])
def test_03_get_portmapping_NETWORK1(self):
response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-NETWORK1")
def test_11_otn_service_path_create_10GE(self):
response = test_utils.otn_service_path_request("create", "service1", "10G", "Ethernet",
- [{"node-id": "SPDR-SA1", "client-tp": "XPDR1-CLIENT1",
+ [{"node-id": "SPDR-SA1", "client-tp": "XPDR1-CLIENT4",
"network-tp": "XPDR1-NETWORK1"}],
{"ethernet-encoding": "eth encode", "trib-slot": ["1"], "trib-port-number": "1"})
time.sleep(3)
res = response.json()
self.assertIn('Otn Service path was set up successfully for node :SPDR-SA1', res["output"]["result"])
self.assertTrue(res["output"]["success"])
- self.assertIn(
- {'node-id': 'SPDR-SA1',
- 'connection-id': ['XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1'],
- 'odu-interface-id': ['XPDR1-NETWORK1-ODU2e-service1', 'XPDR1-CLIENT1-ODU2e-service1'],
- 'eth-interface-id': ['XPDR1-CLIENT1-ETHERNET10G']}, res["output"]['node-interface'])
+ self.assertEqual('SPDR-SA1', res["output"]['node-interface'][0]['node-id'])
+ self.assertIn('XPDR1-CLIENT4-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1',
+ res["output"]['node-interface'][0]['connection-id'])
+ self.assertIn('XPDR1-CLIENT4-ETHERNET10G', res["output"]['node-interface'][0]['eth-interface-id'])
+ self.assertIn('XPDR1-NETWORK1-ODU2e-service1', res["output"]['node-interface'][0]['odu-interface-id'])
+ self.assertIn('XPDR1-CLIENT4-ODU2e-service1', res["output"]['node-interface'][0]['odu-interface-id'])
def test_12_check_interface_10GE_CLIENT(self):
- response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ETHERNET10G")
+ response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT4-ETHERNET10G")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
- input_dict = {'name': 'XPDR1-CLIENT1-ETHERNET10G',
+ input_dict = {'name': 'XPDR1-CLIENT4-ETHERNET10G',
'administrative-state': 'inService',
'supporting-circuit-pack-name': 'CP1-SFP4',
'type': 'org-openroadm-interfaces:ethernetCsmacd',
res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
def test_13_check_interface_ODU2E_CLIENT(self):
- response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ODU2e-service1")
+ response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT4-ODU2e-service1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
- input_dict_1 = {'name': 'XPDR1-CLIENT1-ODU2e-service1',
+ input_dict_1 = {'name': 'XPDR1-CLIENT4-ODU2e-service1',
'administrative-state': 'inService',
'supporting-circuit-pack-name': 'CP1-SFP4',
- 'supporting-interface': 'XPDR1-CLIENT1-ETHERNET10G',
+ 'supporting-interface': 'XPDR1-CLIENT4-ETHERNET10G',
'type': 'org-openroadm-interfaces:otnOdu',
'supporting-port': 'CP1-SFP4-P1'}
input_dict_2 = {
def test_15_check_ODU2E_connection(self):
response = test_utils.check_netconf_node_request(
"SPDR-SA1",
- "odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1")
+ "odu-connection/XPDR1-CLIENT4-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {
'connection-name':
- 'XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1',
+ 'XPDR1-CLIENT4-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1',
'direction': 'bidirectional'
}
res['odu-connection'][0])
self.assertDictEqual({u'dst-if': u'XPDR1-NETWORK1-ODU2e-service1'},
res['odu-connection'][0]['destination'])
- self.assertDictEqual({u'src-if': u'XPDR1-CLIENT1-ODU2e-service1'},
+ self.assertDictEqual({u'src-if': u'XPDR1-CLIENT4-ODU2e-service1'},
res['odu-connection'][0]['source'])
def test_16_otn_service_path_delete_10GE(self):
response = test_utils.otn_service_path_request("delete", "service1", "10G", "Ethernet",
- [{"node-id": "SPDR-SA1", "client-tp": "XPDR1-CLIENT1",
+ [{"node-id": "SPDR-SA1", "client-tp": "XPDR1-CLIENT4",
"network-tp": "XPDR1-NETWORK1"}],
{"ethernet-encoding": "eth encode", "trib-slot": ["1"], "trib-port-number": "1"})
time.sleep(3)
def test_17_check_no_ODU2E_connection(self):
response = test_utils.check_netconf_node_request(
"SPDR-SA1",
- "odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ "odu-connection/XPDR1-CLIENT4-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1")
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_18_check_no_interface_ODU2E_NETWORK(self):
response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU2e-service1")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_19_check_no_interface_ODU2E_CLIENT(self):
- response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ODU2e-service1")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT4-ODU2e-service1")
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_20_check_no_interface_10GE_CLIENT(self):
- response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT1-ETHERNET10G")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-CLIENT4-ETHERNET10G")
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_21_otn_service_path_delete_ODU4(self):
response = test_utils.otn_service_path_request("delete", "service_ODU4", "100G", "ODU",
def test_22_check_no_interface_ODU4(self):
response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-ODU4")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_23_service_path_delete_OCH_OTU4(self):
response = test_utils.service_path_request("delete", "service_OCH_OTU4", "1",
def test_24_check_no_interface_OTU4(self):
response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-OTU")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_25_check_no_interface_OCH(self):
response = test_utils.check_netconf_node_request("SPDR-SA1", "interface/XPDR1-NETWORK1-1")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
def test_26_disconnect_SPDR_SA1(self):
response = test_utils.unmount_device("SPDR-SA1")
def test_16_xpdr_device_disconnected(self):
response = test_utils.get_netconf_oper_request("XPDR-A1")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_17_xpdr_device_not_connected(self):
response = test_utils.portmapping_request("XPDR-A1")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_19_rdm_device_disconnected(self):
response = test_utils.get_netconf_oper_request("ROADM-A1")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_20_rdm_device_not_connected(self):
response = test_utils.portmapping_request("ROADM-A1")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
# -mc supporting interfaces must not be created for SRG, only degrees
def test_09_service_path_create_rdm_check(self):
response = test_utils.check_netconf_node_request("ROADM-A1", "interface/SRG1-PP3-TXRX-mc-7")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_18_service_path_delete_rdm_check(self):
response = test_utils.check_netconf_node_request("ROADM-A1", "interface/DEG1-TTP-TXRX-mc-7")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_19_service_path_delete_rdm_check(self):
response = test_utils.check_netconf_node_request("ROADM-A1", "interface/DEG1-TTP-TXRX-nmc-7")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
def test_20_service_path_delete_rdm_check(self):
response = test_utils.check_netconf_node_request("ROADM-A1", "interface/SRG1-PP3-TXRX-mc-7")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn({
"error-type": "application",
def test_21_service_path_delete_rdm_check(self):
response = test_utils.check_netconf_node_request("ROADM-A1", "interface/SRG1-PP3-TXRX-nmc-7")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn({
"error-type": "application",
def test_22_service_path_delete_rdm_check(self):
response = test_utils.check_netconf_node_request("ROADM-A1", "interface/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn({
"error-type": "application",
def test_23_service_path_delete_xpdr_check(self):
response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-NETWORK1-7")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn({
"error-type": "application",
def test_24_service_path_delete_xpdr_check(self):
response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-NETWORK1-OTU")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn({
"error-type": "application",
def test_25_service_path_delete_xpdr_check(self):
response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-NETWORK1-ODU")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn({
"error-type": "application",
def test_26_service_path_delete_xpdr_check(self):
response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-CLIENT1-ETHERNET")
- self.assertEqual(response.status_code, requests.codes.not_found)
+ self.assertEqual(response.status_code, requests.codes.conflict)
res = response.json()
self.assertIn({
"error-type": "application",
@classmethod
def setUpClass(cls):
cls.init_failed = False
-
+ os.environ['JAVA_MIN_MEM'] = '1024M'
+ os.environ['JAVA_MAX_MEM'] = '4069M'
cls.processes = test_utils.start_tpce()
# TAPI feature is not installed by default in Karaf
if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':