<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath />
</parent>
<dependency>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>mdsal-artifacts</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<scope>import</scope>
<type>pom</type>
</dependency>
<parent>
<groupId>org.opendaylight.odlparent</groupId>
<artifactId>single-feature-parent</artifactId>
- <version>6.0.9</version>
+ <version>6.0.11</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.odlparent</groupId>
<artifactId>single-feature-parent</artifactId>
- <version>6.0.9</version>
+ <version>6.0.11</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.odlparent</groupId>
<artifactId>single-feature-parent</artifactId>
- <version>6.0.9</version>
+ <version>6.0.11</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.odlparent</groupId>
<artifactId>odlparent-lite</artifactId>
- <version>6.0.9</version>
+ <version>6.0.11</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath />
</parent>
<parent>
<groupId>org.opendaylight.odlparent</groupId>
<artifactId>karaf4-parent</artifactId>
- <version>6.0.9</version>
+ <version>6.0.11</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.odlparent</groupId>
<artifactId>bundle-parent</artifactId>
- <version>6.0.9</version>
+ <version>6.0.11</version>
<relativePath/>
</parent>
<dependency>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>mdsal-artifacts</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<scope>import</scope>
<type>pom</type>
</dependency>
.setTpnPool(tpnPool).build()));
tpBldr.addAugmentation(TerminationPoint1.class,
- new TerminationPoint1Builder().setXpdrTpPortConnectionAttributes(xtpcaBldr.build()).build());
+ new TerminationPoint1Builder(tp.augmentation(TerminationPoint1.class))
+ .setXpdrTpPortConnectionAttributes(xtpcaBldr.build()).build());
return tpBldr.build();
}
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath />
</parent>
<parent>
<groupId>org.opendaylight.odlparent</groupId>
<artifactId>odlparent</artifactId>
- <version>6.0.9</version>
+ <version>6.0.11</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath/>
</parent>
import org.opendaylight.yang.gen.v1.http.org.opendaylight.transportpce.portmapping.rev200429.network.nodes.Mapping;
import org.opendaylight.yang.gen.v1.http.org.openroadm.common.types.rev181019.FrequencyGHz;
import org.opendaylight.yang.gen.v1.http.org.openroadm.common.types.rev181019.FrequencyTHz;
+import org.opendaylight.yang.gen.v1.http.org.openroadm.common.types.rev181019.ModulationFormat;
import org.opendaylight.yang.gen.v1.http.org.openroadm.common.types.rev181019.PowerDBm;
import org.opendaylight.yang.gen.v1.http.org.openroadm.common.types.rev181019.R100G;
import org.opendaylight.yang.gen.v1.http.org.openroadm.device.rev181019.interfaces.grp.InterfaceBuilder;
OchBuilder ocIfBuilder = new OchBuilder()
.setFrequency(FrequencyTHz.getDefaultInstance(String.valueOf(fixedFlex.getCenterFrequency())))
.setRate(R100G.class)
+ // TODO: Check if this the correct way to set modulation format here
+ .setModulationFormat(ModulationFormat.DpQpsk)
.setTransmitPower(new PowerDBm(new BigDecimal("-5")));
// Create generic interface
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath/>
</parent>
<dependency>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>mdsal-artifacts</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<scope>import</scope>
<type>pom</type>
</dependency>
<parent>
<groupId>org.opendaylight.mdsal</groupId>
<artifactId>binding-parent</artifactId>
- <version>5.0.12</version>
+ <version>5.0.14</version>
<relativePath/>
</parent>
<parent>
<groupId>org.opendaylight.odlparent</groupId>
<artifactId>bundle-parent</artifactId>
- <version>6.0.9</version>
+ <version>6.0.11</version>
<relativePath/>
</parent>
# test service-create for Eth service from xpdr to xpdr
def test_11_create_eth_service1(self):
- url = "{}/operations/org-openroadm-service:service-create"
data = {
"input": {
"sdnc-request-header": {
"operator-contact": "pw1234"
}
}
- response = test_utils.post_request(url, data)
+ response = test_utils.service_create_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_12_get_eth_service1(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service1"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(2)
def test_22_create_eth_service2(self):
- url = "{}/operations/org-openroadm-service:service-create"
data = {
"input": {
"sdnc-request-header": {
"operator-contact": "pw1234"
}
}
- response = test_utils.post_request(url, data)
+ response = test_utils.service_create_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_23_get_eth_service2(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service2"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service2")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
# creation service test on a non-available resource
def test_28_create_eth_service3(self):
- url = "{}/operations/org-openroadm-service:service-create"
data = {
"input": {
"sdnc-request-header": {
"operator-contact": "pw1234"
}
}
- response = test_utils.post_request(url, data)
+ response = test_utils.service_create_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
# test service-create for Optical Channel (OC) service from srg-pp to srg-pp
def test_36_create_oc_service1(self):
- url = "{}/operations/org-openroadm-service:service-create"
data = {
"input": {
"sdnc-request-header": {
"operator-contact": "pw1234"
}
}
- response = test_utils.post_request(url, data)
+ response = test_utils.service_create_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_37_get_oc_service1(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service1"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(7)
def test_40_create_oc_service2(self):
- url = "{}/operations/org-openroadm-service:service-create"
data = {
"input": {
"sdnc-request-header": {
"operator-contact": "pw1234"
}
}
- response = test_utils.post_request(url, data)
+ response = test_utils.service_create_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_41_get_oc_service2(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service2"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service2")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
def test_46_get_no_oc_services(self):
print("start test")
- url = "{}/operational/org-openroadm-service:service-list"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("")
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
self.test_30_delete_eth_service1()
def test_50_loop_create_oc_service(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service1"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service1")
if response.status_code != 404:
url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
# Mount the different topologies
def test_01_connect_clliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.rawput_request(url, self.topo_cllinet_data)
+ response = test_utils.rawput_request(test_utils.URL_CONFIG_CLLI_NET, self.topo_cllinet_data)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
def test_02_connect_openroadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.rawput_request(url, self.topo_ordnet_data)
+ response = test_utils.rawput_request(test_utils.URL_CONFIG_ORDM_NET, self.topo_ordnet_data)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
time.sleep(3)
def test_09_disconnect_openroadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.delete_request(url)
+ response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_NET)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
def test_10_disconnect_clliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.delete_request(url)
+ response = test_utils.delete_request(test_utils.URL_CONFIG_CLLI_NET)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
# 'connected')
# def test_02_restconfAPI(self):
-# url = ("{}/config/transportpce-portmapping:network/nodes/controller-config")
-# response = requests.request(
-# "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+# response = test_utils.portmapping_request("controller-config")
# self.assertEqual(response.status_code, requests.codes.not_found)
# res = response.json()
# self.assertIn(
time.sleep(10)
def test_03_rdm_portmapping_info(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADMA01/node-info"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01/node-info")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(3)
def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADMA01/mapping/DEG1-TTP-TXRX"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01/mapping/DEG1-TTP-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_05_rdm_portmapping_SRG1_PP7_TXRX(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADMA01/mapping/SRG1-PP7-TXRX"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01/mapping/SRG1-PP7-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_06_rdm_portmapping_SRG3_PP1_TXRX(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADMA01/mapping/SRG3-PP1-TXRX"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01/mapping/SRG3-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
time.sleep(10)
def test_09_xpdr_portmapping_info(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/node-info"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/node-info")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(3)
def test_10_xpdr_portmapping_NETWORK1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-NETWORK1"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_11_xpdr_portmapping_NETWORK2(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-NETWORK2"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK2")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_12_xpdr_portmapping_CLIENT1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-CLIENT1"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_13_xpdr_portmapping_CLIENT2(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-CLIENT2"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT2")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_14_xpdr_portmapping_CLIENT3(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-CLIENT3"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT3")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLYy'},
res['mapping'])
- def test_15_xpdr_portmapping_CLIENT4(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-CLIENT4"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT4")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['errors']['error'])
def test_18_xpdr_device_not_connected(self):
- url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01")
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01")
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
res['errors']['error'])
def test_21_rdm_device_not_connected(self):
- url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA01")
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01")
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_03_rdm_portmapping(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADMA01"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['nodes'][0]['mapping'])
def test_04_xpdr_portmapping(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
for i in range(0, nbNode):
nodeId = resTopo['network'][0]['node'][i]['node-id']
nodeMapId = nodeId.split("-")[0]
- urlMapList = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId
- responseMapList = test_utils.get_request(urlMapList)
+ responseMapList = test_utils.portmapping_request(nodeMapId)
resMapList = responseMapList.json()
nbMappings = len(resMapList['nodes'][0]['mapping']) - nbMapCumul
nbTp = len(resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'])
for j in range(0, nbTp):
tpId = resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
if((not "CP" in tpId) and (not "CTP" in tpId)):
- urlMap = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId + "/mapping/" + tpId
- responseMap = test_utils.get_request(urlMap)
+ responseMap = test_utils.portmapping_request(nodeMapId+"/mapping/"+tpId)
# Disconnect the ROADMA
def test_03_disconnect_rdm(self):
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_02_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'NodeA')
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_03_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'ROADMA01')
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_07_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'NodeA')
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_08_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(response.status_code, requests.codes.created)
def test_16_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(len(listNode), 0)
def test_17_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- )
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(response.status_code, requests.codes.created)
def test_26_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
response = test_utils.unmount_device("ROADMB01")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
# Delete in the clli-network
- url = "{}/config/ietf-network:networks/network/clli-network/node/NodeB"
- data = {}
- response = test_utils.delete_request(url)
+ response = test_utils.del_node_request("NodeB")
self.assertEqual(response.status_code, requests.codes.ok)
def test_31_disconnect_ROADMC(self):
response = test_utils.unmount_device("ROADMC01")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
# Delete in the clli-network
- url = "{}/config/ietf-network:networks/network/clli-network/node/NodeC"
- data = {}
- response = test_utils.delete_request(url)
+ response = test_utils.del_node_request("NodeC")
self.assertEqual(response.status_code, requests.codes.ok)
def test_32_getNodes_OpenRoadmTopology(self):
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'ROADMC01-DEG2')
def test_33_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'ROADMB01')
def test_34_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
def test_36_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- )
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_37_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- )
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
response = test_utils.unmount_device("ROADMA01")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
# Delete in the clli-network
- url = "{}/config/ietf-network:networks/network/clli-network/node/NodeA"
- data = {}
- response = test_utils.delete_request(url)
+ response = test_utils.del_node_request("NodeA")
self.assertEqual(response.status_code, requests.codes.ok)
def test_42_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('node', res['network'][0])
def test_43_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('node', res['network'][0])
# test service-create for Eth service from xpdr to xpdr
def test_11_create_eth_service1(self):
- url = "{}/operations/org-openroadm-service:service-create"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
"operator-contact": "pw1234"
}
}
- response = test_utils.post_request(url, data)
+ response = test_utils.service_create_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_12_get_eth_service1(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service1"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(2)
def test_22_create_eth_service2(self):
- url = "{}/operations/org-openroadm-service:service-create"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
"operator-contact": "pw1234"
}
}
- response = test_utils.post_request(url, data)
+ response = test_utils.service_create_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_23_get_eth_service2(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service2"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service2")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
# creation service test on a non-available resource
def test_28_create_eth_service3(self):
- url = "{}/operations/org-openroadm-service:service-create"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
"operator-contact": "pw1234"
}
}
- response = test_utils.post_request(url, data)
+ response = test_utils.service_create_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
# test service-create for Optical Channel (OC) service from srg-pp to srg-pp
def test_36_create_oc_service1(self):
- url = "{}/operations/org-openroadm-service:service-create"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
"operator-contact": "pw1234"
}
}
- response = test_utils.post_request(url, data)
+ response = test_utils.service_create_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_37_get_oc_service1(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service1"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(7)
def test_40_create_oc_service2(self):
- url = "{}/operations/org-openroadm-service:service-create"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
"operator-contact": "pw1234"
}
}
- response = test_utils.post_request(url, data)
+ response = test_utils.service_create_request(data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_41_get_oc_service2(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service2"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service2")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
def test_46_get_no_oc_services(self):
print("start test")
- url = "{}/operational/org-openroadm-service:service-list"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("")
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
self.test_30_delete_eth_service1()
def test_50_loop_create_oc_service(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service1"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service1")
if response.status_code != 404:
url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
processes = None
WAITING = 20 # nominal value is 300
+ cr_serv_sample_data = {"input": {
+ "sdnc-request-header": {
+ "request-id": "request-1",
+ "rpc-action": "service-create",
+ "request-system-id": "appname"
+ },
+ "service-name": "service1-OCH-OTU4",
+ "common-id": "commonId",
+ "connection-type": "infrastructure",
+ "service-a-end": {
+ "service-rate": "100",
+ "node-id": "SPDR-SA1",
+ "service-format": "OTU",
+ "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
+ "clli": "NodeSA",
+ "subrate-eth-sla": {
+ "subrate-eth-sla": {
+ "committed-info-rate": "100000",
+ "committed-burst-size": "64"
+ }
+ },
+ "tx-direction": {
+ "port": {
+ "port-device-name": "SPDR-SA1-XPDR1",
+ "port-type": "fixed",
+ "port-name": "XPDR1-NETWORK1",
+ "port-rack": "000000.00",
+ "port-shelf": "Chassis#1"
+ },
+ "lgx": {
+ "lgx-device-name": "Some lgx-device-name",
+ "lgx-port-name": "Some lgx-port-name",
+ "lgx-port-rack": "000000.00",
+ "lgx-port-shelf": "00"
+ }
+ },
+ "rx-direction": {
+ "port": {
+ "port-device-name": "SPDR-SA1-XPDR1",
+ "port-type": "fixed",
+ "port-name": "XPDR1-NETWORK1",
+ "port-rack": "000000.00",
+ "port-shelf": "Chassis#1"
+ },
+ "lgx": {
+ "lgx-device-name": "Some lgx-device-name",
+ "lgx-port-name": "Some lgx-port-name",
+ "lgx-port-rack": "000000.00",
+ "lgx-port-shelf": "00"
+ }
+ },
+ "optic-type": "gray"
+ },
+ "service-z-end": {
+ "service-rate": "100",
+ "node-id": "SPDR-SC1",
+ "service-format": "OTU",
+ "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
+ "clli": "NodeSC",
+ "subrate-eth-sla": {
+ "subrate-eth-sla": {
+ "committed-info-rate": "100000",
+ "committed-burst-size": "64"
+ }
+ },
+ "tx-direction": {
+ "port": {
+ "port-device-name": "SPDR-SC1-XPDR1",
+ "port-type": "fixed",
+ "port-name": "XPDR1-NETWORK1",
+ "port-rack": "000000.00",
+ "port-shelf": "Chassis#1"
+ },
+ "lgx": {
+ "lgx-device-name": "Some lgx-device-name",
+ "lgx-port-name": "Some lgx-port-name",
+ "lgx-port-rack": "000000.00",
+ "lgx-port-shelf": "00"
+ }
+ },
+ "rx-direction": {
+ "port": {
+ "port-device-name": "SPDR-SC1-XPDR1",
+ "port-type": "fixed",
+ "port-name": "XPDR1-NETWORK1",
+ "port-rack": "000000.00",
+ "port-shelf": "Chassis#1"
+ },
+ "lgx": {
+ "lgx-device-name": "Some lgx-device-name",
+ "lgx-port-name": "Some lgx-port-name",
+ "lgx-port-rack": "000000.00",
+ "lgx-port-shelf": "00"
+ }
+ },
+ "optic-type": "gray"
+ },
+ "due-date": "2018-06-15T00:00:01Z",
+ "operator-contact": "pw1234"
+ }
+ }
+
+
+
@classmethod
def setUpClass(cls):
cls.processes = test_utils.start_tpce()
# test service-create for OCH-OTU4 service from spdr to spdr
def test_11_check_otn_topology(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(nbNode, 4)
self.assertNotIn('ietf-network-topology:link', res['network'][0])
+
def test_12_create_OCH_OTU4_service(self):
- url = "{}/operations/org-openroadm-service:service-create"
- data = {"input": {
- "sdnc-request-header": {
- "request-id": "request-1",
- "rpc-action": "service-create",
- "request-system-id": "appname"
- },
- "service-name": "service1-OCH-OTU4",
- "common-id": "commonId",
- "connection-type": "infrastructure",
- "service-a-end": {
- "service-rate": "100",
- "node-id": "SPDR-SA1",
- "service-format": "OTU",
- "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
- "clli": "NodeSA",
- "subrate-eth-sla": {
- "subrate-eth-sla": {
- "committed-info-rate": "100000",
- "committed-burst-size": "64"
- }
- },
- "tx-direction": {
- "port": {
- "port-device-name": "SPDR-SA1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-NETWORK1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "rx-direction": {
- "port": {
- "port-device-name": "SPDR-SA1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-NETWORK1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "optic-type": "gray"
- },
- "service-z-end": {
- "service-rate": "100",
- "node-id": "SPDR-SC1",
- "service-format": "OTU",
- "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
- "clli": "NodeSC",
- "subrate-eth-sla": {
- "subrate-eth-sla": {
- "committed-info-rate": "100000",
- "committed-burst-size": "64"
- }
- },
- "tx-direction": {
- "port": {
- "port-device-name": "SPDR-SC1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-NETWORK1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "rx-direction": {
- "port": {
- "port-device-name": "SPDR-SC1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-NETWORK1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "optic-type": "gray"
- },
- "due-date": "2018-06-15T00:00:01Z",
- "operator-contact": "pw1234"
- }
- }
- response = test_utils.post_request(url, data)
+ response = test_utils.service_create_request(self.cr_serv_sample_data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_13_get_OCH_OTU4_service1(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service1-OCH-OTU4"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service1-OCH-OTU4")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
self.assertDictEqual(
{u'frequency': 196.1, u'rate': u'org-openroadm-common-types:R100G',
- u'transmit-power': -5},
+ u'transmit-power': -5, u'modulation-format': 'dp-qpsk'},
res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
def test_15_check_interface_OTU4_spdra(self):
self.assertDictEqual(
{u'frequency': 196.1, u'rate': u'org-openroadm-common-types:R100G',
- u'transmit-power': -5},
+ u'transmit-power': -5, u'modulation-format': 'dp-qpsk'},
res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
def test_17_check_interface_OTU4_spdrc(self):
time.sleep(3)
def test_22_check_otn_topo_otu4_links(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nb_links = len(res['network'][0]['ietf-network-topology:link'])
# test service-create for ODU4 service from spdr to spdr
def test_23_create_ODU4_service(self):
- url = "{}/operations/org-openroadm-service:service-create"
- data = {"input": {
- "sdnc-request-header": {
- "request-id": "request-1",
- "rpc-action": "service-create",
- "request-system-id": "appname"
- },
- "service-name": "service1-ODU4",
- "common-id": "commonId",
- "connection-type": "infrastructure",
- "service-a-end": {
- "service-rate": "100",
- "node-id": "SPDR-SA1",
- "service-format": "ODU",
- "odu-service-rate": "org-openroadm-otn-common-types:ODU4",
- "clli": "NodeSA",
- "subrate-eth-sla": {
- "subrate-eth-sla": {
- "committed-info-rate": "100000",
- "committed-burst-size": "64"
- }
- },
- "tx-direction": {
- "port": {
- "port-device-name": "SPDR-SA1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-NETWORK1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "rx-direction": {
- "port": {
- "port-device-name": "SPDR-SA1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-NETWORK1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "optic-type": "gray"
- },
- "service-z-end": {
- "service-rate": "100",
- "node-id": "SPDR-SC1",
- "service-format": "ODU",
- "odu-service-rate": "org-openroadm-otn-common-types:ODU4",
- "clli": "NodeSC",
- "subrate-eth-sla": {
- "subrate-eth-sla": {
- "committed-info-rate": "100000",
- "committed-burst-size": "64"
- }
- },
- "tx-direction": {
- "port": {
- "port-device-name": "SPDR-SC1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-NETWORK1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "rx-direction": {
- "port": {
- "port-device-name": "SPDR-SC1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-NETWORK1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "optic-type": "gray"
- },
- "due-date": "2018-06-15T00:00:01Z",
- "operator-contact": "pw1234"
- }
- }
- response = test_utils.post_request(url, data)
+ self.cr_serv_sample_data["input"]["service-name"] = "service1-ODU4"
+ self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "ODU"
+ del self.cr_serv_sample_data["input"]["service-a-end"]["otu-service-rate"]
+ self.cr_serv_sample_data["input"]["service-a-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
+ self.cr_serv_sample_data["input"]["service-z-end"]["service-format"] = "ODU"
+ del self.cr_serv_sample_data["input"]["service-z-end"]["otu-service-rate"]
+ self.cr_serv_sample_data["input"]["service-z-end"]["odu-service-rate"] = "org-openroadm-otn-common-types:ODU4"
+
+ response = test_utils.service_create_request(self.cr_serv_sample_data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_24_get_ODU4_service1(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service1-ODU4"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service1-ODU4")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
def test_27_check_otn_topo_links(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nb_links = len(res['network'][0]['ietf-network-topology:link'])
self.fail("this link should not exist")
def test_28_check_otn_topo_tp(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
if (node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1'):
# test service-create for 10GE service from spdr to spdr
def test_29_create_10GE_service(self):
- url = "{}/operations/org-openroadm-service:service-create"
- data = {"input": {
- "sdnc-request-header": {
- "request-id": "request-1",
- "rpc-action": "service-create",
- "request-system-id": "appname"
- },
- "service-name": "service1-10GE",
- "common-id": "commonId",
- "connection-type": "service",
- "service-a-end": {
- "service-rate": "10",
- "node-id": "SPDR-SA1",
- "service-format": "Ethernet",
- "clli": "NodeSA",
- "subrate-eth-sla": {
- "subrate-eth-sla": {
- "committed-info-rate": "100000",
- "committed-burst-size": "64"
- }
- },
- "tx-direction": {
- "port": {
- "port-device-name": "SPDR-SA1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-CLIENT1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "rx-direction": {
- "port": {
- "port-device-name": "SPDR-SA1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-CLIENT1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "optic-type": "gray"
- },
- "service-z-end": {
- "service-rate": "10",
- "node-id": "SPDR-SC1",
- "service-format": "Ethernet",
- "clli": "NodeSC",
- "subrate-eth-sla": {
- "subrate-eth-sla": {
- "committed-info-rate": "100000",
- "committed-burst-size": "64"
- }
- },
- "tx-direction": {
- "port": {
- "port-device-name": "SPDR-SC1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-CLIENT1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "rx-direction": {
- "port": {
- "port-device-name": "SPDR-SC1-XPDR1",
- "port-type": "fixed",
- "port-name": "XPDR1-CLIENT1",
- "port-rack": "000000.00",
- "port-shelf": "Chassis#1"
- },
- "lgx": {
- "lgx-device-name": "Some lgx-device-name",
- "lgx-port-name": "Some lgx-port-name",
- "lgx-port-rack": "000000.00",
- "lgx-port-shelf": "00"
- }
- },
- "optic-type": "gray"
- },
- "due-date": "2018-06-15T00:00:01Z",
- "operator-contact": "pw1234"
- }
- }
- response = test_utils.post_request(url, data)
+ self.cr_serv_sample_data["input"]["service-name"] = "service1-10GE"
+ self.cr_serv_sample_data["input"]["connection-type"] = "service"
+ self.cr_serv_sample_data["input"]["service-a-end"]["service-rate"] = "10"
+ self.cr_serv_sample_data["input"]["service-a-end"]["service-format"] = "Ethernet"
+ del self.cr_serv_sample_data["input"]["service-a-end"]["odu-service-rate"]
+ self.cr_serv_sample_data["input"]["service-a-end"]["tx-direction"]["port"]["port-name"] = "XPDR1-CLIENT1"
+ self.cr_serv_sample_data["input"]["service-a-end"]["rx-direction"]["port"]["port-name"] = "XPDR1-CLIENT1"
+ self.cr_serv_sample_data["input"]["service-z-end"]["service-rate"] = "10"
+ self.cr_serv_sample_data["input"]["service-z-end"]["service-format"] = "Ethernet"
+ del self.cr_serv_sample_data["input"]["service-z-end"]["odu-service-rate"]
+ self.cr_serv_sample_data["input"]["service-z-end"]["tx-direction"]["port"]["port-name"] = "XPDR1-CLIENT1"
+ self.cr_serv_sample_data["input"]["service-z-end"]["rx-direction"]["port"]["port-name"] = "XPDR1-CLIENT1"
+ response = test_utils.service_create_request(self.cr_serv_sample_data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_30_get_10GE_service1(self):
- url = "{}/operational/org-openroadm-service:service-list/services/service1-10GE"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("services/service1-10GE")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
res['odu-connection'][0]['source'])
def test_39_check_otn_topo_links(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nb_links = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 10000)
def test_40_check_otn_topo_tp(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
if (node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1'):
time.sleep(20)
def test_42_check_service_list(self):
- url = "{}/operational/org-openroadm-service:service-list"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(len(res['service-list']['services']), 2)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_47_check_otn_topo_links(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nb_links = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 0)
def test_48_check_otn_topo_tp(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
if (node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1'):
time.sleep(20)
def test_50_check_service_list(self):
- url = "{}/operational/org-openroadm-service:service-list"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(len(res['service-list']['services']), 1)
self.test_22_check_otn_topo_otu4_links()
def test_53_check_otn_topo_tp(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
res = response.json()
for node in res['network'][0]['node']:
if (node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1'):
time.sleep(20)
def test_55_get_no_service(self):
- url = "{}/operational/org-openroadm-service:service-list"
- response = test_utils.get_request(url)
+ response = test_utils.get_service_list_request("")
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
self.assertEqual(response.status_code, requests.codes.not_found)
def test_58_getLinks_OtnTopology(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('ietf-network-topology:link', res['network'][0])
'connected')
def test_02_get_portmapping_CLIENT1(self):
- url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-CLIENT1"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-CLIENT1")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-NETWORK1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
'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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-NETWORK1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
self.assertDictEqual(
{u'frequency': 196.1, u'rate': u'org-openroadm-common-types:R100G',
- u'transmit-power': -5},
+ u'transmit-power': -5, u'modulation-format': 'dp-qpsk'},
res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
def test_07_check_interface_OTU(self):
'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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-NETWORK1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
self.assertDictEqual(
{u'frequency': 196.1, u'rate': u'org-openroadm-common-types:R100G',
- u'transmit-power': -5},
+ u'transmit-power': -5, u'modulation-format': 'dp-qpsk'},
res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
def test_05_check_interface_OTU(self):
self.assertDictEqual(
{u'frequency': 196.1, u'rate': u'org-openroadm-common-types:R100G',
- u'transmit-power': -5},
+ u'transmit-power': -5, u'modulation-format': 'dp-qpsk'},
res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
def test_07_check_interface_OTU(self):
'connected')
def test_02_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
logging.info(res)
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
def test_03_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'SPDR-SA1')
self.assertEqual(len(listNode), 0)
def test_06_getLinks_OtnTopology(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('ietf-network-topology:link', res['network'][0])
def test_07_getNodes_OtnTopology(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
res = response.json()
self.assertEqual(response.status_code, requests.codes.ok)
nbNode = len(res['network'][0]['node'])
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
def test_09_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
def test_10_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('node', res['network'][0])
self.assertNotIn('node', res['network'][0])
def test_12_getNodes_OtnTopology(self):
- url = "{}/config/ietf-network:networks/network/otn-topology"
- response = test_utils.get_request(url)
+ response = test_utils.get_otn_topo_request()
res = response.json()
self.assertNotIn('node', res['network'][0])
time.sleep(10)
def test_03_rdm_portmapping_info(self):
- url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/node-info"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1/node-info")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(3)
def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
- url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/DEG1-TTP-TXRX"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1/mapping/DEG1-TTP-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_05_rdm_portmapping_DEG2_TTP_TXRX_with_ots_oms(self):
- url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/DEG2-TTP-TXRX"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1/mapping/DEG2-TTP-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_06_rdm_portmapping_SRG1_PP3_TXRX(self):
- url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/SRG1-PP3-TXRX"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1/mapping/SRG1-PP3-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_07_rdm_portmapping_SRG3_PP1_TXRX(self):
- url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/SRG3-PP1-TXRX"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1/mapping/SRG3-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
time.sleep(10)
def test_10_xpdr_portmapping_info(self):
- url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/node-info"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1/node-info")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(3)
def test_11_xpdr_portmapping_NETWORK1(self):
- url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-NETWORK1"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-NETWORK1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_12_xpdr_portmapping_NETWORK2(self):
- url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-NETWORK2"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-NETWORK2")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_13_xpdr_portmapping_CLIENT1(self):
- url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-CLIENT1"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-CLIENT1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_14_xpdr_portmapping_CLIENT2(self):
- url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-CLIENT2"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-CLIENT2")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['errors']['error'])
def test_17_xpdr_device_not_connected(self):
- url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1")
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
res['errors']['error'])
def test_20_rdm_device_not_connected(self):
- url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1")
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_03_rdm_portmapping(self):
- url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['nodes'][0]['mapping'])
def test_04_xpdr_portmapping(self):
- url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
self.assertDictEqual(
{u'rate': u'org-openroadm-common-types:R100G',
u'transmit-power': -5,
+ u'modulation-format': 'dp-qpsk',
u'frequency': 195.8},
res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
print("nodeId={}".format(nodeId))
nodeMapId = nodeId.split("-")[0] + "-" + nodeId.split("-")[1]
print("nodeMapId={}".format(nodeMapId))
- urlMapList = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId
- responseMapList = test_utils.get_request(urlMapList)
+ responseMapList = test_utils.portmapping_request(nodeMapId)
resMapList = responseMapList.json()
nbMappings = len(resMapList['nodes'][0]['mapping']) - nbMapCumul
for j in range(0, nbTp):
tpId = resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
if((not "CP" in tpId) and (not "CTP" in tpId)):
- urlMap = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId + "/mapping/" + tpId
- responseMap = test_utils.get_request(urlMap)
+ responseMap = test_utils.portmapping_request(nodeMapId+"/mapping/"+tpId)
self.assertEqual(responseMap.status_code, requests.codes.ok)
if(responseMap.status_code == requests.codes.ok):
nbMapCurrent += 1
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_02_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
logging.info(res)
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_03_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'ROADM-A1')
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_07_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'NodeA')
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_08_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(response.status_code, requests.codes.created)
def test_16_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(len(listNode), 0)
def test_17_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(response.status_code, requests.codes.created)
def test_26_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
# Delete in the topology-netconf
response = test_utils.unmount_device("ROADM-B1")
# Delete in the clli-network
- url = "{}/config/ietf-network:networks/network/clli-network/node/NodeB"
- data = {}
- response = test_utils.delete_request(url)
+ response = test_utils.del_node_request("NodeB")
self.assertEqual(response.status_code, requests.codes.ok)
def test_31_disconnect_ROADMC(self):
response = test_utils.unmount_device("ROADM-C1")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
# Delete in the clli-network
- url = "{}/config/ietf-network:networks/network/clli-network/node/NodeC"
- data = {}
- response = test_utils.delete_request(url)
+ response = test_utils.del_node_request("NodeC")
self.assertEqual(response.status_code, requests.codes.ok)
# def test_24_check_roadm2roadm_links_deletion(self):
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'ROADM-C1-DEG2')
def test_33_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'ROADM-B1')
def test_34_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
def test_36_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_37_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
response = test_utils.unmount_device("ROADM-A1")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
# Delete in the clli-network
- url = "{}/config/ietf-network:networks/network/clli-network/node/NodeA"
- data = {}
- response = test_utils.delete_request(url)
+ response = test_utils.del_node_request("NodeA")
self.assertEqual(response.status_code, requests.codes.ok)
def test_42_getClliNetwork(self):
- url = "{}/config/ietf-network:networks/network/clli-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('node', res['network'][0])
def test_43_getOpenRoadmNetwork(self):
- url = "{}/config/ietf-network:networks/network/openroadm-network"
- response = test_utils.get_request(url)
+ response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('node', res['network'][0])
NODES_PWD = "admin"
URL_CONFIG_NETCONF_TOPO = "{}/config/network-topology:network-topology/topology/topology-netconf/"
URL_CONFIG_ORDM_TOPO = "{}/config/ietf-network:networks/network/openroadm-topology/"
+URL_CONFIG_OTN_TOPO = "{}/config/ietf-network:networks/network/otn-topology/"
+URL_CONFIG_CLLI_NET = "{}/config/ietf-network:networks/network/clli-network/"
+URL_CONFIG_ORDM_NET = "{}/config/ietf-network:networks/network/openroadm-network/"
+URL_PORTMAPPING = "{}/config/transportpce-portmapping:network/nodes/"
+URL_OPER_SERV_LIST = "{}/operational/org-openroadm-service:service-list/"
+URL_SERV_CREATE = "{}/operations/org-openroadm-service:service-create"
TYPE_APPLICATION_JSON = {'Content-Type': 'application/json', 'Accept': 'application/json'}
TYPE_APPLICATION_XML = {'Content-Type': 'application/xml', 'Accept': 'application/xml'}
return delete_request(url)
+def get_clli_net_request():
+ return get_request(URL_CONFIG_CLLI_NET)
+
+
+def get_ordm_net_request():
+ return get_request(URL_CONFIG_ORDM_NET)
+
+
+def get_otn_topo_request():
+ return get_request(URL_CONFIG_OTN_TOPO)
+
+
def del_link_request(link: str):
url = URL_CONFIG_ORDM_TOPO + ("ietf-network-topology:link/" + link)
return delete_request(url)
+def del_node_request(node: str):
+ url = URL_CONFIG_CLLI_NET + ("node/" + node)
+ return delete_request(url)
+
+
+def portmapping_request(suffix: str):
+ url = URL_PORTMAPPING + suffix
+ return get_request(url)
+
+def get_service_list_request(suffix: str):
+ url = URL_OPER_SERV_LIST + suffix
+ return get_request(url)
+
+def service_create_request(attr):
+ return post_request(URL_SERV_CREATE, attr)
+
+
def shutdown_process(process):
if process is not None:
for child in psutil.Process(process.pid).children():
{py3,tapi221}: nosetests --with-xunit transportpce_tests/2.2.1/test_tapi.py
{py3,otnend2end}: nosetests --with-xunit transportpce_tests/2.2.1/test_otn_end2end.py
{py3,end2end}: nosetests --with-xunit transportpce_tests/1.2.1/test_end2end.py
- {py3,otnend2end}: nosetests --with-xunit transportpce_tests/2.2.1/test_end2end.py
+ {py3,end2end221}: nosetests --with-xunit transportpce_tests/2.2.1/test_end2end.py
#{gnpy}: - sudo docker pull atriki/gnpyrest:v1.2
{gnpy}: - sudo docker run -d -p 8008:5000 --name gnpy_tpce_rest1 atriki/gnpyrest:v1.2
{gnpy}: nosetests --with-xunit transportpce_tests/1.2.1/test_gnpy.py
Code Review / transportpce.git / commitdiff