@classmethod
def setUpClass(cls):
cls.processes = test_utils.start_tpce()
- cls.processes = test_utils.start_sims(['spdrav2'])
+ cls.processes = test_utils.start_sims(['spdra'])
@classmethod
def tearDownClass(cls):
time.sleep(5)
def test_01_connect_SPDR_SA1(self):
- response = test_utils.mount_device("SPDR-SA1", 'spdrav2')
+ response = test_utils.mount_device("SPDR-SA1", 'spdra')
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
time.sleep(10)
- url = ("{}/operational/network-topology:"
- "network-topology/topology/topology-netconf/node/SPDR-SA1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/network-topology:network-topology/topology/topology-netconf/node/SPDR-SA1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
'connected')
def test_02_get_portmapping_CLIENT1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/SPDR-SA1/mapping/XPDR1-CLIENT1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-CLIENT1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_03_get_portmapping_NETWORK1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_04_service_path_create_OCH_OTU4(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {"renderer:input": {
"service-name": "service_ODU4",
"wave-number": "1",
"nodes": [
{"node-id": "SPDR-SA1",
"dest-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
'och-interface-id': ['XPDR1-NETWORK1-1']}, res["output"]['node-interface'])
def test_05_get_portmapping_NETWORK1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
- input_dict = {'name': 'XPDR1-NETWORK1-1',
- 'administrative-state': 'inService',
- 'supporting-circuit-pack-name': 'CP1-CFP0',
- 'type': 'org-openroadm-interfaces:opticalChannel',
- 'supporting-port': 'CP1-CFP0-P1'
- }
- # assertDictContainsSubset is deprecated
- '''
- self.assertDictContainsSubset({'name': 'XPDR1-NETWORK1-1', 'administrative-state': 'inService',
- 'supporting-circuit-pack-name': 'CP1-CFP0',
- 'type': 'org-openroadm-interfaces:opticalChannel',
- 'supporting-port': 'CP1-CFP0-P1'}, res['interface'][0])
- '''
- self.assertDictEqual(dict({'name': 'XPDR1-NETWORK1-1',
- 'administrative-state': 'inService',
- 'supporting-circuit-pack-name': 'CP1-CFP0',
- 'type': 'org-openroadm-interfaces:opticalChannel',
- 'supporting-port': 'CP1-CFP0-P1'
- } ,**res['interface'][0]),
+ self.assertDictEqual(dict(res['interface'][0], **{'name': 'XPDR1-NETWORK1-1',
+ 'administrative-state': 'inService',
+ 'supporting-circuit-pack-name': 'CP1-CFP0',
+ 'type': 'org-openroadm-interfaces:opticalChannel',
+ 'supporting-port': 'CP1-CFP0-P1'
+ }),
res['interface'][0])
self.assertDictEqual(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-OTU"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {'name': 'XPDR1-NETWORK1-OTU',
'fec': 'scfec'
}
- self.assertDictEqual(dict(input_dict_1, **res['interface'][0]),
+ self.assertDictEqual(dict(res['interface'][0], **input_dict_1),
res['interface'][0])
self.assertDictEqual(input_dict_2,
- res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
+ res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
def test_08_otn_service_path_create_ODU4(self):
- url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:otn-service-path"
data = {"renderer:input": {
"service-name": "service_ODU4",
"operation": "create",
"nodes": [
{"node-id": "SPDR-SA1",
"network-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
'odu-interface-id': ['XPDR1-NETWORK1-ODU4']}, res["output"]['node-interface'])
def test_09_get_portmapping_NETWORK1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU4"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU4', 'administrative-state': 'inService',
'type': 'org-openroadm-interfaces:otnOdu',
'supporting-port': 'CP1-CFP0-P1'}
input_dict_2 = {'odu-function': 'org-openroadm-otn-common-types:ODU-TTP',
- 'rate': 'org-openroadm-otn-common-types:ODU4'}
+ 'rate': 'org-openroadm-otn-common-types:ODU4',
+ 'expected-dapi': 'Swfw02qXGyI=',
+ 'expected-sapi': 'Swfw02qXGyI=',
+ 'tx-dapi': 'Swfw02qXGyI=',
+ 'tx-sapi': 'Swfw02qXGyI='}
- self.assertDictEqual(dict(input_dict_1, **res['interface'][0]),
+ self.assertDictEqual(dict(res['interface'][0], **input_dict_1),
res['interface'][0])
- self.assertDictEqual(dict(input_dict_2,
- **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']
+ self.assertDictEqual(dict(res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'],
+ **input_dict_2
),
- res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']
+ res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']
)
self.assertDictEqual(
{u'payload-type': u'21', u'exp-payload-type': u'21'},
res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
def test_11_otn_service_path_create_10GE(self):
- url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:otn-service-path"
data = {"renderer:input": {
"service-name": "service1",
"operation": "create",
{"node-id": "SPDR-SA1",
"client-tp": "XPDR1-CLIENT1",
"network-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ETHERNET10G"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict = {'name': 'XPDR1-CLIENT1-ETHERNET10G',
'type': 'org-openroadm-interfaces:ethernetCsmacd',
'supporting-port': 'CP1-SFP4-P1'
}
- self.assertDictEqual(dict(input_dict, **res['interface'][0]),
+ self.assertDictEqual(dict(res['interface'][0], **input_dict),
res['interface'][0])
self.assertDictEqual(
{u'speed': 10000},
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
'rate': 'org-openroadm-otn-common-types:ODU2e',
'monitoring-mode': 'terminated'}
- self.assertDictEqual(dict(input_dict_1, **res['interface'][0]),
+ self.assertDictEqual(dict(res['interface'][0], **input_dict_1),
res['interface'][0])
- self.assertDictEqual(dict(input_dict_2,
- **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']),
+ self.assertDictEqual(dict(res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'],
+ **input_dict_2),
res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
self.assertDictEqual(
{u'payload-type': u'03', u'exp-payload-type': u'03'},
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU2e-service1', 'administrative-state': 'inService',
input_dict_3 = {'trib-port-number': 1}
- self.assertDictEqual(dict(input_dict_1, **res['interface'][0]),
+ self.assertDictEqual(dict(res['interface'][0], **input_dict_1),
res['interface'][0])
- self.assertDictEqual(dict(input_dict_2,
- **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']),
+ self.assertDictEqual(dict(res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'],
+ **input_dict_2),
res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
- self.assertDictEqual(dict(input_dict_3,
- **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][
- 'parent-odu-allocation']),
- res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][
- 'parent-odu-allocation'])
+ self.assertDictEqual(dict(res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][
+ 'parent-odu-allocation'], **input_dict_3
+ ),
+ res['interface'][0]['org-openroadm-otn-odu-interfaces:odu'][
+ 'parent-odu-allocation'])
self.assertIn(1,
res['interface'][0][
'org-openroadm-otn-odu-interfaces:odu'][
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {
'direction': 'bidirectional'
}
- self.assertDictEqual(dict(input_dict_1, **res['odu-connection'][0]),
+ self.assertDictEqual(dict(res['odu-connection'][0], **input_dict_1),
res['odu-connection'][0])
self.assertDictEqual({u'dst-if': u'XPDR1-NETWORK1-ODU2e-service1'},
res['odu-connection'][0]['destination'])
res['odu-connection'][0]['source'])
def test_16_otn_service_path_delete_10GE(self):
- url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:otn-service-path"
data = {"renderer:input": {
"service-name": "service1",
"operation": "delete",
{"node-id": "SPDR-SA1",
"client-tp": "XPDR1-CLIENT1",
"network-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_18_check_no_interface_ODU2E_NETWORK(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_19_check_no_interface_ODU2E_CLIENT(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_20_check_no_interface_10GE_CLIENT(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ETHERNET10G"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_21_otn_service_path_delete_ODU4(self):
- url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:otn-service-path"
data = {"renderer:input": {
"service-name": "service_ODU4",
"operation": "delete",
"nodes": [
{"node-id": "SPDR-SA1",
"network-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU4"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_23_service_path_delete_OCH_OTU4(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {"renderer:input": {
"service-name": "service_OTU4",
"wave-number": "1",
"nodes": [
{"node-id": "SPDR-SA1",
"dest-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-OTU"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_25_check_no_interface_OCH(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_26_disconnect_SPDR_SA1(self):