import static org.mockito.ArgumentMatchers.any;
-import com.google.common.util.concurrent.FluentFuture;
-import com.google.common.util.concurrent.ListeningExecutorService;
-import com.google.common.util.concurrent.MoreExecutors;
import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import org.eclipse.jdt.annotation.NonNull;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.opendaylight.mdsal.binding.api.MountPoint;
import org.opendaylight.mdsal.binding.api.MountPointService;
import org.opendaylight.mdsal.binding.api.ReadWriteTransaction;
-import org.opendaylight.mdsal.common.api.CommitInfo;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.network.rev180226.networks.Network;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.network.rev180226.networks.NetworkBuilder;
@RunWith(MockitoJUnitRunner.class)
public class DeviceTransactionManagerTest {
- @Mock private MountPointService mountPointServiceMock;
- @Mock private MountPoint mountPointMock;
- @Mock private DataBroker dataBrokerMock;
- @Mock private ReadWriteTransaction rwTransactionMock;
+ @Mock
+ private MountPointService mountPointServiceMock;
+ @Mock
+ private MountPoint mountPointMock;
+ @Mock
+ private DataBroker dataBrokerMock;
+ @Mock
+ private ReadWriteTransaction rwTransactionMock;
+
private DeviceTransactionManagerImpl transactionManager;
private String defaultDeviceId = "device-id";
Mockito.verify(rwTransactionMock, Mockito.times(1)).commit();
}
- @Test
- @Ignore
- public void submitTxTimeoutTransactionTest() {
- Future<java.util.Optional<DeviceTransaction>> deviceTxFuture =
- transactionManager.getDeviceTransaction(defaultDeviceId);
- DeviceTransaction deviceTx;
- try {
- deviceTx = deviceTxFuture.get().get();
- } catch (InterruptedException | ExecutionException e) {
- Assert.fail("Exception catched! " + e);
- return;
- }
-
- deviceTx.put(defaultDatastore, defaultIid, defaultData);
-
- Exception throwedException = null;
-
- FluentFuture<? extends @NonNull CommitInfo> submitFuture = deviceTx.commit(200, defaultTimeUnit);
- try {
- submitFuture.get();
- } catch (InterruptedException e) {
- Assert.fail("Exception catched! " + e);
- } catch (ExecutionException e) {
- throwedException = e;
- }
-
- if (throwedException == null
- || !throwedException.getMessage().contains(TimeoutException.class.getName())) {
- Assert.fail("TimeoutException inside of should be thrown!");
- return;
- }
-
-
- Mockito.doReturn(FluentFutures.immediateNullFluentFuture()).when(rwTransactionMock.commit());
-
- try {
- putAndSubmit(transactionManager, defaultDeviceId, defaultDatastore, defaultIid, defaultData);
- } catch (InterruptedException | ExecutionException e) {
- Assert.fail("Exception catched! " + e);
- return;
- }
-
- Mockito.verify(rwTransactionMock, Mockito.times(2)).put(defaultDatastore, defaultIid, defaultData);
- Mockito.verify(rwTransactionMock, Mockito.times(2)).commit();
- ListeningExecutorService executor = MoreExecutors
- .listeningDecorator(Executors.newSingleThreadExecutor());
- executor.shutdown();
- }
-
private <T extends DataObject> void putAndSubmit(DeviceTransactionManagerImpl deviceTxManager, String deviceId,
LogicalDatastoreType store, InstanceIdentifier<T> path, T data)
throws ExecutionException, InterruptedException {
which defines interoperability specifications, consisting of both Optical
interoperability and Yang data models.
-Experimental support of OTN layer is introduced in Magnesium release of
-OpenDaylight. By experimental, we mean not all features can be accessed through
-northbound API based on RESTCONF encoded OpenROADM Service model. In the meanwhile,
-"east/west" APIs shall be used to trigger a path computation in the PCE (using
-path-computation-request RPC) and to create services (using otn-service-path RPC).
-OTN support will be improved in the following magnesium releases.
+End to end OTN services such as OCH-OTU4, structured ODU4 or 10GE-ODU2e
+services are supported since Magnesium SR2. OTN support will continue to be
+improved in the following releases of Magnesium and Aluminium.
+An experimental support of Flexgrid is introduced in Aluminium. Depending on
+OpenROADM device models, optical interfaces can be created according to the
+initial fixed grid (for R1.2.1, 96 channels regularly spaced of 50 GHz), or to
+a flexgrid (for R2.2.1 use of specific number of subsequent frequency slots of
+6.25 GHz depending on one side of ROADMs and transponders capabilities and on
+the other side of the rate of the channel. The full support of Flexgrid,
+including path computation and the creation of B100G (Beyond 100 Gbps) higher
+rate interfaces will be added in the following releases of Aluminium.
Module description
service. The service-list is updated following a successful service deletion. In Neon SR0 is
added the support for service from ROADM to ROADM, which brings additional flexibility and
notably allows reserving resources when transponders are not in place at day one.
-The full support of OTN services, including OTU, HO-ODU and LO-ODU will be introduced
-in next release of Magnesium.
+Magnesium SR2 fully supports end-to-end OTN services which are part of the OTN infrastructure.
+It concerns the management of OCH-OTU4 (also part of the optical infrastructure) and structured
+HO-ODU4 services. Moreover, once these two kinds of OTN infrastructure service created, it is
+possible to manage some LO-ODU services (for the time being, only 10GE-ODU2e services).
+The full support of OTN services, including 1GE-ODU0 or 100GE, will be introduced along next
+releases (Mg/Al).
PCE
^^^
about current and planned services is available in the MD-SAL data store.
Current implementation of PCE allows finding the shortest path, minimizing either the hop
-count (default) or the propagation delay. Wavelength is assigned considering a fixed grid of
-96 wavelengths. In Neon SR0, the PCE calculates the OSNR, on the base of incremental
-noise specifications provided in Open ROADM MSA. The support of unidirectional ports is
-also added. PCE handles the following constraints as hard constraints:
+count (default) or the propagation delay. Central wavelength is assigned considering a fixed
+grid of 96 wavelengths 50 GHz spaced. The assignment of wavelengths according to a flexible
+grid considering 768 subsequent slots of 6,25 GHz (total spectrum of 4.8 Thz), and their
+occupation by existing services is planned for later releases.
+In Neon SR0, the PCE calculates the OSNR, on the base of incremental noise specifications
+provided in Open ROADM MSA. The support of unidirectional ports is also added.
+
+PCE handles the following constraints as hard constraints:
- **Node exclusion**
- **SRLG exclusion**
Add/Drop modules ("SRGs") are separated from the degrees which includes line
amplifiers and WSS that switch wavelengths from one to another degree**
- **OTN layer introduced in Magnesium includes transponders as well as switch-ponders and
- mux-ponders having the ability to switch OTN containers from client to line cards. SR0 release
- includes creation of the switching pool (used to model cross-connect matrices),
+ mux-ponders having the ability to switch OTN containers from client to line cards. Mg SR0
+ release includes creation of the switching pool (used to model cross-connect matrices),
tributary-ports and tributary-slots at the initial connection of NETCONF devices.
- However, the population of OTN links, and the adjustment of the tributary ports/slots
- pull occupancy when OTN services are created will be handled in later Magnesium release.**
+ The population of OTN links (OTU4 and ODU4), and the adjustment of the tributary ports/slots
+ pool occupancy when OTN services are created is supported since Magnesium SR2.**
Renderer
Magnesium brings the support of OTN services. SR0 supports the creation of OTU4, ODU4, ODU2/ODU2e
and ODU0 interfaces. The creation of these low-order otn interfaces must be triggered through
-otn-service-path RPC. Full support (service-implementation-request /service delete rpc, topology
-alignement after the service has been created) will be provided in later releases of Magnesium.
+otn-service-path RPC. Magnesium SR2 fully supports end-to-end otn service implementation into devices
+(service-implementation-request /service delete rpc, topology alignement after the service has been created).
OLM
**REST API** : *GET /restconf/config/ietf-network:network/otn-topology*
-An optical connectivity service shall have been created in a first setp. In Magnesium SR0, the OTN
-links are not automatically populated in the topology after the Och, OTU4 and ODU4 interfaces have
-been created on the two network ports of the xpdr. Thus the otn link must be posted manually through
-the REST API (as APIDoc for example).
-
-**REST API** : *POST /restconf/config/ietf-network:networks/network/otn-topology/ietf-network-topology:link/<link-id>*
-
-**Sample JSON Data**
-
-.. code:: json
-
- {
- "ietf-network-topology:link": [
- {
- "link-id": "<link-id>",
- "source": {
- "source-node": "<xpdr-node-id>",
- "source-tp": "<xpdr-network-port-id>"
- },
- "org-openroadm-common-network:link-type": "OTN-LINK",
- "destination": {
- "dest-node": "<xpdr-node-id>",
- "dest-tp": "<xpdr-network-port-id>"
- }
- }
- ]
- }
-
+An optical connectivity service shall have been created in a first setp. Since Magnesium SR2, the OTN
+links are automatically populated in the topology after the Och, OTU4 and ODU4 interfaces have
+been created on the two network ports of the xpdr.
Creating a service
~~~~~~~~~~~~~~~~~~
Use the *service handler* module to create any end-to-end connectivity service on an OpenROADM
-network. Two kind of end-to-end services are managed by TransportPCE :
+network. Two kind of end-to-end "optical" services are managed by TransportPCE:
- 100GE service from client port to client port of two transponders (TPDR)
- Optical Channel (OC) service from client add/drop port (PP port of SRG) to client add/drop port of
two ROADMs.
As an example, the creation of a 100GE service implies among other things, the creation of OCH, OTU4
and ODU4 interfaces on the Network port of TPDR devices.
-In Magnesium SR0, the *service handler* module does not manage directly the end-to-end aspect of otn
-connectivity services. OTN services must be manualy created invoking directly the *renderer* module.
+Since Magnesium SR2, the *service handler* module directly manages some end-to-end otn
+connectivity services.
+Before creating a low-order OTN service (1GE or 10GE services terminating on client port of MUXPDR
+or SWITCH), the user must ensure that a high-order ODU4 container exists and has previously been
+configured (it means structured to support low-order otn services) to support low-order OTN containers.
+Thus, OTN service creation implies three steps:
+1. OCH-OTU4 service from network port to network port of two OTN Xponders (MUXPDR or SWITCH)
+2. HO-ODU4 service from network port to network port of two OTN Xponders (MUXPDR or SWITCH)
+3. 10GE service creation from client port to client port of two OTN Xponders (MUXPDR or SWITCH)
+The management of other OTN services (1GE-ODU0, 100GE...) is planned for future releases.
-Before creating a low-order otn service (1GE or 10GE services terminating on client port of MUXPDR
-or SWITCH), the user must ensure that a high-order ODU4 container exists and has previously been
-configured (low-order structured) to support low-order OTN containers. Thus, otn service creation
-implies two steps:
-1. OCH, OTU-4 and ODU-4 service from network port to network port of two OTN Xponders
-(MUXPDR or SWITCH)
-2. 1GE or 10GE service creation from client port to client port of two OTN Xponders
-(MUXPDR or SWITCH)
-
-Since in Magnesium SR0 otn service creation is not automatically done by TransportPCE, the user has
-to check the connectivity between nodes using the internal API of the *PCE* module, through the
-path-computation-request rpc.
-Afterwards, for the first step otn service creation, the user has to use the internal service-path
-rpc of the *renderer* module.
-Finally, the 1GE and 10GE services creation is performed using internal otn-service-path rpc of the
-*renderer* module.
100GE service creation
^^^^^^^^^^^^^^^^^^^^^^
*openroadm-topology* and then invokes *renderer* and *OLM* to implement the end-to-end path into
the devices.
-OTN OCH, OTU4 and ODU4 service creation
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+OTN OCH-OTU4 service creation
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-Use the following REST RPC to invoke *renderer* module in order to create adequate interfaces on
-otn device in order to support a bidirectional end-to-end Low-Order OTN connectivity service on
-network port of a MUXPDR or SWITCH xpdr node.
+Use the following REST RPC to invoke *service handler* module in order to create over the optical
+infrastructure a bidirectional end-to-end OTU4 over an optical wavelength connectivity service
+between two optical network ports of OTN Xponder (MUXPDR or SWITCH). Such service configure the
+optical network infrastructure composed of rdm nodes.
-**REST API** : *POST /restconf/operations/transportpce-device-renderer:service-path*
+**REST API** : *POST /restconf/operations/org-openroadm-service:service-create*
**Sample JSON Data**
.. code:: json
- {
- "input": {
- "nodes": [
- {
- "node-id": "<otn-node-id>",
- "dest-tp": "<otn-network-port-logical-connection-point>"
- }
- ],
- "modulation-format": "qpsk",
- "operation": "create",
- "service-name": "<service-name>",
- "wave-number": "<wavenumber-returned-by-PCE>"
- }
- }
+ {
+ "input": {
+ "sdnc-request-header": {
+ "request-id": "request-1",
+ "rpc-action": "service-create",
+ "request-system-id": "appname"
+ },
+ "service-name": "something",
+ "common-id": "commonId",
+ "connection-type": "infrastructure",
+ "service-a-end": {
+ "service-rate": "100",
+ "node-id": "<xpdr-node-id>",
+ "service-format": "OTU",
+ "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
+ "clli": "<ccli-name>",
+ "tx-direction": {
+ "port": {
+ "port-device-name": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-network-port-in-otn-topology>",
+ "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": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-network-port-in-otn-topology>",
+ "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": "<xpdr-node-id>",
+ "service-format": "OTU",
+ "otu-service-rate": "org-openroadm-otn-common-types:OTU4",
+ "clli": "<ccli-name>",
+ "tx-direction": {
+ "port": {
+ "port-device-name": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-network-port-in-otn-topology>",
+ "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": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-network-port-in-otn-topology>",
+ "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": "yyyy-mm-ddT00:00:01Z",
+ "operator-contact": "some-contact-info"
+ }
+ }
-1GE/ODU0 and 10GE/ODU2e service creation
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+As for the previous RPC, this RPC invokes the *PCE* module to compute a path over the
+*openroadm-topology* and then invokes *renderer* and *OLM* to implement the end-to-end path into
+the devices.
+
+OTN HO-ODU4 service creation
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Use the following REST RPC to invoke *service handler* module in order to create over the optical
+infrastructure a bidirectional end-to-end ODU4 OTN service over an OTU4 and structured to support
+low-order OTN services (ODU2e, ODU0). As for OTU4, such a service must be created between two network
+ports of OTN Xponder (MUXPDR or SWITCH).
+
+**REST API** : *POST /restconf/operations/org-openroadm-service:service-create*
+
+**Sample JSON Data**
+
+.. code:: json
+
+ {
+ "input": {
+ "sdnc-request-header": {
+ "request-id": "request-1",
+ "rpc-action": "service-create",
+ "request-system-id": "appname"
+ },
+ "service-name": "something",
+ "common-id": "commonId",
+ "connection-type": "infrastructure",
+ "service-a-end": {
+ "service-rate": "100",
+ "node-id": "<xpdr-node-id>",
+ "service-format": "ODU",
+ "otu-service-rate": "org-openroadm-otn-common-types:ODU4",
+ "clli": "<ccli-name>",
+ "tx-direction": {
+ "port": {
+ "port-device-name": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-network-port-in-otn-topology>",
+ "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": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-network-port-in-otn-topology>",
+ "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": "<xpdr-node-id>",
+ "service-format": "ODU",
+ "otu-service-rate": "org-openroadm-otn-common-types:ODU4",
+ "clli": "<ccli-name>",
+ "tx-direction": {
+ "port": {
+ "port-device-name": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-network-port-in-otn-topology>",
+ "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": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-network-port-in-otn-topology>",
+ "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": "yyyy-mm-ddT00:00:01Z",
+ "operator-contact": "some-contact-info"
+ }
+ }
-Use the following REST RPC to invoke *renderer* module and create adequate interfaces on otn xpdr
-device (MUXPDR or SWITCH) in order to support a low-order otn service on its client port.
+As for the previous RPC, this RPC invokes the *PCE* module to compute a path over the
+*otn-topology* that must contains OTU4 links with valid bandwidth parameters, and then
+invokes *renderer* and *OLM* to implement the end-to-end path into the devices.
-- 1GE and ODU0 interfaces for 1GE services
-- 10GE and ODU2e interfaces for 10GE services
+OTN 10GE-ODU2e service creation
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-The following example corresponds to the creation of a 10GE service
+Use the following REST RPC to invoke *service handler* module in order to create over the OTN
+infrastructure a bidirectional end-to-end 10GE-ODU2e OTN service over an ODU4.
+Such a service must be created between two client ports of OTN Xponder (MUXPDR or SWITCH)
+configured to support 10GE interfaces.
-*REST API** : *POST /restconf/operations/transportpce-device-renderer:otn-service-path*
+**REST API** : *POST /restconf/operations/org-openroadm-service:service-create*
**Sample JSON Data**
.. code:: json
- {
- "input": {
- "service-rate": "10G",
- "service-type": "Ethernet",
- "ethernet-encoding": "something",
- "trib-slot": "<trib-slot-number-inside-supported-ODU4>",
- "trib-port-number": "<trib-port-number-inside-supported-ODU4>",
- "operation": "create",
- "service-name": "something",
- "nodes": [
- {
- "node-id": "<otn-node-id>",
- "client-tp": "<client-port-logical-connection-point>",
- "network-tp": "<network-port-logical-connection-point>"
- }
- ]
- }
- }
+ {
+ "input": {
+ "sdnc-request-header": {
+ "request-id": "request-1",
+ "rpc-action": "service-create",
+ "request-system-id": "appname"
+ },
+ "service-name": "something",
+ "common-id": "commonId",
+ "connection-type": "service",
+ "service-a-end": {
+ "service-rate": "10",
+ "node-id": "<xpdr-node-id>",
+ "service-format": "Ethernet",
+ "clli": "<ccli-name>",
+ "subrate-eth-sla": {
+ "subrate-eth-sla": {
+ "committed-info-rate": "10000",
+ "committed-burst-size": "64"
+ }
+ },
+ "tx-direction": {
+ "port": {
+ "port-device-name": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-client-port-in-otn-topology>",
+ "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": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-client-port-in-otn-topology>",
+ "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": "<xpdr-node-id>",
+ "service-format": "Ethernet",
+ "clli": "<ccli-name>",
+ "subrate-eth-sla": {
+ "subrate-eth-sla": {
+ "committed-info-rate": "10000",
+ "committed-burst-size": "64"
+ }
+ },
+ "tx-direction": {
+ "port": {
+ "port-device-name": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-client-port-in-otn-topology>",
+ "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": "<xpdr-node-id-in-otn-topology>",
+ "port-type": "fixed",
+ "port-name": "<xpdr-client-port-in-otn-topology>",
+ "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": "yyyy-mm-ddT00:00:01Z",
+ "operator-contact": "some-contact-info"
+ }
+ }
+
+As for the previous RPC, this RPC invokes the *PCE* module to compute a path over the
+*otn-topology* that must contains ODU4 links with valid bandwidth parameters, and then
+invokes *renderer* and *OLM* to implement the end-to-end path into the devices.
-.. note::
- OTN links are not automatically populated in the topology after the ODU2e interfaces have
- been created on the two client ports of the xpdr. The otn link can be posted manually through
- the REST API (APIDoc).
.. note::
- With Magnesium SR0, the service-list corresponding to 1GE/10GE and OTU4/ODU4 services is not
- updated in the datastore after the interfaces have been created in the device.
+ Since Magnesium SR2, the service-list corresponding to OCH-OTU4, ODU4 or again 10GE-ODU2e services is
+ updated in the service-list datastore.
.. note::
- trib-slot is used when the equipment supports contiguous trib-slot allocation (supported in
+ trib-slot is used when the equipment supports contiguous trib-slot allocation (supported from
Magnesium SR0). The trib-slot provided corresponds to the first of the used trib-slots.
complex-trib-slots will be used when the equipment does not support contiguous trib-slot
allocation. In this case a list of the different trib-slots to be used shall be provided.
Deleting a service
~~~~~~~~~~~~~~~~~~
-Deleting a 100GE and OC service
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Deleting any kind of service
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Use the following REST RPC to invoke *service handler* module in order to delete a given optical
connectivity service.
Most important parameters for this REST RPC is the *service-name*.
-Deleting 1GE/ODU0 or 10GE/ODU2e
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-
-Use the following REST RPC to invoke *renderer* module and delete adequate interfaces on otn node.
-The following example corresponds to the deletion of a 10GE service
-
-*REST API** : *POST /restconf/operations/transportpce-device-renderer:otn-service-path*
-
-**Sample JSON Data**
-
-.. code:: json
-
- {
- "input": {
- "service-rate": "10G",
- "service-type": "Ethernet",
- "ethernet-encoding": "something",
- "trib-slot": "<trib-slot-number-inside-supported-ODU4>",
- "trib-port-number": "<trib-port-number-inside-supported-ODU4>",
- "operation": "delete",
- "service-name": "something",
- "nodes": [
- {
- "node-id": "<otn-node-id>",
- "client-tp": "<client-port-logical-connection-point>",
- "network-tp": "<network-port-logical-connection-point>"
- }
- ]
- }
- }
-
-
-Deleting OTN OCH, OTU4 and ODU4 service
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-
-Use the following REST RPC to invoke *renderer* module in order to delete adequate interfaces on
-otn node.
-
-**REST API** : *POST /restconf/operations/transportpce-device-renderer:service-path*
-
-**Sample JSON Data**
-
-.. code:: json
-
- {
- "input": {
- "nodes": [
- {
- "node-id": "<otn-node-id>",
- "dest-tp": "<otn-network-port-logical-connection-point>"
- }
- ],
- "modulation-format": "qpsk",
- "operation": "delete",
- "service-name": "<service-name>",
- "wave-number": "<wavenumber-returned-by-PCE>",
- }
- }
.. note::
- Be sure to have deleted all low-order otn services before deleting high-order OTN container
-
+ Deleting OTN services implies proceeding in the reverse way to their creation. Thus, OTN
+ service deletion must respect the three following steps:
+ 1. delete first all 10GE services supported over any ODU4 to be deleted
+ 2. delete ODU4
+ 3. delete OCH-OTU4 supporting the just deleted ODU4
Invoking PCE module
~~~~~~~~~~~~~~~~~~~
1. Devices must support the standard OpenROADM Models more precisely versions 1.2.1 and 2.2.1.
Since Magnesium SR0, an OTN experimental support is provided for OpenROADM devices 2.2.1.
+ Magnesium SR2 is the first release managing end-to-end OTN services, as OCH-OTU4,
+ structured ODU4 or again 10GE-ODU2e services.
2. Devices must support configuration through NETCONF protocol/API.
feature:install odl-transportpce-tapi
+When installing the TAPI feature, you might encounter a heap memory size problem in Karaf.
+In that case, consider increasing Karaf heap memory size.
+For example by modifying the environment variables JAVA_MIN_MEM and JAVA_MAX_MEM before starting Karaf::
+
+ export JAVA_MIN_MEM=1024M
+ export JAVA_MAX_MEM=4069M
+
if you need the inventory external connector support limited to 1.2.1 OpenROADM devices, then run::
feature:install odl-transportpce-inventory
<groupId>${project.groupId}</groupId>
<artifactId>test-common</artifactId>
<version>${project.version}</version>
+ <scope>test</scope>
</dependency>
-
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>3.0.2</version>
<optional>true</optional>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>transportpce-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
</dependencies>
-
</project>
Assert.assertEquals("Failed", output.getResult());
}
- /*
- @Test
- public void testCalculateSpanlossBase() {
- CalculateSpanlossBaseInput input = OlmPowerServiceRpcImplUtil.getCalculateSpanlossBaseInput();
- //TODO
- Mockito.when(this.olmPowerServiceMock.calculateSpanlossBase(Mockito.any()))
- .thenReturn(new CalculateSpanlossBaseOutputBuilder().setResult("Failed").build());
- CalculateSpanlossBaseOutput output = this.olmPowerServiceMock.calculateSpanlossBase(input);
- Assert.assertEquals("Failed", output.getResult());
-
- }
- */
-
- /*
- @Test
- public void testCalculateSpanlossBase2() {
- NetworkKey overlayTopologyKey = new NetworkKey(new NetworkId(NetworkUtils.OVERLAY_NETWORK_ID));
-<<<<<<< HEAD
- InstanceIdentifier<Network1> networkIID = InstanceIdentifier.builder(Networks.class)
- .child(Network.class, overlayTopologyKey)
-=======
- InstanceIdentifier<Network1> networkTopoIID = InstanceIdentifier.builder(Networks.class)
- .child(org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang
- .ietf.network.rev180226.networks.Network.class,
- overlayTopologyKey)
->>>>>>> 87d8bf0... Retrieve OLM modifs from change 80051
- .augmentation(Network1.class)
- .build();
- InstanceIdentifier<Network> networkIID = InstanceIdentifier.builder(Networks.class)
- .child(org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang
- .ietf.network.rev180226.networks.Network.class,
- overlayTopologyKey)
- .build();
- Network1 network = TransactionUtils.getNetwork();
- Network ietfNetwork = TransactionUtils.getOverLayNetwork();
- TransactionUtils.writeTransaction(this.getDataBroker(), networkIID, ietfNetwork);
- TransactionUtils.writeTransaction(this.getDataBroker(), networkTopoIID, network);
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- LOG.error("Write transaction failed !",e);
- }
- CalculateSpanlossBaseInput input = OlmPowerServiceRpcImplUtil.getCalculateSpanlossBaseInput();
- //TODO
- Mockito.when(this.olmPowerServiceMock.calculateSpanlossBase(Mockito.any()))
- .thenReturn(new CalculateSpanlossBaseOutputBuilder().setResult("Failed").build());
- CalculateSpanlossBaseOutput output = this.olmPowerServiceMock.calculateSpanlossBase(input);
- Assert.assertEquals("Failed", output.getResult());
-
- }
- */
@Ignore
@Test
public void testCalculateSpanlossBase3() {
}
-// @Test
-// public void testCalculateSpanlossBase4() throws InterruptedException {
-// NetworkKey overlayTopologyKey = new NetworkKey(new NetworkId(NetworkUtils.OVERLAY_NETWORK_ID));
-// InstanceIdentifier<Network1> networkIID = InstanceIdentifier.builder(Network.class, overlayTopologyKey)
-// .augmentation(Network1.class)
-// .build();
-// Network1 network = TransactionUtils.getNetwork2();
-// TransactionUtils.writeTransaction(this.getDataBroker(), networkIID, network);
-//
-// Thread.sleep(500);
-//
-// List<NodeId> nodes = TransactionUtils.getNodes();
-// Node n;
-// SupportingNode sn;
-//
-// List<SupportingNode> snl;
-// NodeId n5 = new NodeId("node 5");
-// KeyedInstanceIdentifier<Node, NodeKey> mappedNodeII2 =
-// InstanceIdentifiers.UNDERLAY_NETWORK_II.child(Node.class, new NodeKey(n5));
-//
-// sn = new SupportingNodeBuilder().setNetworkRef(new NetworkId(NetworkUtils.UNDERLAY_NETWORK_ID))
-// .setNodeRef(n5).build();
-// snl = new ArrayList<>();
-// snl.add(sn);
-// n = new org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.network.rev150608
-// .network.NodeBuilder().setNodeId(n5).build();
-// TransactionUtils.writeTransaction(this.getDataBroker(), mappedNodeII2, n);
-// Thread.sleep(500);
-//
-// for(int i=0; i < nodes.size(); i++) {
-// KeyedInstanceIdentifier<Node, NodeKey> mappedNodeII =
-// InstanceIdentifiers.OVERLAY_NETWORK_II.child(Node.class, new NodeKey(nodes.get(i)));
-// if (i != 0){
-// sn = new SupportingNodeBuilder().setNetworkRef(new NetworkId(NetworkUtils.UNDERLAY_NETWORK_ID))
-// .setNodeRef(nodes.get(i-1)).build();
-// }else {
-// sn = new SupportingNodeBuilder().setNodeRef(n5).build();
-// }
-// snl = new ArrayList<>();
-// snl.add(sn);
-// n = new org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.network.rev150608
-// .network.NodeBuilder().setNodeId(nodes.get(i)).setSupportingNode(snl).build();
-// TransactionUtils.writeTransaction(this.getDataBroker(), mappedNodeII, n);
-// Thread.sleep(500);
-// }
-//
-// Thread.sleep(1000);
-//
-// CalculateSpanlossBaseInput input = OlmPowerServiceRpcImplUtil.getCalculateSpanlossBaseInput2();
-// CalculateSpanlossBaseOutput output = this.olmPowerService.calculateSpanlossBase(input);
-// Assert.assertEquals("Failed", output.getResult());
-//
-// }
-
@Test
public void testCalculateSpanlossCurrent1() {
CalculateSpanlossCurrentInput input = OlmPowerServiceRpcImplUtil.getCalculateSpanlossCurrentInput();
import org.opendaylight.yang.gen.v1.http.org.opendaylight.transportpce.pce.rev200128.CancelResourceReserveInputBuilder;
-
@Ignore
public class PceServiceRPCImplTest extends AbstractTest {
import org.opendaylight.yangtools.yang.common.Uint32;
-
@Ignore
public class PceLinkTest extends AbstractTest {
import org.opendaylight.yang.gen.v1.gnpy.path.rev200909.result.response.response.type.NoPathCaseBuilder;
import org.opendaylight.yang.gen.v1.gnpy.path.rev200909.result.response.response.type.PathCaseBuilder;
-
@Ignore
public class PathComputationServiceImplTest extends AbstractTest {
saveOpenRoadmNetwork(networkMap.get(0), NetworkUtils.UNDERLAY_NETWORK_ID, dataBroker);
networks = (Networks) JsonUtil.getInstance().getDataObjectFromJson(topoReader,
QName.create("urn:ietf:params:xml:ns:yang:ietf-network", "2018-02-26", "networks"));
- saveOpenRoadmNetwork(networkMap.get(0), NetworkUtils.OVERLAY_NETWORK_ID, dataBroker);
+ saveOpenRoadmNetwork(networkMap.get(0), NetworkUtils.UNDERLAY_NETWORK_ID, dataBroker);
} catch (IOException | ExecutionException | InterruptedException e) {
LOG.error("Cannot init test ", e);
Assert.fail("Cannot init test ");
<groupId>${project.groupId}</groupId>
<artifactId>test-common</artifactId>
<version>${project.version}</version>
+ <scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
-import json
import time
import unittest
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
-import json
import os
import sys
import time
class TransportGNPYtesting(unittest.TestCase):
- @classmethod
- def __init_logfile(cls):
- GNPY_LOGFILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
- "..", "..", "transportpce_tests", "gnpy.log")
- if os.path.isfile(GNPY_LOFGILE):
- os.remove(GNPY_LOFGILE)
-
topo_cllinet_data = None
topo_ordnet_data = None
topo_ordtopo_data = None
@classmethod
def setUpClass(cls):
+ # pylint: disable=bare-except
try:
sample_files_parsed = False
TOPO_CLLINET_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
# Path computed by PCE is feasible according to Gnpy
def test_04_path_computation_FeasibleWithPCE(self):
response = test_utils.path_computation_request("request-1", "service-1",
- {"node-id": "XPONDER-1", "service-rate": "100", "service-format": "Ethernet", "clli": "Node1"},
- {"node-id": "XPONDER-5", "service-rate": "100", "service-format": "Ethernet", "clli": "Node5"})
+ {"node-id": "XPONDER-1", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node1"},
+ {"node-id": "XPONDER-5", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node5"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# another one (low SNR)
def test_05_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
response = test_utils.path_computation_request("request-2", "service-2",
- {"node-id": "XPONDER-1", "service-rate": "100", "service-format": "Ethernet", "clli": "Node1"},
- {"node-id": "XPONDER-5", "service-rate": "100", "service-format": "Ethernet", "clli": "Node5"},
- {"include_": { "ordered-hops": [
- {"hop-number": "0", "hop-type": {"node-id": "OpenROADM-2"}},
- {"hop-number": "1", "hop-type": {"node-id": "OpenROADM-3"}},
- {"hop-number": "2", "hop-type": {"node-id": "OpenROADM-4"}}]}})
+ {"node-id": "XPONDER-1", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node1"},
+ {"node-id": "XPONDER-5", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node5"},
+ {"include_": {"ordered-hops": [
+ {"hop-number": "0", "hop-type": {"node-id": "OpenROADM-2"}},
+ {"hop-number": "1", "hop-type": {"node-id": "OpenROADM-3"}},
+ {"hop-number": "2", "hop-type": {"node-id": "OpenROADM-4"}}]}
+ })
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# #PCE cannot find a path while GNPy finds a feasible one
def test_06_path_computation_NotFoundByPCE_FoundByGNPy(self):
response = test_utils.path_computation_request("request-3", "service-3",
- {"node-id": "XPONDER-1", "service-rate": "100", "service-format": "Ethernet", "clli": "Node1"},
- {"node-id": "XPONDER-4", "service-rate": "100", "service-format": "Ethernet", "clli": "Node5"},
- {"include_": {"ordered-hops": [
- {"hop-number": "0", "hop-type": {"node-id": "OpenROADM-2"}},
- {"hop-number": "1", "hop-type": {"node-id": "OpenROADM-3"}}]}})
+ {"node-id": "XPONDER-1", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node1"},
+ {"node-id": "XPONDER-4", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node5"},
+ {"include_": {"ordered-hops": [
+ {"hop-number": "0", "hop-type": {"node-id": "OpenROADM-2"}},
+ {"hop-number": "1", "hop-type": {"node-id": "OpenROADM-3"}}]}
+ })
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# Not found path by PCE and GNPy cannot find another one
def test_07_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
response = test_utils.path_computation_request("request-4", "service-4",
- {"node-id": "XPONDER-1", "service-rate": "100", "service-format": "Ethernet", "clli": "Node1"},
- {"node-id": "XPONDER-4", "service-rate": "100", "service-format": "Ethernet", "clli": "Node5"},
- {"include_": {"ordered-hops": [
- {"hop-number": "0","hop-type": {"node-id": "OpenROADM-2"}},
- {"hop-number": "1","hop-type": {"node-id": "OpenROADM-3"}},
- {"hop-number": "2","hop-type": {"node-id": "OpenROADM-4"}},
- {"hop-number": "3","hop-type": {"node-id": "OpenROADM-3"}}]}})
+ {"node-id": "XPONDER-1", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node1"},
+ {"node-id": "XPONDER-4", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "Node5"},
+ {"include_": {"ordered-hops": [
+ {"hop-number": "0", "hop-type": {"node-id": "OpenROADM-2"}},
+ {"hop-number": "1", "hop-type": {"node-id": "OpenROADM-3"}},
+ {"hop-number": "2", "hop-type": {"node-id": "OpenROADM-4"}},
+ {"hop-number": "3", "hop-type": {"node-id": "OpenROADM-3"}}]}
+ })
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# http://www.apache.org/licenses/LICENSE-2.0
#############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
import time
-import json
import requests
from common import test_utils
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertIn('Request processed', res["output"]["result"])
time.sleep(10)
- """to test case where SRG where the xpdr is connected to has no optical range data"""
+ #"""to test case where SRG where the xpdr is connected to has no optical range data"""
def test_31_connect_xprdA_to_roadmA(self):
response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "2",
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
-import json
import os
import sys
import time
@classmethod
def setUpClass(cls):
+ # pylint: disable=bare-except
try:
sample_files_parsed = False
TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
# Path Computation success
def test_04_path_computation_xpdr_bi(self):
response = test_utils.path_computation_request("request-1", "service-1",
- {"node-id": "XPDRA01", "service-rate": "100", "service-format": "Ethernet", "clli": "nodeA"},
- {"node-id": "XPDRC01", "service-rate": "100", "service-format": "Ethernet", "clli": "nodeC"})
+ {"node-id": "XPDRA01", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "nodeA"},
+ {"node-id": "XPDRC01", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "nodeC"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Path Computation success
def test_05_path_computation_rdm_bi(self):
response = test_utils.path_computation_request("request-1", "service-1",
- {"node-id": "ROADMA01", "service-rate": "100", "service-format": "Ethernet", "clli": "NodeA"},
- {"node-id": "ROADMC01", "service-rate": "100", "service-format": "Ethernet", "clli": "NodeC"})
+ {"node-id": "ROADMA01", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "NodeA"},
+ {"node-id": "ROADMC01", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "NodeC"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Path Computation success
def test_11_path_computation_xpdr_uni(self):
response = test_utils.path_computation_request("request-1", "service-1",
- {"node-id": "XPONDER-1-2", "service-rate": "100", "service-format": "Ethernet", "clli": "ORANGE1"},
- {"node-id": "XPONDER-3-2", "service-rate": "100", "service-format": "Ethernet", "clli": "ORANGE3"})
+ {"node-id": "XPONDER-1-2", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "ORANGE1"},
+ {"node-id": "XPONDER-3-2", "service-rate": "100",
+ "service-format": "Ethernet", "clli": "ORANGE3"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Path Computation success
def test_12_path_computation_rdm_uni(self):
response = test_utils.path_computation_request("request1", "service1",
- {"service-rate": "100", "service-format": "Ethernet", "clli": "cll21", "node-id": "OpenROADM-2-1"},
- {"service-rate": "100", "service-format": "Ethernet", "clli": "ncli22", "node-id": "OpenROADM-2-2"})
+ {"service-rate": "100", "service-format": "Ethernet",
+ "clli": "cll21", "node-id": "OpenROADM-2-1"},
+ {"service-rate": "100", "service-format": "Ethernet",
+ "clli": "ncli22", "node-id": "OpenROADM-2-2"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
for i in range(0, 15):
atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i]
- if (atoz['id'] == '14'):
+ if atoz['id'] == '14':
self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX')
- if (ztoa['id'] == '0'):
+ if ztoa['id'] == '0':
self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX')
time.sleep(5)
# Test failed path computation
def test_17_fail_path_computation(self):
response = test_utils.post_request(test_utils.URL_PATH_COMPUTATION_REQUEST,
- {"input": { "service-handler-header": { "request-id": "request-1" }}})
+ {"input": {"service-handler-header": {"request-id": "request-1"}}})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Service Name is not set',
# Test1 success path computation
def test_18_success1_path_computation(self):
response = test_utils.path_computation_request("request1", "service1",
- {"service-format": "Ethernet","service-rate": "100", "clli": "ORANGE2", "node-id": "XPONDER-2-2",
- "tx-direction": {"port": {
- "port-device-name": "Some port-device-name",
- "port-type": "Some port-type",
- "port-name": "Some port-name",
- "port-rack": "Some port-rack",
- "port-shelf": "Some port-shelf",
- "port-slot": "Some port-slot",
- "port-sub-slot": "Some port-sub-slot"
- }},
- "rx-direction": {"port": {
- "port-device-name": "Some port-device-name",
- "port-type": "Some port-type",
- "port-name": "Some port-name",
- "port-rack": "Some port-rack",
- "port-shelf": "Some port-shelf",
- "port-slot": "Some port-slot",
- "port-sub-slot": "Some port-sub-slot"
- }}},
- {"service-format": "Ethernet", "service-rate": "100", "clli": "ORANGE1", "node-id": "XPONDER-1-2",
- "tx-direction": {"port": {
- "port-device-name": "Some port-device-name",
- "port-type": "Some port-type",
- "port-name": "Some port-name",
- "port-rack": "Some port-rack",
- "port-shelf": "Some port-shelf",
- "port-slot": "Some port-slot",
- "port-sub-slot": "Some port-sub-slot"
- }},
- "rx-direction": {"port": {
- "port-device-name": "Some port-device-name",
- "port-type": "Some port-type",
- "port-name": "Some port-name",
- "port-rack": "Some port-rack",
- "port-shelf": "Some port-shelf",
- "port-slot": "Some port-slot",
- "port-sub-slot": "Some port-sub-slot"
- }}},
- {"customer-code": ["Some customer-code"], "co-routing": {"existing-service": ["Some existing-service"]}},
- {"customer-code": ["Some customer-code"], "co-routing": {"existing-service": ["Some existing-service"]}},
- "hop-count", {"locally-protected-links": "true"})
+ {"service-format": "Ethernet", "service-rate": "100",
+ "clli": "ORANGE2", "node-id": "XPONDER-2-2",
+ "tx-direction": {"port": {
+ "port-device-name": "Some port-device-name",
+ "port-type": "Some port-type",
+ "port-name": "Some port-name",
+ "port-rack": "Some port-rack",
+ "port-shelf": "Some port-shelf",
+ "port-slot": "Some port-slot",
+ "port-sub-slot": "Some port-sub-slot"
+ }},
+ "rx-direction": {"port": {
+ "port-device-name": "Some port-device-name",
+ "port-type": "Some port-type",
+ "port-name": "Some port-name",
+ "port-rack": "Some port-rack",
+ "port-shelf": "Some port-shelf",
+ "port-slot": "Some port-slot",
+ "port-sub-slot": "Some port-sub-slot"
+ }}},
+ {"service-format": "Ethernet", "service-rate": "100",
+ "clli": "ORANGE1", "node-id": "XPONDER-1-2",
+ "tx-direction": {"port": {
+ "port-device-name": "Some port-device-name",
+ "port-type": "Some port-type",
+ "port-name": "Some port-name",
+ "port-rack": "Some port-rack",
+ "port-shelf": "Some port-shelf",
+ "port-slot": "Some port-slot",
+ "port-sub-slot": "Some port-sub-slot"
+ }},
+ "rx-direction": {"port": {
+ "port-device-name": "Some port-device-name",
+ "port-type": "Some port-type",
+ "port-name": "Some port-name",
+ "port-rack": "Some port-rack",
+ "port-shelf": "Some port-shelf",
+ "port-slot": "Some port-slot",
+ "port-sub-slot": "Some port-sub-slot"
+ }}},
+ {"customer-code": ["Some customer-code"],
+ "co-routing": {"existing-service": ["Some existing-service"]}
+ },
+ {"customer-code": ["Some customer-code"],
+ "co-routing": {"existing-service": ["Some existing-service"]}
+ },
+ "hop-count", {"locally-protected-links": "true"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Test2 success path computation with path description
def test_19_success2_path_computation(self):
response = test_utils.path_computation_request("request 1", "service 1",
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-3-2", "clli": "ORANGE3"})
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-3-2", "clli": "ORANGE3"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Test3 success path computation with hard-constraints exclude
def test_20_success3_path_computation(self):
response = test_utils.path_computation_request("request 1", "service 1",
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-3-2", "clli": "ORANGE3"},
- {"exclude_": { "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"] }})
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-3-2", "clli": "ORANGE3"},
+ {"exclude_": {"node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]}})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
def test_21_path_computation_before_oms_attribute_deletion(self):
response = test_utils.path_computation_request("request 1", "service 1",
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
find = False
for i in range(0, nbElmPath):
- resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
- if(resource_i == link):
+ resource_i = (res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
+ ['resource'])
+ if resource_i == link:
find = True
self.assertEqual(find, True)
time.sleep(5)
# Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
def test_23_path_computation_after_oms_attribute_deletion(self):
response = test_utils.path_computation_request("request 1", "service 1",
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
- {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
+ {"service-rate": "100", "service-format": "Ethernet",
+ "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
find = False
for i in range(0, nbElmPath):
- resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
- if (resource_i == link):
+ resource_i = (res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
+ ['resource'])
+ if resource_i == link:
find = True
self.assertNotEqual(find, True)
time.sleep(5)
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
import time
import requests
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
# http://www.apache.org/licenses/LICENSE-2.0
#############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
-import json
import time
import requests
from common import test_utils
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
def test_05_service_path_create(self):
response = test_utils.service_path_request("create", "service_test", "7",
[{"renderer:node-id": "ROADMA01",
- "renderer:src-tp": "SRG1-PP7-TXRX", "renderer:dest-tp": "DEG1-TTP-TXRX"},
+ "renderer:src-tp": "SRG1-PP7-TXRX",
+ "renderer:dest-tp": "DEG1-TTP-TXRX"},
{"renderer:node-id": "XPDRA01",
- "renderer:src-tp": "XPDR1-CLIENT1", "renderer:dest-tp": "XPDR1-NETWORK1"}])
+ "renderer:src-tp": "XPDR1-CLIENT1",
+ "renderer:dest-tp": "XPDR1-NETWORK1"}])
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes: ROADMA01', res["output"]["result"])
def test_14_service_path_delete(self):
response = test_utils.service_path_request("delete", "service_test", "7",
[{"renderer:node-id": "ROADMA01",
- "renderer:src-tp": "SRG1-PP7-TXRX", "renderer:dest-tp": "DEG1-TTP-TXRX"},
+ "renderer:src-tp": "SRG1-PP7-TXRX",
+ "renderer:dest-tp": "DEG1-TTP-TXRX"},
{"renderer:node-id": "XPDRA01",
- "renderer:src-tp": "XPDR1-CLIENT1", "renderer:dest-tp": "XPDR1-NETWORK1"}])
+ "renderer:src-tp": "XPDR1-CLIENT1",
+ "renderer:dest-tp": "XPDR1-NETWORK1"}])
self.assertEqual(response.status_code, requests.codes.ok)
self.assertEqual(response.json(), {
'output': {'result': 'Request processed', 'success': True}})
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_17_service_path_delete_rdm_check(self):
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_18_service_path_delete_xpdr_check(self):
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_19_service_path_delete_xpdr_check(self):
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_20_service_path_delete_xpdr_check(self):
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_21_service_path_delete_xpdr_check(self):
res = response.json()
self.assertIn(
{"error-type": "application", "error-tag": "data-missing",
- "error-message": "Request could not be completed because the relevant data model content does not exist"},
+ "error-message":
+ "Request could not be completed because the relevant data model content does not exist"},
res['errors']['error'])
def test_22_service_path_delete_xpdr_check(self):
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+# a pylint false positive due to unittest
+# pylint: disable=no-self-use
+
import time
import unittest
import requests
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
# Verify the termination points of the ROADMA
- def test_02_compareOpenroadmTopologyPortMapping_rdm(self):
+ def test_02_compare_Openroadm_topology_portmapping_rdm(self):
responseTopo = test_utils.get_ordm_topo_request("")
resTopo = responseTopo.json()
nbNode = len(resTopo['network'][0]['node'])
- nbMapCumul = 0
- nbMappings = 0
for i in range(0, nbNode):
nodeId = resTopo['network'][0]['node'][i]['node-id']
nodeMapId = nodeId.split("-")[0]
- responseMapList = test_utils.portmapping_request(nodeMapId)
- resMapList = responseMapList.json()
- nbMappings = len(resMapList['nodes'][0]['mapping']) - nbMapCumul
+ test_utils.portmapping_request(nodeMapId)
nbTp = len(resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- nbMapCurrent = 0
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)):
- responseMap = test_utils.portmapping_request(nodeMapId+"/mapping/"+tpId)
+ 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)
# #Verify the termination points related to XPDR
- def test_05_compareOpenroadmTopologyPortMapping_xpdr(self):
- self.test_02_compareOpenroadmTopologyPortMapping_rdm()
+ def test_05_compare_Openroadm_topology_portmapping_xpdr(self):
+ self.test_02_compare_Openroadm_topology_portmapping_rdm()
# Disconnect the XPDRA
def test_06_disconnect_device(self):
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-import json
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+# pylint: disable=too-many-lines
+
import time
import unittest
import requests
class TransportPCETopologyTesting(unittest.TestCase):
processes = None
+ CHECK_DICT1 = {
+ 'ROADMA01-SRG1': {
+ 'node_type': 'SRG',
+ 'checks_tp': [({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'}),
+ ({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'})]
+ },
+ 'ROADMA01-SRG3': {
+ 'node_type': 'SRG',
+ 'checks_tp': [({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'}),
+ ({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'})]
+ },
+ 'ROADMA01-DEG1': {
+ 'node_type': 'DEGREE',
+ 'checks_tp': [({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'}),
+ ({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'})]
+ },
+ 'ROADMA01-DEG2': {
+ 'node_type': 'DEGREE',
+ 'checks_tp': [({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'}),
+ ({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'})]
+ }
+ }
+ CHECK_DICT2 = {
+ 'ROADMC01-SRG1': {
+ 'node_type': 'SRG',
+ 'checks_tp': [({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'}),
+ ({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'})]
+ },
+ 'ROADMC01-DEG1': {
+ 'node_type': 'DEGREE',
+ 'checks_tp': [({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'}),
+ ({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'})]
+ },
+ 'ROADMC01-DEG2': {
+ 'node_type': 'DEGREE',
+ 'checks_tp': [({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'}),
+ ({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'})]
+ }
+ }
@classmethod
def setUpClass(cls):
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:model'], '2')
def test_04_getLinks_OpenroadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(nbLink, 10)
- expressLink = ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
- 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX']
+ check_list = {'EXPRESS-LINK': ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX'],
+ 'ADD-LINK': ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX'],
+ 'DROP-LINK': ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX']
+ }
for i in range(0, nbLink):
linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'EXPRESS-LINK':
- find = linkId in expressLink
- self.assertEqual(find, True)
- expressLink.remove(linkId)
- elif res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ADD-LINK':
- find = linkId in addLink
- self.assertEqual(find, True)
- addLink.remove(linkId)
- elif res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'DROP-LINK':
- find = linkId in dropLink
+ linkType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
+ if linkType in check_list:
+ find = linkId in check_list[linkType]
self.assertEqual(find, True)
- dropLink.remove(linkId)
+ (check_list[linkType]).remove(linkId)
else:
self.assertFalse(True)
- self.assertEqual(len(expressLink), 0)
- self.assertEqual(len(addLink), 0)
- self.assertEqual(len(dropLink), 0)
+ for link_type in check_list:
+ self.assertEqual(len(check_list[link_type]), 0)
def test_05_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
res['network'][0]['node'][i]['supporting-node'])
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
nodeId = res['network'][0]['node'][i]['node-id']
- if nodeId == 'ROADMA01-SRG1':
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif nodeId == 'ROADMA01-SRG3':
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
- self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif nodeId == 'ROADMA01-DEG1':
- # Test related to DEG1
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- listNode.remove(nodeId)
- self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
- res['network'][0]['node'][i]['supporting-node'])
- elif nodeId == 'ROADMA01-DEG2':
- # Test related to DEG2
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ if nodeId in self.CHECK_DICT1:
+ self.assertEqual(nodeType, self.CHECK_DICT1[nodeId]['node_type'])
+ if self.CHECK_DICT1[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
+ for item in self.CHECK_DICT1[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
res['network'][0]['node'][i]['supporting-node'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
listNode.remove(nodeId)
else:
self.assertFalse(True)
+
self.assertEqual(len(listNode), 0)
def test_06_connect_XPDRA(self):
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_08_getOpenRoadmNetwork(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['network-ref'], 'clli-network')
self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'], 'NodeA')
nodeId = res['network'][0]['node'][i]['node-id']
- if(nodeId == 'XPDRA01'):
+ if nodeId == 'XPDRA01':
self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'XPONDER')
self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], '1')
- elif(nodeId == 'ROADMA01'):
+ elif nodeId == 'ROADMA01':
self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'ROADM')
self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], '2')
else:
self.assertFalse(True)
def test_09_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
nodeId = res['network'][0]['node'][i]['node-id']
# Tests related to XPDRA nodes
- if(nodeId == 'XPDRA01-XPDR1'):
+ if nodeId == 'XPDRA01-XPDR1':
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDRA01'},
res['network'][0]['node'][i]['supporting-node'])
self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
+ tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j][
+ 'org-openroadm-common-network:tp-type']
tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
- if (tpType == 'XPONDER-CLIENT'):
+ if tpType == 'XPONDER-CLIENT':
client += 1
- elif (tpType == 'XPONDER-NETWORK'):
+ elif tpType == 'XPONDER-NETWORK':
network += 1
- if (tpId == 'XPDR1-NETWORK2'):
+ if tpId == 'XPDR1-NETWORK2':
self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
[j]['transportpce-topology:associated-connection-map-port'], 'XPDR1-CLIENT3')
- if (tpId == 'XPDR1-CLIENT3'):
+ if tpId == 'XPDR1-CLIENT3':
self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
[j]['transportpce-topology:associated-connection-map-port'], 'XPDR1-NETWORK2')
self.assertTrue(client == 4)
self.assertTrue(network == 2)
listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-SRG1'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-SRG3'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
- self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-DEG1'):
- # Test related to DEG1
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
+ # Tests related to ROADMA nodes
+ elif nodeId in self.CHECK_DICT1:
+ self.assertEqual(nodeType, self.CHECK_DICT1[nodeId]['node_type'])
+ if self.CHECK_DICT1[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
+ for item in self.CHECK_DICT1[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-DEG2'):
- # Test related to DEG2
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
res['network'][0]['node'][i]['supporting-node'])
listNode.remove(nodeId)
self.assertEqual(response.status_code, requests.codes.ok)
def test_12_getLinks_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(nbLink, 12)
- expressLink = ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
- 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX']
- XPDR_IN = ['ROADMA01-SRG1-SRG1-PP1-TXRXtoXPDRA01-XPDR1-XPDR1-NETWORK1']
- XPDR_OUT = ['XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX']
+ check_list = {'EXPRESS-LINK': ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX'],
+ 'ADD-LINK': ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX'],
+ 'DROP-LINK': ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX'],
+ 'XPONDER-INPUT': ['ROADMA01-SRG1-SRG1-PP1-TXRXtoXPDRA01-XPDR1-XPDR1-NETWORK1'],
+ 'XPONDER-OUTPUT': ['XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX']
+ }
for i in range(0, nbLink):
- nodeType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if(nodeType == 'EXPRESS-LINK'):
- find = linkId in expressLink
- self.assertEqual(find, True)
- expressLink.remove(linkId)
- elif(nodeType == 'ADD-LINK'):
- find = linkId in addLink
- self.assertEqual(find, True)
- addLink.remove(linkId)
- elif(nodeType == 'DROP-LINK'):
- find = linkId in dropLink
+ linkType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
+ if linkType in check_list:
+ find = linkId in check_list[linkType]
self.assertEqual(find, True)
- dropLink.remove(linkId)
- elif(nodeType == 'XPONDER-INPUT'):
- find = linkId in XPDR_IN
- self.assertEqual(find, True)
- XPDR_IN.remove(linkId)
- elif(nodeType == 'XPONDER-OUTPUT'):
- find = linkId in XPDR_OUT
- self.assertEqual(find, True)
- XPDR_OUT.remove(linkId)
+ (check_list[linkType]).remove(linkId)
else:
self.assertFalse(True)
- self.assertEqual(len(expressLink), 0)
- self.assertEqual(len(addLink), 0)
- self.assertEqual(len(dropLink), 0)
- self.assertEqual(len(XPDR_IN), 0)
- self.assertEqual(len(XPDR_OUT), 0)
+ for link_type in check_list:
+ self.assertEqual(len(check_list[link_type]), 0)
def test_13_connect_ROADMC(self):
response = test_utils.mount_device("ROADMC01", 'roadmc')
self.assertEqual(response.status_code, requests.codes.created)
def test_16_getClliNetwork(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
listNode = ['NodeA', 'NodeC']
for i in range(0, nbNode):
nodeId = res['network'][0]['node'][i]['node-id']
- find = nodeId in listNode
- self.assertEqual(find, True)
- if(nodeId == 'NodeA'):
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeA')
+ if nodeId in listNode:
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], nodeId)
+ listNode.remove(nodeId)
else:
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeC')
- listNode.remove(nodeId)
-
+ self.assertFalse(True)
self.assertEqual(len(listNode), 0)
def test_17_getOpenRoadmNetwork(self):
+ # pylint: disable=redundant-unittest-assert
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(nbNode, 3)
listNode = ['XPDRA01', 'ROADMA01', 'ROADMC01']
+ CHECK_LIST = {'XPDRA01': {'node-ref': 'NodeA', 'node-type': 'XPONDER', 'model': '1'},
+ 'ROADMA01': {'node-ref': 'NodeA', 'node-type': 'ROADM', 'model': '2'},
+ 'ROADMC01': {'node-ref': 'NodeC', 'node-type': 'ROADM', 'model': '2'}
+ }
for i in range(0, nbNode):
self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['network-ref'], 'clli-network')
nodeId = res['network'][0]['node'][i]['node-id']
- if(nodeId == 'XPDRA01'):
- self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'], 'NodeA')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'XPONDER')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], '1')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01'):
- self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'], 'NodeA')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'ROADM')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], '2')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMC01'):
- self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'], 'NodeC')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'ROADM')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], '2')
+ if nodeId in CHECK_LIST:
+ self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'],
+ CHECK_LIST[nodeId]['node-ref'])
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'],
+ CHECK_LIST[nodeId]['node-type'])
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'],
+ CHECK_LIST[nodeId]['model'])
listNode.remove(nodeId)
else:
self.assertFalse(True)
self.assertEqual(len(listNode), 0)
def test_18_getROADMLinkOpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(nbLink, 20)
- expressLink = ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX', 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
- 'ROADMC01-DEG2-DEG2-CTP-TXRXtoROADMC01-DEG1-DEG1-CTP-TXRX', 'ROADMC01-DEG1-DEG1-CTP-TXRXtoROADMC01-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMC01-SRG1-SRG1-CP-TXRXtoROADMC01-DEG2-DEG2-CTP-TXRX', 'ROADMC01-SRG1-SRG1-CP-TXRXtoROADMC01-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
- 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
- 'ROADMC01-DEG1-DEG1-CTP-TXRXtoROADMC01-SRG1-SRG1-CP-TXRX', 'ROADMC01-DEG2-DEG2-CTP-TXRXtoROADMC01-SRG1-SRG1-CP-TXRX']
- R2RLink = ['ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX',
- 'ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX']
- XPDR_IN = ['ROADMA01-SRG1-SRG1-PP1-TXRXtoXPDRA01-XPDR1-XPDR1-NETWORK1']
- XPDR_OUT = ['XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX']
+ check_list = {'EXPRESS-LINK': ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMC01-DEG2-DEG2-CTP-TXRXtoROADMC01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMC01-DEG1-DEG1-CTP-TXRXtoROADMC01-DEG2-DEG2-CTP-TXRX'],
+ 'ADD-LINK': ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMC01-SRG1-SRG1-CP-TXRXtoROADMC01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMC01-SRG1-SRG1-CP-TXRXtoROADMC01-DEG1-DEG1-CTP-TXRX'],
+ 'DROP-LINK': ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
+ 'ROADMC01-DEG1-DEG1-CTP-TXRXtoROADMC01-SRG1-SRG1-CP-TXRX',
+ 'ROADMC01-DEG2-DEG2-CTP-TXRXtoROADMC01-SRG1-SRG1-CP-TXRX'],
+ 'ROADM-TO-ROADM': ['ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX',
+ 'ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX'],
+ 'XPONDER-INPUT': ['ROADMA01-SRG1-SRG1-PP1-TXRXtoXPDRA01-XPDR1-XPDR1-NETWORK1'],
+ 'XPONDER-OUTPUT': ['XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX']
+ }
for i in range(0, nbLink):
- nodeType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if(nodeType == 'EXPRESS-LINK'):
- find = linkId in expressLink
- self.assertEqual(find, True)
- expressLink.remove(linkId)
- elif(nodeType == 'ADD-LINK'):
- find = linkId in addLink
- self.assertEqual(find, True)
- addLink.remove(linkId)
- elif(nodeType == 'DROP-LINK'):
- find = linkId in dropLink
+ linkType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
+ if linkType in check_list:
+ find = linkId in check_list[linkType]
self.assertEqual(find, True)
- dropLink.remove(linkId)
- elif(nodeType == 'ROADM-TO-ROADM'):
- find = linkId in R2RLink
- self.assertEqual(find, True)
- R2RLink.remove(linkId)
- elif(nodeType == 'XPONDER-INPUT'):
- find = linkId in XPDR_IN
- self.assertEqual(find, True)
- XPDR_IN.remove(linkId)
- elif(nodeType == 'XPONDER-OUTPUT'):
- find = linkId in XPDR_OUT
- self.assertEqual(find, True)
- XPDR_OUT.remove(linkId)
+ (check_list[linkType]).remove(linkId)
else:
self.assertFalse(True)
- self.assertEqual(len(expressLink), 0)
- self.assertEqual(len(addLink), 0)
- self.assertEqual(len(dropLink), 0)
- self.assertEqual(len(R2RLink), 0)
- self.assertEqual(len(XPDR_IN), 0)
- self.assertEqual(len(XPDR_OUT), 0)
+ for link_type in check_list:
+ self.assertEqual(len(check_list[link_type]), 0)
def test_19_getLinkOmsAttributesOpenRoadmTopology(self):
response = test_utils.get_ordm_topo_request("")
R2RLink = ['ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX',
'ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX']
for i in range(0, nbLink):
- nodeType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if(link_id in R2RLink):
+ if link_id in R2RLink:
find = False
- spanLoss = res['network'][0]['ietf-network-topology:link'][i][
- 'org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"]
- length = res['network'][0]['ietf-network-topology:link'][i][
- 'org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]['SRLG-length']
- if((spanLoss != None) & (length != None)):
+ spanLoss = (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"])
+ length = (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]
+ ['SRLG-length'])
+ if (spanLoss is not None) & (length is not None):
find = True
self.assertTrue(find)
R2RLink.remove(link_id)
self.assertEqual(len(R2RLink), 0)
def test_20_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
listNode = ['XPDRA01-XPDR1',
'ROADMA01-SRG1', 'ROADMA01-SRG3', 'ROADMA01-DEG1', 'ROADMA01-DEG2',
'ROADMC01-SRG1', 'ROADMC01-DEG1', 'ROADMC01-DEG2']
- # ************************Tests related to XPDRA nodes
for i in range(0, nbNode):
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
nodeId = res['network'][0]['node'][i]['node-id']
- if(nodeId == 'XPDRA01-XPDR1'):
+ # Tests related to XPDRA nodes
+ if nodeId == 'XPDRA01-XPDR1':
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDRA01'},
res['network'][0]['node'][i]['supporting-node'])
self.assertEqual(nodeType, 'XPONDER')
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
- if (tpType == 'XPONDER-CLIENT'):
+ tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j][
+ 'org-openroadm-common-network:tp-type']
+ if tpType == 'XPONDER-CLIENT':
client += 1
- elif (tpType == 'XPONDER-NETWORK'):
+ elif tpType == 'XPONDER-NETWORK':
network += 1
self.assertTrue(client == 4)
self.assertTrue(network == 2)
listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-SRG1'):
- # Test related to SRG1
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'SRG')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-SRG3'):
- # Test related to SRG1
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
- self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'SRG')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-DEG1'):
- # Test related to DEG1
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
+ # Tests related to ROADMA nodes
+ elif nodeId in self.CHECK_DICT1:
+ self.assertEqual(nodeType, self.CHECK_DICT1[nodeId]['node_type'])
+ if self.CHECK_DICT1[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
+ for item in self.CHECK_DICT1[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'DEGREE')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-DEG2'):
- # Test related to DEG2
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'DEGREE')
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'],
+ self.CHECK_DICT1[nodeId]['node_type'])
listNode.remove(nodeId)
- elif(nodeId == 'ROADMC01-SRG1'):
- # Test related to SRG1
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMC01'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'SRG')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMC01-DEG1'):
- # Test related to DEG1
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMC01'},
+ # Tests related to ROADMC nodes
+ elif nodeId in self.CHECK_DICT2:
+ self.assertEqual(nodeType, self.CHECK_DICT2[nodeId]['node_type'])
+ if self.CHECK_DICT2[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
+ for item in self.CHECK_DICT2[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeC'},
res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'DEGREE')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMC01-DEG2'):
- # Test related to DEG2
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMC01'},
res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'DEGREE')
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'],
+ self.CHECK_DICT2[nodeId]['node_type'])
listNode.remove(nodeId)
else:
self.assertFalse(True)
self.assertEqual(response.status_code, requests.codes.created)
def test_26_getClliNetwork(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
listNode = ['NodeA', 'NodeB', 'NodeC']
for i in range(0, nbNode):
nodeId = res['network'][0]['node'][i]['node-id']
- find = nodeId in listNode
- self.assertEqual(find, True)
- if(nodeId == 'NodeA'):
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeA')
- elif(nodeId == 'NodeB'):
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeB')
+ if nodeId in listNode:
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], nodeId)
+ listNode.remove(nodeId)
else:
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeC')
- listNode.remove(nodeId)
+ self.assertFalse(True)
self.assertEqual(len(listNode), 0)
def test_27_verifyDegree(self):
res = response.json()
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
- listR2RLink = ['ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX', 'ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX',
- 'ROADMA01-DEG2-DEG2-TTP-TXRXtoROADMB01-DEG1-DEG1-TTP-TXRX', 'ROADMC01-DEG1-DEG1-TTP-TXRXtoROADMB01-DEG2-DEG2-TTP-TXRX',
- 'ROADMB01-DEG1-DEG1-TTP-TXRXtoROADMA01-DEG2-DEG2-TTP-TXRX', 'ROADMB01-DEG2-DEG2-TTP-TXRXtoROADMC01-DEG1-DEG1-TTP-TXRX']
+ listR2RLink = ['ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX',
+ 'ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX',
+ 'ROADMA01-DEG2-DEG2-TTP-TXRXtoROADMB01-DEG1-DEG1-TTP-TXRX',
+ 'ROADMC01-DEG1-DEG1-TTP-TXRXtoROADMB01-DEG2-DEG2-TTP-TXRX',
+ 'ROADMB01-DEG1-DEG1-TTP-TXRXtoROADMA01-DEG2-DEG2-TTP-TXRX',
+ 'ROADMB01-DEG2-DEG2-TTP-TXRXtoROADMC01-DEG1-DEG1-TTP-TXRX']
for i in range(0, nbLink):
- if res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ROADM-TO-ROADM':
+ if (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'ROADM-TO-ROADM'):
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
find = link_id in listR2RLink
self.assertEqual(find, True)
link_type = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
link_src = res['network'][0]['ietf-network-topology:link'][i]['source']['source-node']
link_dest = res['network'][0]['ietf-network-topology:link'][i]['destination']['dest-node']
- oppLink_id = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:opposite-link']
+ oppLink_id = (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:opposite-link'])
# Find the opposite link
response_oppLink = test_utils.get_ordm_topo_request("ietf-network-topology:link/"+oppLink_id)
self.assertEqual(response_oppLink.status_code, requests.codes.ok)
self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['source']['source-node'], link_dest)
self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['destination']['dest-node'], link_src)
oppLink_type = res_oppLink['ietf-network-topology:link'][0]['org-openroadm-common-network:link-type']
- if link_type == 'ADD-LINK':
- self.assertEqual(oppLink_type, 'DROP-LINK')
- elif link_type == 'DROP-LINK':
- self.assertEqual(oppLink_type, 'ADD-LINK')
- elif link_type == 'EXPRESS-LINK':
- self.assertEqual(oppLink_type, 'EXPRESS-LINK')
- elif link_type == 'ROADM-TO-ROADM':
- self.assertEqual(oppLink_type, 'ROADM-TO-ROADM')
- elif link_type == 'XPONDER-INPUT':
- self.assertEqual(oppLink_type, 'XPONDER-OUTPUT')
- elif link_type == 'XPONDER-OUTPUT':
- self.assertEqual(oppLink_type, 'XPONDER-INPUT')
+ CHECK_DICT = {'ADD-LINK': 'DROP-LINK', 'DROP-LINK': 'ADD-LINK',
+ 'EXPRESS-LINK': 'EXPRESS-LINK', 'ROADM-TO-ROADM': 'ROADM-TO-ROADM',
+ 'XPONDER-INPUT': 'XPONDER-OUTPUT', 'XPONDER-OUTUT': 'XPONDER-INPUT'}
+ if link_type in CHECK_DICT:
+ self.assertEqual(oppLink_type, CHECK_DICT[link_type])
def test_29_getLinkOmsAttributesOpenRoadmTopology(self):
response = test_utils.get_ordm_topo_request("")
'ROADMB01-DEG1-DEG1-TTP-TXRXtoROADMA01-DEG2-DEG2-TTP-TXRX',
'ROADMB01-DEG2-DEG2-TTP-TXRXtoROADMC01-DEG1-DEG1-TTP-TXRX']
for i in range(0, nbLink):
- nodeType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if(link_id in R2RLink):
+ if link_id in R2RLink:
find = False
spanLoss = res['network'][0]['ietf-network-topology:link'][i][
'org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"]
length = res['network'][0]['ietf-network-topology:link'][i][
'org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]['SRLG-length']
- if((spanLoss != None) & (length != None)):
+ if (spanLoss is not None) & (length is not None):
find = True
self.assertTrue(find)
R2RLink.remove(link_id)
self.assertEqual(response.status_code, requests.codes.ok)
def test_32_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
nodeId = res['network'][0]['node'][i]['node-id']
# Tests related to XPDRA nodes
- if(nodeId == 'XPDRA01-XPDR1'):
+ if nodeId == 'XPDRA01-XPDR1':
nbTp = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
for j in range(0, nbTp):
tpid = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
- if (tpid == 'XPDR1-CLIENT1'):
+ if tpid == 'XPDR1-CLIENT1':
self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
['org-openroadm-common-network:tp-type'], 'XPONDER-CLIENT')
- if (tpid == 'XPDR1-NETWORK1'):
+ if tpid == 'XPDR1-NETWORK1':
self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
['org-openroadm-common-network:tp-type'], 'XPONDER-NETWORK')
self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-network-topology:xpdr-network-attributes']['tail-equipment-id'],
+ ['org-openroadm-network-topology:xpdr-network-attributes']
+ ['tail-equipment-id'],
'ROADMA01-SRG1--SRG1-PP1-TXRX')
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDRA01'},
res['network'][0]['node'][i]['supporting-node'])
listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-SRG1'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-SRG3'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
- self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-DEG1'):
- # Test related to DEG1
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
+ # Tests related to ROADMA nodes
+ elif nodeId in self.CHECK_DICT1:
+ self.assertEqual(nodeType, self.CHECK_DICT1[nodeId]['node_type'])
+ if self.CHECK_DICT1[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
+ for item in self.CHECK_DICT1[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-DEG2'):
- # Test related to DEG2
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
res['network'][0]['node'][i]['supporting-node'])
listNode.remove(nodeId)
nbNode = len(res['network'][0]['node'])
self.assertEqual(nbNode, 1)
for i in range(0, nbNode-1):
- self.assertNotEqual(res['network'][0]['node'][1]['org-openroadm-clli-network:clli'], 'NodeC')
+ self.assertNotEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeC')
def test_35_disconnect_XPDRA(self):
response = test_utils.unmount_device("XPDRA01")
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'XPDRA01')
def test_38_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
res['network'][0]['node'][i]['supporting-node'])
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
nodeId = res['network'][0]['node'][i]['node-id']
- if(nodeId == 'ROADMA01-SRG1'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-SRG3'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
- self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-DEG1'):
- # Test related to DEG1
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADMA01-DEG2'):
- # Test related to DEG2
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ if nodeId in self.CHECK_DICT1:
+ self.assertEqual(nodeType, self.CHECK_DICT1[nodeId]['node_type'])
+ if self.CHECK_DICT1[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 17)
+ for item in self.CHECK_DICT1[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
+ res['network'][0]['node'][i]['supporting-node'])
+ self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADMA01'},
+ res['network'][0]['node'][i]['supporting-node'])
listNode.remove(nodeId)
else:
self.assertFalse(True)
res = response.json()
nbLink = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(nbLink, 16)
- expressLink = ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
- 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX', 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
- 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX', 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX']
+ check_list = {'EXPRESS-LINK': ['ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX'],
+ 'ADD-LINK': ['ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG1-SRG1-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG2-DEG2-CTP-TXRX',
+ 'ROADMA01-SRG3-SRG3-CP-TXRXtoROADMA01-DEG1-DEG1-CTP-TXRX'],
+ 'DROP-LINK': ['ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG1-SRG1-CP-TXRX',
+ 'ROADMA01-DEG1-DEG1-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX',
+ 'ROADMA01-DEG2-DEG2-CTP-TXRXtoROADMA01-SRG3-SRG3-CP-TXRX']
+ }
roadmtoroadmLink = 0
for i in range(0, nbLink):
- if (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'EXPRESS-LINK'):
- link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- find = link_id in expressLink
- self.assertEqual(find, True)
- expressLink.remove(link_id)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ADD-LINK'):
- link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- find = link_id in addLink
- self.assertEqual(find, True)
- addLink.remove(link_id)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'DROP-LINK'):
- link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- find = link_id in dropLink
+ linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id']
+ linkType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
+ if linkType in check_list:
+ find = linkId in check_list[linkType]
self.assertEqual(find, True)
- dropLink.remove(link_id)
+ (check_list[linkType]).remove(linkId)
else:
roadmtoroadmLink += 1
- self.assertEqual(len(expressLink), 0)
- self.assertEqual(len(addLink), 0)
- self.assertEqual(len(dropLink), 0)
+ for link_type in check_list:
+ self.assertEqual(len(check_list[link_type]), 0)
self.assertEqual(roadmtoroadmLink, 6)
for i in range(0, nbLink):
self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
import unittest
-import json
import time
import requests
from common import test_utils
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
for ele in liste_tp:
if ele[u'org-openroadm-common-network:tp-type'] == 'XPONDER-CLIENT':
self.assertNotIn('org-openroadm-network-topology:xpdr-client-attributes', dict.keys(ele))
- elif (ele[u'org-openroadm-common-network:tp-type'] == 'XPONDER-NETWORK'):
+ elif ele[u'org-openroadm-common-network:tp-type'] == 'XPONDER-NETWORK':
self.assertIn(u'tail-equipment-id',
dict.keys(ele[u'org-openroadm-network-topology:xpdr-network-attributes']))
self.assertNotIn('wavelength', dict.keys(
# http://www.apache.org/licenses/LICENSE-2.0
#############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
import time
-import json
import requests
from common import test_utils
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertIn('Request processed', res["output"]["result"])
time.sleep(10)
- """to test case where SRG where the xpdr is connected to has no optical range data"""
+ #"""to test case where SRG where the xpdr is connected to has no optical range data"""
def test_31_connect_xprdA_to_roadmA(self):
response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2",
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
import time
import requests
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertEqual(nb_links, 4)
for link in res['network'][0]['ietf-network-topology:link']:
linkId = link['link-id']
- if (linkId == 'OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
- linkId == 'OTU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+ if (linkId in ('OTU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+ 'OTU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 0)
self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 100000)
- elif (linkId == 'ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
- linkId == 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+ elif (linkId in ('ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+ 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 100000)
self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 0)
self.assertEqual(link['transportpce-topology:otn-link-type'], 'ODTU4')
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'):
+ if node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1':
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
- if (tp['tp-id'] == 'XPDR1-NETWORK1'):
+ if tp['tp-id'] == 'XPDR1-NETWORK1':
xpdrTpPortConAt = tp['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
self.assertEqual(len(xpdrTpPortConAt['ts-pool']), 80)
self.assertEqual(len(xpdrTpPortConAt['odtu-tpn-pool'][0]['tpn-pool']), 80)
self.assertEqual(nb_links, 4)
for link in res['network'][0]['ietf-network-topology:link']:
linkId = link['link-id']
- if (linkId == 'ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
- linkId == 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+ if (linkId in ('ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+ 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 90000)
self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 10000)
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'):
+ if node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1':
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
- if (tp['tp-id'] == 'XPDR1-NETWORK1'):
+ if tp['tp-id'] == 'XPDR1-NETWORK1':
xpdrTpPortConAt = tp['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
self.assertEqual(len(xpdrTpPortConAt['ts-pool']), 72)
- tsPoolList = [i for i in range(1, 9)]
+ tsPoolList = list(range(1, 9))
self.assertNotIn(tsPoolList, xpdrTpPortConAt['ts-pool'])
self.assertEqual(len(xpdrTpPortConAt['odtu-tpn-pool'][0]['tpn-pool']), 79)
self.assertNotIn(1, xpdrTpPortConAt['odtu-tpn-pool'][0]['tpn-pool'])
self.assertEqual(nb_links, 4)
for link in res['network'][0]['ietf-network-topology:link']:
linkId = link['link-id']
- if (linkId == 'ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1' or
- linkId == 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1'):
+ if (linkId in ('ODU4-SPDR-SA1-XPDR1-XPDR1-NETWORK1toSPDR-SC1-XPDR1-XPDR1-NETWORK1',
+ 'ODU4-SPDR-SC1-XPDR1-XPDR1-NETWORK1toSPDR-SA1-XPDR1-XPDR1-NETWORK1')):
self.assertEqual(link['org-openroadm-otn-network-topology:available-bandwidth'], 100000)
self.assertEqual(link['org-openroadm-otn-network-topology:used-bandwidth'], 0)
if (node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1'):
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
- if (tp['tp-id'] == 'XPDR1-NETWORK1'):
+ if tp['tp-id'] == 'XPDR1-NETWORK1':
xpdrTpPortConAt = tp['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
self.assertEqual(len(xpdrTpPortConAt['ts-pool']), 80)
self.assertEqual(len(xpdrTpPortConAt['odtu-tpn-pool'][0]['tpn-pool']), 80)
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'):
+ if node['node-id'] == 'SPDR-SA1-XPDR1' or 'SPDR-SC1-XPDR1':
tpList = node['ietf-network-topology:termination-point']
for tp in tpList:
- if (tp['tp-id'] == 'XPDR1-NETWORK1'):
+ if tp['tp-id'] == 'XPDR1-NETWORK1':
xpdrTpPortConAt = tp['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
self.assertNotIn('ts-pool', dict.keys(xpdrTpPortConAt))
self.assertNotIn('odtu-tpn-pool', dict.keys(xpdrTpPortConAt))
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
-import json
import time
import requests
from common import test_utils
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
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 = 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'])
+ res_mapping = (response.json())['mapping'][0]
+ self.assertEqual('CP1-SFP4-P1', res_mapping['supporting-port'])
+ self.assertEqual('CP1-SFP4', res_mapping['supporting-circuit-pack-name'])
+ self.assertEqual('XPDR1-CLIENT4', res_mapping['logical-connection-point'])
+ self.assertEqual('bidirectional', res_mapping['port-direction'])
+ self.assertEqual('xpdr-client', res_mapping['port-qual'])
+ self.assertEqual('FqlcrxV7p3g=', res_mapping['lcp-hash-val'])
+ self.assertIn('org-openroadm-port-types:if-10GE-ODU2e', res_mapping['supported-interface-capability'])
+ self.assertIn('org-openroadm-port-types:if-10GE-ODU2', res_mapping['supported-interface-capability'])
+ self.assertIn('org-openroadm-port-types:if-10GE', res_mapping['supported-interface-capability'])
def test_03_get_portmapping_NETWORK1(self):
response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-NETWORK1")
response = test_utils.otn_service_path_request("create", "service1", "10G", "Ethernet",
[{"node-id": "SPDR-SA1", "client-tp": "XPDR1-CLIENT4",
"network-tp": "XPDR1-NETWORK1"}],
- {"ethernet-encoding": "eth encode", "trib-slot": ["1"], "trib-port-number": "1"})
+ {"ethernet-encoding": "eth encode",
+ "trib-slot": ["1"], "trib-port-number": "1"})
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertTrue(res["output"]["success"])
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'])
+ 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'])
response = test_utils.otn_service_path_request("delete", "service1", "10G", "Ethernet",
[{"node-id": "SPDR-SA1", "client-tp": "XPDR1-CLIENT4",
"network-tp": "XPDR1-NETWORK1"}],
- {"ethernet-encoding": "eth encode", "trib-slot": ["1"], "trib-port-number": "1"})
+ {"ethernet-encoding": "eth encode",
+ "trib-slot": ["1"], "trib-port-number": "1"})
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
import time
import requests
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
import time
import logging
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertNotIn('ietf-network-topology:link', res['network'][0])
def test_05_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
self.assertIn('node', res['network'][0])
- if ('node' in res['network'][0]):
+ if 'node' in res['network'][0]:
nbNode = len(res['network'][0]['node'])
self.assertEqual(nbNode, 2)
listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2']
for i in range(0, nbNode):
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
nodeId = res['network'][0]['node'][i]['node-id']
- if (nodeId == 'SPDR-SA1-XPDR1'):
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(nodeType, 'XPONDER')
- nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- client = 0
- network = 0
- for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
- tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
- if (tpType == 'XPONDER-CLIENT'):
- client += 1
- elif (tpType == 'XPONDER-NETWORK'):
- network += 1
- self.assertTrue(client == 0)
+ if nodeId not in listNode:
+ self.assertFalse(True)
+ continue
+ self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
+ res['network'][0]['node'][i]['supporting-node'])
+ self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
+ res['network'][0]['node'][i]['supporting-node'])
+ self.assertEqual(nodeType, 'XPONDER')
+ nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ client = 0
+ network = 0
+ for j in range(0, nbTps):
+ tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type'])
+ if tpType == 'XPONDER-CLIENT':
+ client += 1
+ elif tpType == 'XPONDER-NETWORK':
+ network += 1
+ self.assertTrue(client == 0)
+ if nodeId == 'SPDR-SA1-XPDR1':
self.assertTrue(network == 1)
- listNode.remove(nodeId)
- elif (nodeId == 'SPDR-SA1-XPDR2'):
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(nodeType, 'XPONDER')
- nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- client = 0
- network = 0
- for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
- tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
- if (tpType == 'XPONDER-CLIENT'):
- client += 1
- elif (tpType == 'XPONDER-NETWORK'):
- network += 1
- self.assertTrue(client == 0)
- self.assertTrue(network == 4)
- listNode.remove(nodeId)
else:
- self.assertFalse(True)
+ # elif nodeId == 'SPDR-SA1-XPDR2':
+ self.assertTrue(network == 4)
+ listNode.remove(nodeId)
self.assertEqual(len(listNode), 0)
def test_06_getLinks_OtnTopology(self):
self.assertNotIn('ietf-network-topology:link', res['network'][0])
def test_07_getNodes_OtnTopology(self):
+ # pylint: disable=redundant-unittest-assert
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(nbNode, 2)
listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2']
+ CHECK_LIST = {
+ 'SPDR-SA1-XPDR1': {
+ 'node-type': 'MUXPDR',
+ 'xpdr-number': 1,
+ 'port-types': 'org-openroadm-port-types:if-10GE-ODU2e',
+ 'otn-common-types': 'org-openroadm-otn-common-types:ODU2e',
+ 'network_nb': 1,
+ 'nbl_nb': 4,
+ 'tp-checklist': ['XPDR1-NETWORK1', 'XPDR1-CLIENT1'],
+ 'tp-unchecklist': ['XPDR1-CLIENT2']
+ },
+ 'SPDR-SA1-XPDR2': {
+ 'node-type': 'SWITCH',
+ 'xpdr-number': 2,
+ 'port-types': 'org-openroadm-port-types:if-100GE-ODU4',
+ 'otn-common-types': 'org-openroadm-otn-common-types:ODU4',
+ 'network_nb': 4,
+ 'nbl_nb': 1,
+ 'tp-checklist': ['XPDR2-NETWORK4', 'XPDR2-CLIENT1', 'XPDR2-NETWORK3', 'XPDR2-CLIENT4',
+ 'XPDR2-CLIENT2', 'XPDR2-NETWORK2', 'XPDR2-CLIENT3', 'XPDR2-NETWORK1'],
+ 'tp-unchecklist': []
+ }
+ }
for i in range(0, nbNode):
nodeId = res['network'][0]['node'][i]['node-id']
- if(nodeId == 'SPDR-SA1-XPDR1'):
+ if nodeId in CHECK_LIST:
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
- self.assertEqual(nodeType, 'MUXPDR')
+ self.assertEqual(nodeType, CHECK_LIST[nodeId]['node-type'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
res['network'][0]['node'][i]['supporting-node'])
- self.assertIn({'network-ref': 'openroadm-topology', 'node-ref': 'SPDR-SA1-XPDR1'},
+ self.assertIn({'network-ref': 'openroadm-topology', 'node-ref': nodeId},
res['network'][0]['node'][i]['supporting-node'])
self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
res['network'][0]['node'][i]['supporting-node'])
self.assertEqual(res['network'][0]['node'][i]
- ['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'], 1)
+ ['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'],
+ CHECK_LIST[nodeId]['xpdr-number'])
nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
+ tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type'])
tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
- if (tpType == 'XPONDER-CLIENT'):
+ if tpType == 'XPONDER-CLIENT':
client += 1
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
- ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-10GE-ODU2e')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU2e')
- elif (tpType == 'XPONDER-NETWORK'):
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:tp-supported-interfaces']
+ ['supported-interface-capability'][0]['if-cap-type']),
+ CHECK_LIST[nodeId]['port-types'])
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
+ ['rate']),
+ CHECK_LIST[nodeId]['otn-common-types'])
+ elif tpType == 'XPONDER-NETWORK':
network += 1
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
- ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-OCH-OTU4-ODU4')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR1')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['tp-ref'], 'XPDR1-NETWORK1')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:tp-supported-interfaces']
+ ['supported-interface-capability'][0]['if-cap-type']),
+ 'org-openroadm-port-types:if-OCH-OTU4-ODU4')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
+ ['rate']),
+ 'org-openroadm-otn-common-types:ODU4')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['supporting-termination-point'][0]['network-ref']), 'openroadm-topology')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['supporting-termination-point'][0]['node-ref']), nodeId)
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['supporting-termination-point'][0]['tp-ref']), tpId)
self.assertTrue(client == 4)
- self.assertTrue(network == 1)
+ self.assertTrue(network == CHECK_LIST[nodeId]['network_nb'])
listNode.remove(nodeId)
nbNbl = len(res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']
['odu-switching-pools'][0]['non-blocking-list'])
- self.assertEqual(nbNbl, 4)
+ self.assertEqual(nbNbl, CHECK_LIST[nodeId]['nbl_nb'])
for k in range(0, nbNbl):
- nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][k]
- if (nbl['nbl-number'] == 1):
- self.assertEqual(nbl['available-interconnect-bandwidth'], 10)
- self.assertEqual(nbl['interconnect-bandwidth-unit'], 1000000000)
- self.assertIn('XPDR1-NETWORK1', nbl['tp-list'])
- self.assertIn('XPDR1-CLIENT1', nbl['tp-list'])
- self.assertNotIn('XPDR1-CLIENT2', nbl['tp-list'])
- elif(nodeId == 'SPDR-SA1-XPDR2'):
- nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
- self.assertEqual(nodeType, 'SWITCH')
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertIn({'network-ref': 'openroadm-topology', 'node-ref': 'SPDR-SA1-XPDR2'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]
- ['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'], 2)
- nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- client = 0
- network = 0
- for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
- tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
- if (tpType == 'XPONDER-CLIENT'):
- client += 1
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
- ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-100GE-ODU4')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
- elif (tpType == 'XPONDER-NETWORK'):
- network += 1
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
- ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-OCH-OTU4-ODU4')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR2')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
- [j]['supporting-termination-point'][0]['tp-ref'], tpId)
- self.assertTrue(client == 4)
- self.assertTrue(network == 4)
- listNode.remove(nodeId)
- nbNbl = len(res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']
- ['odu-switching-pools'][0]['non-blocking-list'])
- self.assertEqual(nbNbl, 1)
- nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][0]
- self.assertIn('XPDR2-NETWORK1', nbl['tp-list'])
- self.assertIn('XPDR2-CLIENT1', nbl['tp-list'])
- self.assertIn('XPDR2-NETWORK2', nbl['tp-list'])
- self.assertIn('XPDR2-CLIENT2', nbl['tp-list'])
- self.assertIn('XPDR2-NETWORK3', nbl['tp-list'])
- self.assertIn('XPDR2-CLIENT3', nbl['tp-list'])
- self.assertIn('XPDR2-NETWORK4', nbl['tp-list'])
- self.assertIn('XPDR2-CLIENT4', nbl['tp-list'])
+ nbl = (res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']
+ ['odu-switching-pools'][0]['non-blocking-list'][k])
+ if nbl['nbl-number'] == 1:
+ if nodeId == 'SPDR-SA1-XPDR1':
+ self.assertEqual(nbl['available-interconnect-bandwidth'], 10)
+ self.assertEqual(nbl['interconnect-bandwidth-unit'], 1000000000)
+ for tp in CHECK_LIST[nodeId]['tp-checklist']:
+ self.assertIn(tp, nbl['tp-list'])
+ for tp in CHECK_LIST[nodeId]['tp-unchecklist']:
+ self.assertNotIn(tp, nbl['tp-list'])
else:
self.assertFalse(True)
self.assertEqual(len(listNode), 0)
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
import time
import requests
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
# http://www.apache.org/licenses/LICENSE-2.0
#############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
-import json
#from unittest.result import failfast
import requests
from common import test_utils
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
def test_05_service_path_create(self):
response = test_utils.service_path_request("create", "service_test", "7",
[{"renderer:node-id": "ROADM-A1",
- "renderer:src-tp": "SRG1-PP3-TXRX", "renderer:dest-tp": "DEG1-TTP-TXRX"},
+ "renderer:src-tp": "SRG1-PP3-TXRX",
+ "renderer:dest-tp": "DEG1-TTP-TXRX"},
{"renderer:node-id": "XPDR-A1",
- "renderer:src-tp": "XPDR1-CLIENT1", "renderer:dest-tp": "XPDR1-NETWORK1"}])
+ "renderer:src-tp": "XPDR1-CLIENT1",
+ "renderer:dest-tp": "XPDR1-NETWORK1"}])
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes: ROADM-A1', res["output"]["result"])
def test_17_service_path_delete(self):
response = test_utils.service_path_request("delete", "service_test", "7",
[{"renderer:node-id": "ROADM-A1",
- "renderer:src-tp": "SRG1-PP3-TXRX", "renderer:dest-tp": "DEG1-TTP-TXRX"},
+ "renderer:src-tp": "SRG1-PP3-TXRX",
+ "renderer:dest-tp": "DEG1-TTP-TXRX"},
{"renderer:node-id": "XPDR-A1",
- "renderer:src-tp": "XPDR1-CLIENT1", "renderer:dest-tp": "XPDR1-NETWORK1"}])
+ "renderer:src-tp": "XPDR1-CLIENT1",
+ "renderer:dest-tp": "XPDR1-NETWORK1"}])
self.assertEqual(response.status_code, requests.codes.ok)
self.assertEqual(response.json(), {
'output': {'result': 'Request processed', 'success': True}})
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+# some pylint false positives specific to tapi test
+# pylint: disable=unsubscriptable-object
+# pylint: disable=unsupported-assignment-operation
+
import os
import sys
import time
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+
import unittest
-import json
import time
import requests
from common import test_utils
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
nbMapCurrent = 0
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)):
+ if (not "CP" in tpId) and (not "CTP" in tpId):
responseMap = test_utils.portmapping_request(nodeMapId+"/mapping/"+tpId)
self.assertEqual(responseMap.status_code, requests.codes.ok)
- if(responseMap.status_code == requests.codes.ok):
+ if responseMap.status_code == requests.codes.ok:
nbMapCurrent += 1
nbMapCumul += nbMapCurrent
nbMappings -= nbMapCurrent
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+# pylint: disable=no-member
+# pylint: disable=too-many-public-methods
+# pylint: disable=too-many-lines
+
import unittest
import time
-import json
import logging
import requests
from common import test_utils
class TransportPCEtesting(unittest.TestCase):
processes = None
+ CHECK_DICT1 = {
+ 'ROADM-A1-SRG1': {
+ 'node_type': 'SRG',
+ 'checks_tp': [({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'}),
+ ({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'})]
+ },
+ 'ROADM-A1-SRG3': {
+ 'node_type': 'SRG',
+ 'checks_tp': [({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'}),
+ ({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'})]
+ },
+ 'ROADM-A1-DEG1': {
+ 'node_type': 'DEGREE',
+ 'checks_tp': [({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'}),
+ ({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'})]
+ },
+ 'ROADM-A1-DEG2': {
+ 'node_type': 'DEGREE',
+ 'checks_tp': [({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'}),
+ ({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'})]
+ }
+ }
+ CHECK_DICT2 = {
+ 'ROADM-C1-SRG1': {
+ 'node_type': 'SRG',
+ 'checks_tp': [({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'}),
+ ({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'})]
+ },
+ 'ROADM-C1-DEG1': {
+ 'node_type': 'DEGREE',
+ 'checks_tp': [({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'}),
+ ({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'})]
+ },
+ 'ROADM-C1-DEG2': {
+ 'node_type': 'DEGREE',
+ 'checks_tp': [({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'}),
+ ({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'})]
+ }
+ }
@classmethod
def setUpClass(cls):
@classmethod
def tearDownClass(cls):
+ # pylint: disable=not-an-iterable
for process in cls.processes:
test_utils.shutdown_process(process)
print("all processes killed")
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:model'], 'model2')
def test_04_getLinks_OpenroadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(nbLink, 10)
- expressLink = ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
- 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
- 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
- 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
- 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
- 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX']
+ check_list = {'EXPRESS-LINK': ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX'],
+ 'ADD-LINK': ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX'],
+ 'DROP-LINK': ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX']
+ }
for i in range(0, nbLink):
linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'EXPRESS-LINK'):
- find = linkId in expressLink
- self.assertEqual(find, True)
- expressLink.remove(linkId)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ADD-LINK'):
- find = linkId in addLink
- self.assertEqual(find, True)
- addLink.remove(linkId)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'DROP-LINK'):
- find = linkId in dropLink
+ linkType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
+ if linkType in check_list:
+ find = linkId in check_list[linkType]
self.assertEqual(find, True)
- dropLink.remove(linkId)
+ (check_list[linkType]).remove(linkId)
else:
self.assertFalse(True)
- self.assertEqual(len(expressLink), 0)
- self.assertEqual(len(addLink), 0)
- self.assertEqual(len(dropLink), 0)
+ for link_type in check_list:
+ self.assertEqual(len(check_list[link_type]), 0)
def test_05_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
res['network'][0]['node'][i]['supporting-node'])
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
nodeId = res['network'][0]['node'][i]['node-id']
- if(nodeId == 'ROADM-A1-SRG1'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-SRG3'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
- self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-DEG1'):
- # Test related to DEG1
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-DEG2'):
- # Test related to DEG2
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ if nodeId in self.CHECK_DICT1:
+ self.assertEqual(nodeType, self.CHECK_DICT1[nodeId]['node_type'])
+ if self.CHECK_DICT1[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
+ for item in self.CHECK_DICT1[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
res['network'][0]['node'][i]['supporting-node'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_08_getOpenRoadmNetwork(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['network-ref'], 'clli-network')
self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'], 'NodeA')
nodeId = res['network'][0]['node'][i]['node-id']
- if(nodeId == 'XPDR-A1'):
+ if nodeId == 'XPDR-A1':
self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'XPONDER')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], 'model2')
- elif(nodeId == 'ROADM-A1'):
+ elif nodeId == 'ROADM-A1':
self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'ROADM')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], 'model2')
else:
self.assertFalse(True)
+ continue
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], 'model2')
def test_09_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
nodeId = res['network'][0]['node'][i]['node-id']
# Tests related to XPDRA nodes
- if(nodeId == 'XPDR-A1-XPDR1'):
+ if nodeId == 'XPDR-A1-XPDR1':
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDR-A1'},
res['network'][0]['node'][i]['supporting-node'])
self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
+ tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type'])
tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
- if (tpType == 'XPONDER-CLIENT'):
+ if tpType == 'XPONDER-CLIENT':
client += 1
- elif (tpType == 'XPONDER-NETWORK'):
+ elif tpType == 'XPONDER-NETWORK':
network += 1
- if (tpId == 'XPDR1-NETWORK2'):
+ if tpId == 'XPDR1-NETWORK2':
self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
['transportpce-topology:associated-connection-map-port'], 'XPDR1-CLIENT2')
- if (tpId == 'XPDR1-CLIENT2'):
+ if tpId == 'XPDR1-CLIENT2':
self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
['transportpce-topology:associated-connection-map-port'], 'XPDR1-NETWORK2')
self.assertTrue(client == 2)
self.assertTrue(network == 2)
listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-SRG1'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-SRG3'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
- self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-DEG1'):
- # Test related to DEG1
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
+ elif nodeId in self.CHECK_DICT1:
+ self.assertEqual(nodeType, self.CHECK_DICT1[nodeId]['node_type'])
+ if self.CHECK_DICT1[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
+ for item in self.CHECK_DICT1[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-DEG2'):
- # Test related to DEG2
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
res['network'][0]['node'][i]['supporting-node'])
listNode.remove(nodeId)
def test_10_connect_tail_xpdr_rdm(self):
# Connect the tail: XPDRA to ROADMA
response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
- "ROADM-A1", "1", "SRG1-PP1-TXRX")
+ "ROADM-A1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
def test_11_connect_tail_rdm_xpdr(self):
response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
- "ROADM-A1", "1", "SRG1-PP1-TXRX")
+ "ROADM-A1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
def test_12_getLinks_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(nbLink, 12)
- expressLink = ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
- 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX']
- XPDR_IN = ['ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1']
- XPDR_OUT = ['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX']
+ check_list = {'EXPRESS-LINK': ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX'],
+ 'ADD-LINK': ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX'],
+ 'DROP-LINK': ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX'],
+ 'XPONDER-INPUT': ['ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1'],
+ 'XPONDER-OUTPUT': ['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX']
+ }
for i in range(0, nbLink):
- nodeType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if(nodeType == 'EXPRESS-LINK'):
- find = linkId in expressLink
- self.assertEqual(find, True)
- expressLink.remove(linkId)
- elif(nodeType == 'ADD-LINK'):
- find = linkId in addLink
- self.assertEqual(find, True)
- addLink.remove(linkId)
- elif(nodeType == 'DROP-LINK'):
- find = linkId in dropLink
+ linkType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
+ if linkType in check_list:
+ find = linkId in check_list[linkType]
self.assertEqual(find, True)
- dropLink.remove(linkId)
- elif(nodeType == 'XPONDER-INPUT'):
- find = linkId in XPDR_IN
- self.assertEqual(find, True)
- XPDR_IN.remove(linkId)
- elif(nodeType == 'XPONDER-OUTPUT'):
- find = linkId in XPDR_OUT
- self.assertEqual(find, True)
- XPDR_OUT.remove(linkId)
+ (check_list[linkType]).remove(linkId)
else:
self.assertFalse(True)
- self.assertEqual(len(expressLink), 0)
- self.assertEqual(len(addLink), 0)
- self.assertEqual(len(dropLink), 0)
- self.assertEqual(len(XPDR_IN), 0)
- self.assertEqual(len(XPDR_OUT), 0)
+ for link_type in check_list:
+ self.assertEqual(len(check_list[link_type]), 0)
def test_13_connect_ROADMC(self):
response = test_utils.mount_device("ROADM-C1", 'roadmc')
self.assertEqual(response.status_code, requests.codes.created)
def test_16_getClliNetwork(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
listNode = ['NodeA', 'NodeC']
for i in range(0, nbNode):
nodeId = res['network'][0]['node'][i]['node-id']
- find = nodeId in listNode
- self.assertEqual(find, True)
- if(nodeId == 'NodeA'):
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeA')
+ if nodeId in listNode:
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], nodeId)
+ listNode.remove(nodeId)
else:
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeC')
- listNode.remove(nodeId)
+ self.assertFalse(True)
self.assertEqual(len(listNode), 0)
def test_17_getOpenRoadmNetwork(self):
+ # pylint: disable=redundant-unittest-assert
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(nbNode, 3)
listNode = ['XPDR-A1', 'ROADM-A1', 'ROADM-C1']
+ CHECK_LIST = {'XPDR-A1': {'node-ref': 'NodeA', 'node-type': 'XPONDER'},
+ 'ROADM-A1': {'node-ref': 'NodeA', 'node-type': 'ROADM'},
+ 'ROADM-C1': {'node-ref': 'NodeC', 'node-type': 'ROADM'}
+ }
for i in range(0, nbNode):
self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['network-ref'], 'clli-network')
nodeId = res['network'][0]['node'][i]['node-id']
- if(nodeId == 'XPDR-A1'):
- self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'], 'NodeA')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'XPONDER')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], 'model2')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1'):
- self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'], 'NodeA')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'ROADM')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], 'model2')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-C1'):
- self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'], 'NodeC')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'ROADM')
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], 'model2')
+ if nodeId in CHECK_LIST:
+ self.assertEqual(res['network'][0]['node'][i]['supporting-node'][0]['node-ref'],
+ CHECK_LIST[nodeId]['node-ref'])
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'],
+ CHECK_LIST[nodeId]['node-type'])
listNode.remove(nodeId)
else:
self.assertFalse(True)
+ continue
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-network:model'], 'model2')
self.assertEqual(len(listNode), 0)
def test_18_getROADMLinkOpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(nbLink, 20)
- expressLink = ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX', 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
- 'ROADM-C1-DEG2-DEG2-CTP-TXRXtoROADM-C1-DEG1-DEG1-CTP-TXRX', 'ROADM-C1-DEG1-DEG1-CTP-TXRXtoROADM-C1-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-C1-SRG1-SRG1-CP-TXRXtoROADM-C1-DEG2-DEG2-CTP-TXRX', 'ROADM-C1-SRG1-SRG1-CP-TXRXtoROADM-C1-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
- 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
- 'ROADM-C1-DEG1-DEG1-CTP-TXRXtoROADM-C1-SRG1-SRG1-CP-TXRX', 'ROADM-C1-DEG2-DEG2-CTP-TXRXtoROADM-C1-SRG1-SRG1-CP-TXRX']
- R2RLink = ['ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX',
- 'ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX']
- XPDR_IN = ['ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1']
- XPDR_OUT = ['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX']
+ check_list = {'EXPRESS-LINK': ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-C1-DEG2-DEG2-CTP-TXRXtoROADM-C1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-C1-DEG1-DEG1-CTP-TXRXtoROADM-C1-DEG2-DEG2-CTP-TXRX'],
+ 'ADD-LINK': ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-C1-SRG1-SRG1-CP-TXRXtoROADM-C1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-C1-SRG1-SRG1-CP-TXRXtoROADM-C1-DEG1-DEG1-CTP-TXRX'],
+ 'DROP-LINK': ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
+ 'ROADM-C1-DEG1-DEG1-CTP-TXRXtoROADM-C1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-C1-DEG2-DEG2-CTP-TXRXtoROADM-C1-SRG1-SRG1-CP-TXRX'],
+ 'ROADM-TO-ROADM': ['ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX',
+ 'ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX'],
+ 'XPONDER-INPUT': ['ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1'],
+ 'XPONDER-OUTPUT': ['XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX']
+ }
for i in range(0, nbLink):
- nodeType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if(nodeType == 'EXPRESS-LINK'):
- find = linkId in expressLink
- self.assertEqual(find, True)
- expressLink.remove(linkId)
- elif(nodeType == 'ADD-LINK'):
- find = linkId in addLink
+ linkType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
+ if linkType in check_list:
+ find = linkId in check_list[linkType]
self.assertEqual(find, True)
- addLink.remove(linkId)
- elif(nodeType == 'DROP-LINK'):
- find = linkId in dropLink
- self.assertEqual(find, True)
- dropLink.remove(linkId)
- elif(nodeType == 'ROADM-TO-ROADM'):
- find = linkId in R2RLink
- self.assertEqual(find, True)
- R2RLink.remove(linkId)
- elif(nodeType == 'XPONDER-INPUT'):
- find = linkId in XPDR_IN
- self.assertEqual(find, True)
- XPDR_IN.remove(linkId)
- elif(nodeType == 'XPONDER-OUTPUT'):
- find = linkId in XPDR_OUT
- self.assertEqual(find, True)
- XPDR_OUT.remove(linkId)
+ (check_list[linkType]).remove(linkId)
else:
self.assertFalse(True)
- self.assertEqual(len(expressLink), 0)
- self.assertEqual(len(addLink), 0)
- self.assertEqual(len(dropLink), 0)
- self.assertEqual(len(R2RLink), 0)
- self.assertEqual(len(XPDR_IN), 0)
- self.assertEqual(len(XPDR_OUT), 0)
+ for link_type in check_list:
+ self.assertEqual(len(check_list[link_type]), 0)
def test_19_getLinkOmsAttributesOpenRoadmTopology(self):
response = test_utils.get_ordm_topo_request("")
R2RLink = ['ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX',
'ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX']
for i in range(0, nbLink):
- nodeType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if(link_id in R2RLink):
+ if link_id in R2RLink:
find = False
spanLoss = res['network'][0]['ietf-network-topology:link'][i][
'org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"]
length = res['network'][0]['ietf-network-topology:link'][i][
'org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]['SRLG-length']
- if((spanLoss != None) & (length != None)):
+ if (spanLoss is not None) & (length is not None):
find = True
self.assertTrue(find)
R2RLink.remove(link_id)
self.assertEqual(len(R2RLink), 0)
def test_20_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
listNode = ['XPDR-A1-XPDR1',
'ROADM-A1-SRG1', 'ROADM-A1-SRG3', 'ROADM-A1-DEG1', 'ROADM-A1-DEG2',
'ROADM-C1-SRG1', 'ROADM-C1-DEG1', 'ROADM-C1-DEG2']
- # ************************Tests related to XPDRA nodes
+ # Tests related to XPDRA nodes
for i in range(0, nbNode):
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
nodeId = res['network'][0]['node'][i]['node-id']
- if(nodeId == 'XPDR-A1-XPDR1'):
+ if nodeId == 'XPDR-A1-XPDR1':
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDR-A1'},
res['network'][0]['node'][i]['supporting-node'])
self.assertEqual(nodeType, 'XPONDER')
client = 0
network = 0
for j in range(0, nbTps):
- tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
- if (tpType == 'XPONDER-CLIENT'):
+ tpType = (res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type'])
+ if tpType == 'XPONDER-CLIENT':
client += 1
- elif (tpType == 'XPONDER-NETWORK'):
+ elif tpType == 'XPONDER-NETWORK':
network += 1
self.assertTrue(client == 2)
self.assertTrue(network == 2)
listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-SRG1'):
- # Test related to SRG1
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'SRG')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-SRG3'):
- # Test related to SRG1
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
- self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'SRG')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-DEG1'):
- # Test related to DEG1
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
+ # Tests related to ROADMA nodes
+ elif nodeId in self.CHECK_DICT1:
+ self.assertEqual(nodeType, self.CHECK_DICT1[nodeId]['node_type'])
+ if self.CHECK_DICT1[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
+ for item in self.CHECK_DICT1[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'DEGREE')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-DEG2'):
- # Test related to DEG2
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'DEGREE')
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'],
+ self.CHECK_DICT1[nodeId]['node_type'])
listNode.remove(nodeId)
- elif(nodeId == 'ROADM-C1-SRG1'):
- # Test related to SRG1
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-C1'},
+ # Tests related to ROADMA nodes
+ elif nodeId in self.CHECK_DICT2:
+ self.assertEqual(nodeType, self.CHECK_DICT2[nodeId]['node_type'])
+ if self.CHECK_DICT2[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
+ for item in self.CHECK_DICT2[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeC'},
res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'SRG')
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-C1-DEG1'):
- # Test related to DEG1
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-C1'},
res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'DEGREE')
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'],
+ self.CHECK_DICT2[nodeId]['node_type'])
listNode.remove(nodeId)
- elif(nodeId == 'ROADM-C1-DEG2'):
- # Test related to DEG1
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-C1'},
- res['network'][0]['node'][i]['supporting-node'])
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-common-network:node-type'], 'DEGREE')
- listNode.remove(nodeId)
- else:
- self.assertFalse(True)
self.assertEqual(len(listNode), 0)
def test_21_connect_ROADMB(self):
self.assertEqual(response.status_code, requests.codes.created)
def test_26_getClliNetwork(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_clli_net_request()
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
listNode = ['NodeA', 'NodeB', 'NodeC']
for i in range(0, nbNode):
nodeId = res['network'][0]['node'][i]['node-id']
- find = nodeId in listNode
- self.assertEqual(find, True)
- if(nodeId == 'NodeA'):
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeA')
- elif(nodeId == 'NodeB'):
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeB')
+ if nodeId in listNode:
+ self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], nodeId)
+ listNode.remove(nodeId)
else:
- self.assertEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeC')
- listNode.remove(nodeId)
+ self.assertFalse(True)
self.assertEqual(len(listNode), 0)
def test_27_verifyDegree(self):
res = response.json()
# Tests related to links
nbLink = len(res['network'][0]['ietf-network-topology:link'])
- listR2RLink = ['ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX', 'ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX',
- 'ROADM-A1-DEG1-DEG1-TTP-TXRXtoROADM-B1-DEG1-DEG1-TTP-TXRX', 'ROADM-C1-DEG2-DEG2-TTP-TXRXtoROADM-B1-DEG2-DEG2-TTP-TXRX',
- 'ROADM-B1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG1-DEG1-TTP-TXRX', 'ROADM-B1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG2-DEG2-TTP-TXRX']
+ listR2RLink = ['ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX',
+ 'ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-TTP-TXRXtoROADM-B1-DEG1-DEG1-TTP-TXRX',
+ 'ROADM-C1-DEG2-DEG2-TTP-TXRXtoROADM-B1-DEG2-DEG2-TTP-TXRX',
+ 'ROADM-B1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG1-DEG1-TTP-TXRX',
+ 'ROADM-B1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG2-DEG2-TTP-TXRX']
for i in range(0, nbLink):
- if res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ROADM-TO-ROADM':
+ if (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:link-type'] == 'ROADM-TO-ROADM'):
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
find = link_id in listR2RLink
self.assertEqual(find, True)
link_type = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
link_src = res['network'][0]['ietf-network-topology:link'][i]['source']['source-node']
link_dest = res['network'][0]['ietf-network-topology:link'][i]['destination']['dest-node']
- oppLink_id = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:opposite-link']
+ oppLink_id = (res['network'][0]['ietf-network-topology:link'][i]
+ ['org-openroadm-common-network:opposite-link'])
# Find the opposite link
response_oppLink = test_utils.get_ordm_topo_request("ietf-network-topology:link/"+oppLink_id)
self.assertEqual(response_oppLink.status_code, requests.codes.ok)
self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['source']['source-node'], link_dest)
self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['destination']['dest-node'], link_src)
oppLink_type = res_oppLink['ietf-network-topology:link'][0]['org-openroadm-common-network:link-type']
- if link_type == 'ADD-LINK':
- self.assertEqual(oppLink_type, 'DROP-LINK')
- elif link_type == 'DROP-LINK':
- self.assertEqual(oppLink_type, 'ADD-LINK')
- elif link_type == 'EXPRESS-LINK':
- self.assertEqual(oppLink_type, 'EXPRESS-LINK')
- elif link_type == 'ROADM-TO-ROADM':
- self.assertEqual(oppLink_type, 'ROADM-TO-ROADM')
- elif link_type == 'XPONDER-INPUT':
- self.assertEqual(oppLink_type, 'XPONDER-OUTPUT')
- elif link_type == 'XPONDER-OUTPUT':
- self.assertEqual(oppLink_type, 'XPONDER-INPUT')
+ CHECK_DICT = {'ADD-LINK': 'DROP-LINK', 'DROP-LINK': 'ADD-LINK',
+ 'EXPRESS-LINK': 'EXPRESS-LINK', 'ROADM-TO-ROADM': 'ROADM-TO-ROADM',
+ 'XPONDER-INPUT': 'XPONDER-OUTPUT', 'XPONDER-OUTUT': 'XPONDER-INPUT'}
+ if link_type in CHECK_DICT:
+ self.assertEqual(oppLink_type, CHECK_DICT[link_type])
def test_29_getLinkOmsAttributesOpenRoadmTopology(self):
response = test_utils.get_ordm_topo_request("")
'ROADM-B1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG1-DEG1-TTP-TXRX',
'ROADM-B1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG2-DEG2-TTP-TXRX']
for i in range(0, nbLink):
- nodeType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- if(link_id in R2RLink):
+ if link_id in R2RLink:
find = False
spanLoss = res['network'][0]['ietf-network-topology:link'][i][
'org-openroadm-network-topology:OMS-attributes']['span']["engineered-spanloss"]
length = res['network'][0]['ietf-network-topology:link'][i][
'org-openroadm-network-topology:OMS-attributes']['span']['link-concatenation'][0]['SRLG-length']
- if((spanLoss != None) & (length != None)):
+ if (spanLoss is not None) & (length is not None):
find = True
self.assertTrue(find)
R2RLink.remove(link_id)
response = test_utils.del_node_request("NodeC")
self.assertEqual(response.status_code, requests.codes.ok)
-# def test_24_check_roadm2roadm_links_deletion(self):
-# response = test_utils.get_ordm_topo_request("")
-# self.assertEqual(response.status_code, requests.codes.ok)
-# res = response.json()
-# #Write the response in the log
-# with open('./transportpce_tests/log/response.log', 'a') as outfile1:
-# outfile1.write(str(res))
-# #Tests related to links
-# nbLink=len(res['network'][0]['ietf-network-topology:link'])
-# self.assertEqual(nbLink,8)
-# expressLink=['ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX','ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX']
-# addLink=['ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG2-DEG2-CTP-TXRX','ROADMA-SRG1-SRG1-CP-TXRXtoROADMA-DEG1-DEG1-CTP-TXRX',]
-# dropLink=['ROADMA-DEG1-DEG1-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX','ROADMA-DEG2-DEG2-CTP-TXRXtoROADMA-SRG1-SRG1-CP-TXRX']
-# XPDR_IN=['ROADMA-SRG1-SRG1-PP1-TXRXtoXPDRA-XPDR1-XPDR1-NETWORK1']
-# XPDR_OUT=['XPDRA-XPDR1-XPDR1-NETWORK1toROADMA-SRG1-SRG1-PP1-TXRX']
-# for i in range(0,nbLink):
-# nodeType=res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
-# linkId=res['network'][0]['ietf-network-topology:link'][i]['link-id']
-# if(nodeType=='EXPRESS-LINK'):
-# find= linkId in expressLink
-# self.assertEqual(find, True)
-# expressLink.remove(linkId)
-# elif(nodeType=='ADD-LINK'):
-# find= linkId in addLink
-# self.assertEqual(find, True)
-# addLink.remove(linkId)
-# elif(nodeType=='DROP-LINK'):
-# find= linkId in dropLink
-# self.assertEqual(find, True)
-# dropLink.remove(linkId)
-# elif(nodeType=='XPONDER-INPUT'):
-# find= linkId in XPDR_IN
-# self.assertEqual(find, True)
-# XPDR_IN.remove(linkId)
-# elif(nodeType=='XPONDER-OUTPUT'):
-# find= linkId in XPDR_OUT
-# self.assertEqual(find, True)
-# XPDR_OUT.remove(linkId)
-# else:
-# self.assertFalse(True)
-# self.assertEqual(len(expressLink),0)
-# self.assertEqual(len(addLink),0)
-# self.assertEqual(len(dropLink),0)
-# self.assertEqual(len(XPDR_IN),0)
-# self.assertEqual(len(XPDR_OUT),0)
-#
-# for i in range(0,nbLink):
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'],'ROADM-TO-ROADM')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-SRG1-SRG1-CP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG1-DEG1-CTP-TXRXtoROADMC-SRG1-SRG1-CP-TXRX')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-SRG1-SRG1-CP-TXRXtoROADMC-DEG2-DEG1-CTP-TXRX')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG1-DEG2-CTP-TXRXtoROADMC-SRG1-SRG1-CP-TXRX')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG1-DEG1-CTP-TXRXtoROADMC-DEG2-DEG2-CTP-TXRX')
-# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG2-DEG2-CTP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX')
-
def test_32_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
nodeId = res['network'][0]['node'][i]['node-id']
# Tests related to XPDRA nodes
- if(nodeId == 'XPDR-A1-XPDR1'):
+ if nodeId == 'XPDR-A1-XPDR1':
nbTp = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
for j in range(0, nbTp):
tpid = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
- if (tpid == 'XPDR1-CLIENT1'):
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-common-network:tp-type'], 'XPONDER-CLIENT')
- if (tpid == 'XPDR1-NETWORK1'):
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-common-network:tp-type'], 'XPONDER-NETWORK')
- self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
- ['org-openroadm-network-topology:xpdr-network-attributes']['tail-equipment-id'],
+ if tpid == 'XPDR1-CLIENT1':
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type']), 'XPONDER-CLIENT')
+ if tpid == 'XPDR1-NETWORK1':
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-common-network:tp-type']), 'XPONDER-NETWORK')
+ self.assertEqual((res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
+ ['org-openroadm-network-topology:xpdr-network-attributes']
+ ['tail-equipment-id']),
'ROADM-A1-SRG1--SRG1-PP1-TXRX')
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'XPDR-A1'},
res['network'][0]['node'][i]['supporting-node'])
listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-SRG1'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-SRG3'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
- self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
- res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-DEG1'):
- # Test related to DEG1
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
+ # Tests related to ROADMA nodes
+ elif nodeId in self.CHECK_DICT1:
+ self.assertEqual(nodeType, self.CHECK_DICT1[nodeId]['node_type'])
+ if self.CHECK_DICT1[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
+ for item in self.CHECK_DICT1[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
res['network'][0]['node'][i]['supporting-node'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-DEG2'):
- # Test related to DEG2
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
res['network'][0]['node'][i]['supporting-node'])
listNode.remove(nodeId)
nbNode = len(res['network'][0]['node'])
self.assertEqual(nbNode, 1)
for i in range(0, nbNode-1):
- self.assertNotEqual(res['network'][0]['node'][1]['org-openroadm-clli-network:clli'], 'NodeC')
+ self.assertNotEqual(res['network'][0]['node'][i]['org-openroadm-clli-network:clli'], 'NodeC')
def test_35_disconnect_XPDRA(self):
response = test_utils.unmount_device("XPDR-A1")
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'XPDR-A1')
def test_38_getNodes_OpenRoadmTopology(self):
+ # pylint: disable=redundant-unittest-assert
response = test_utils.get_ordm_topo_request("")
res = response.json()
# Tests related to nodes
res['network'][0]['node'][i]['supporting-node'])
nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
nodeId = res['network'][0]['node'][i]['node-id']
- if(nodeId == 'ROADM-A1-SRG1'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
- self.assertIn({'tp-id': 'SRG1-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG1-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-SRG3'):
- # Test related to SRG1
- self.assertEqual(nodeType, 'SRG')
- self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
- self.assertIn({'tp-id': 'SRG3-CP-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-CP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'SRG3-PP1-TXRX', 'org-openroadm-common-network:tp-type': 'SRG-TXRX-PP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-DEG1'):
- # Test related to DEG1
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG1-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG1-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- listNode.remove(nodeId)
- elif(nodeId == 'ROADM-A1-DEG2'):
- # Test related to DEG2
- self.assertEqual(nodeType, 'DEGREE')
- self.assertIn({'tp-id': 'DEG2-TTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-TTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
- self.assertIn({'tp-id': 'DEG2-CTP-TXRX', 'org-openroadm-common-network:tp-type': 'DEGREE-TXRX-CTP'},
- res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ if nodeId in self.CHECK_DICT1:
+ self.assertEqual(nodeType, self.CHECK_DICT1[nodeId]['node_type'])
+ if self.CHECK_DICT1[nodeId]['node_type'] == 'SRG':
+ self.assertEqual(len(res['network'][0]['node'][i]['ietf-network-topology:termination-point']), 5)
+ for item in self.CHECK_DICT1[nodeId]['checks_tp']:
+ self.assertIn(item, res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
+ self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeA'},
+ res['network'][0]['node'][i]['supporting-node'])
+ self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'ROADM-A1'},
+ res['network'][0]['node'][i]['supporting-node'])
listNode.remove(nodeId)
else:
self.assertFalse(True)
res = response.json()
nbLink = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(nbLink, 16)
- expressLink = ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX']
- addLink = ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
- 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX', 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX']
- dropLink = ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
- 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX', 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX']
+ check_list = {'EXPRESS-LINK': ['ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX'],
+ 'ADD-LINK': ['ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG1-SRG1-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG2-DEG2-CTP-TXRX',
+ 'ROADM-A1-SRG3-SRG3-CP-TXRXtoROADM-A1-DEG1-DEG1-CTP-TXRX'],
+ 'DROP-LINK': ['ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG1-SRG1-CP-TXRX',
+ 'ROADM-A1-DEG1-DEG1-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX',
+ 'ROADM-A1-DEG2-DEG2-CTP-TXRXtoROADM-A1-SRG3-SRG3-CP-TXRX']
+ }
roadmtoroadmLink = 0
for i in range(0, nbLink):
- if (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'EXPRESS-LINK'):
- link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- find = link_id in expressLink
- self.assertEqual(find, True)
- expressLink.remove(link_id)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'ADD-LINK'):
- link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- find = link_id in addLink
- self.assertEqual(find, True)
- addLink.remove(link_id)
- elif (res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type'] == 'DROP-LINK'):
- link_id = res['network'][0]['ietf-network-topology:link'][i]['link-id']
- find = link_id in dropLink
+ linkId = res['network'][0]['ietf-network-topology:link'][i]['link-id']
+ linkType = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:link-type']
+ if linkType in check_list:
+ find = linkId in check_list[linkType]
self.assertEqual(find, True)
- dropLink.remove(link_id)
+ (check_list[linkType]).remove(linkId)
else:
roadmtoroadmLink += 1
- self.assertEqual(len(expressLink), 0)
- self.assertEqual(len(addLink), 0)
- self.assertEqual(len(dropLink), 0)
+ for link_type in check_list:
+ self.assertEqual(len(check_list[link_type]), 0)
self.assertEqual(roadmtoroadmLink, 6)
for i in range(0, nbLink):
self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]
#!/usr/bin/env python
+
##############################################################################
# Copyright (c) 2020 Orange, Inc. and others. All rights reserved.
#
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+
+# pylint: disable=no-member
+
import json
import os
import sys
"..", "..", "..", "karaf", "target", "assembly", "bin", "client")
return subprocess.run([executable],
input='feature:install ' + feature_name + '\n feature:list | grep tapi \n logout \n',
- universal_newlines=True)
+ universal_newlines=True, check=False)
def get_request(url):
data=json.dumps(data),
headers=TYPE_APPLICATION_JSON,
auth=(ODL_LOGIN, ODL_PWD))
- else:
- return requests.request(
- "POST", url.format(RESTCONF_BASE_URL),
- headers=TYPE_APPLICATION_JSON,
- auth=(ODL_LOGIN, ODL_PWD))
+
+ return requests.request(
+ "POST", url.format(RESTCONF_BASE_URL),
+ headers=TYPE_APPLICATION_JSON,
+ auth=(ODL_LOGIN, ODL_PWD))
def post_xmlrequest(url, data):
data=data,
headers=TYPE_APPLICATION_XML,
auth=(ODL_LOGIN, ODL_PWD))
+ return None
def put_request(url, data):
def service_create_request(attr):
return post_request(URL_SERV_CREATE, attr)
-def service_delete_request(servicename : str,
- requestid = "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
+
+def service_delete_request(servicename: str,
+ requestid="e3028bae-a90f-4ddd-a83f-cf224eba0e58",
notificationurl="http://localhost:8585/NotificationServer/notify"):
attr = {"input": {
- "sdnc-request-header": {
- "request-id": requestid,
- "rpc-action": "service-delete",
- "request-system-id": "appname",
- "notification-url": notificationurl},
- "service-delete-req-info": {
- "service-name": servicename,
- "tail-retention": "no"}}}
+ "sdnc-request-header": {
+ "request-id": requestid,
+ "rpc-action": "service-delete",
+ "request-system-id": "appname",
+ "notification-url": notificationurl},
+ "service-delete-req-info": {
+ "service-name": servicename,
+ "tail-retention": "no"}}}
return post_request(URL_SERV_DELETE, attr)
def service_path_request(operation: str, servicename: str, wavenumber: str, nodes):
attr = {"renderer:input": {
- "renderer:service-name": servicename,
- "renderer:wave-number": wavenumber,
- "renderer:modulation-format": "qpsk",
- "renderer:operation": operation,
- "renderer:nodes": nodes}}
+ "renderer:service-name": servicename,
+ "renderer:wave-number": wavenumber,
+ "renderer:modulation-format": "qpsk",
+ "renderer:operation": operation,
+ "renderer:nodes": nodes}}
return post_request(URL_SERVICE_PATH, attr)
-def otn_service_path_request(operation: str, servicename: str, servicerate: str, servicetype: str, nodes, eth_attr=None):
+def otn_service_path_request(operation: str, servicename: str, servicerate: str, servicetype: str, nodes,
+ eth_attr=None):
attr = {"service-name": servicename,
"operation": operation,
"service-rate": servicerate,
def create_ots_oms_request(nodeid: str, lcp: str):
attr = {"input": {
- "node-id": nodeid,
- "logical-connection-point": lcp}}
+ "node-id": nodeid,
+ "logical-connection-point": lcp}}
return post_request(URL_CREATE_OTS_OMS, attr)
+
def path_computation_request(requestid: str, servicename: str, serviceaend, servicezend,
hardconstraints=None, softconstraints=None, metric="hop-count", other_attr=None):
- attr = {"service-name": servicename,
- "resource-reserve": "true",
- "service-handler-header": { "request-id": requestid },
- "service-a-end": serviceaend,
- "service-z-end": servicezend,
- "pce-metric": metric}
+ attr = {"service-name": servicename,
+ "resource-reserve": "true",
+ "service-handler-header": {"request-id": requestid},
+ "service-a-end": serviceaend,
+ "service-z-end": servicezend,
+ "pce-metric": metric}
if hardconstraints:
- attr.update({ "hard-constraints": hardconstraints})
+ attr.update({"hard-constraints": hardconstraints})
if softconstraints:
- attr.update({ "soft-constraints": softconstraints})
+ attr.update({"soft-constraints": softconstraints})
if other_attr:
attr.update(other_attr)
- return post_request(URL_PATH_COMPUTATION_REQUEST, {"input": attr })
+ return post_request(URL_PATH_COMPUTATION_REQUEST, {"input": attr})
def shutdown_process(process):
return subprocess.Popen(
[HONEYNODE_EXECUTABLE, node_port, os.path.join(SAMPLES_DIRECTORY, node_config_file_name)],
stdout=outfile, stderr=outfile)
+ return None
def wait_until_log_contains(log_file, regexp, time_to_wait=20):
+ # pylint: disable=lost-exception
stringfound = False
filefound = False
line = None
signal.alarm(self.seconds)
def __exit__(self, type, value, traceback):
+ # pylint: disable=W0622
signal.alarm(0)
{py3,portmapping,topoPortMapping,flexgrid,rspn,topology,pce,olm,end2end,portmapping221,rspn221,otnrenderer,otnshrenderer,topology221,otntopology,olm221,tapi221,otnend2end,end2end221,gnpy}: - sh -c 'if [ "$USE_LIGHTY" = "True" ]; then (cd ../lighty && ./build.sh); fi'
#run 1.2.1 functional tests
{py3,portmapping}: nosetests --with-xunit transportpce_tests/1.2.1/test_portmapping.py
- {py3,topoPortMapping}: nosetests --with-xunit transportpce_tests/1.2.1/test_topoPortMapping.py
+ {py3,topoPortMapping}: nosetests --with-xunit transportpce_tests/1.2.1/test_topo_portmapping.py
{py3,topology}: nosetests --with-xunit transportpce_tests/1.2.1/test_topology.py
{py3,rspn}: nosetests --with-xunit transportpce_tests/1.2.1/test_renderer_service_path_nominal.py
{py3,pce}: nosetests --with-xunit transportpce_tests/1.2.1/test_pce.py
deps = pylint
whitelist_externals = find
commands =
- find transportpce_tests/ -name *.py -exec pylint --max-line-length=120 --disable=missing-docstring --reports=y --score=y --output-format=colorized \{\} +
+ find transportpce_tests/ -name *.py -exec pylint --max-line-length=120 --disable=missing-docstring --method-rgx="(([a-z_][a-zA-Z0-9_]{2,})|(_[a-z0-9_]*)|(__[a-zA-Z][a-zA-Z0-9_]+__))$" --variable-rgx="[a-zA-Z_][a-zA-Z0-9_]{1,30}$" --reports=y --score=y --output-format=colorized \{\} +
Code Review / transportpce.git / commitdiff