mirror of
https://github.com/Telecominfraproject/oopt-gnpy.git
synced 2025-11-01 02:28:05 +00:00
small linter fixes
Signed-off-by: EstherLerouzic <esther.lerouzic@orange.com>
This commit is contained in:
@@ -1,13 +1,15 @@
|
||||
#!/usr/bin/env python3
|
||||
# TelecomInfraProject/gnpy/examples
|
||||
# Module name : test_disjunction.py
|
||||
# Version :
|
||||
# License : BSD 3-Clause Licence
|
||||
# Version:
|
||||
# License: BSD 3-Clause Licence
|
||||
# Copyright (c) 2018, Telecom Infra Project
|
||||
|
||||
"""
|
||||
@author: esther.lerouzic
|
||||
checks that computed paths are disjoint as specified in the json service file
|
||||
that computed paths do not loop
|
||||
that include node constraints are correctly taken into account
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
@@ -15,22 +17,24 @@ import pytest
|
||||
from gnpy.core.equipment import load_equipment, trx_mode_params, automatic_nch
|
||||
from gnpy.core.network import load_network, build_network
|
||||
from gnpy.core.exceptions import ServiceError
|
||||
from examples.path_requests_run import (requests_from_json , correct_route_list ,
|
||||
load_requests , disjunctions_from_json)
|
||||
from gnpy.core.request import compute_path_dsjctn, isdisjoint , find_reversed_path, Path_request
|
||||
from gnpy.core.utils import db2lin, lin2db
|
||||
from examples.path_requests_run import (requests_from_json, correct_route_list,
|
||||
load_requests, disjunctions_from_json)
|
||||
from gnpy.core.request import compute_path_dsjctn, isdisjoint, find_reversed_path, Path_request
|
||||
from gnpy.core.utils import lin2db
|
||||
from gnpy.core.elements import Roadm
|
||||
from gnpy.core.spectrum_assignment import build_oms_list
|
||||
|
||||
network_file_name = Path(__file__).parent.parent / 'tests/data/testTopology_expected.json'
|
||||
service_file_name = Path(__file__).parent.parent / 'tests/data/testTopology_testservices.json'
|
||||
result_file_name = Path(__file__).parent.parent / 'tests/data/testTopology_testresults.json'
|
||||
eqpt_library_name = Path(__file__).parent.parent / 'tests/data/eqpt_config.json'
|
||||
NETWORK_FILE_NAME = Path(__file__).parent.parent / 'tests/data/testTopology_expected.json'
|
||||
SERVICE_FILE_NAME = Path(__file__).parent.parent / 'tests/data/testTopology_testservices.json'
|
||||
RESULT_FILE_NAME = Path(__file__).parent.parent / 'tests/data/testTopology_testresults.json'
|
||||
EQPT_LIBRARY_NAME = Path(__file__).parent.parent / 'tests/data/eqpt_config.json'
|
||||
|
||||
@pytest.fixture()
|
||||
def serv(test_setup):
|
||||
""" common setup for service list
|
||||
"""
|
||||
network, equipment = test_setup
|
||||
data = load_requests(service_file_name, eqpt_library_name, bidir=False)
|
||||
data = load_requests(SERVICE_FILE_NAME, EQPT_LIBRARY_NAME, bidir=False)
|
||||
rqs = requests_from_json(data, equipment)
|
||||
rqs = correct_route_list(network, rqs)
|
||||
dsjn = disjunctions_from_json(data)
|
||||
@@ -40,13 +44,13 @@ def serv(test_setup):
|
||||
def test_setup():
|
||||
""" common setup for tests: builds network, equipment and oms only once
|
||||
"""
|
||||
equipment = load_equipment(eqpt_library_name)
|
||||
network = load_network(network_file_name,equipment)
|
||||
equipment = load_equipment(EQPT_LIBRARY_NAME)
|
||||
network = load_network(NETWORK_FILE_NAME, equipment)
|
||||
# Build the network once using the default power defined in SI in eqpt config
|
||||
# power density : db2linp(ower_dbm": 0)/power_dbm": 0 * nb channels as defined by
|
||||
# spacing, f_min and f_max
|
||||
# spacing, f_min and f_max
|
||||
p_db = equipment['SI']['default'].power_dbm
|
||||
|
||||
|
||||
p_total_db = p_db + lin2db(automatic_nch(equipment['SI']['default'].f_min,\
|
||||
equipment['SI']['default'].f_max, equipment['SI']['default'].spacing))
|
||||
build_network(network, equipment, p_db, p_total_db)
|
||||
@@ -62,7 +66,7 @@ def test_disjunction(serv):
|
||||
pths = compute_path_dsjctn(network, equipment, rqs, dsjn)
|
||||
print(dsjn)
|
||||
|
||||
dsjn_list = [d.disjunctions_req for d in dsjn ]
|
||||
dsjn_list = [d.disjunctions_req for d in dsjn]
|
||||
|
||||
# assumes only pairs in dsjn list
|
||||
test = True
|
||||
@@ -83,28 +87,24 @@ def test_does_not_loop_back(serv):
|
||||
"""
|
||||
network, equipment, rqs, dsjn = serv
|
||||
pths = compute_path_dsjctn(network, equipment, rqs, dsjn)
|
||||
|
||||
test = True
|
||||
for p in pths :
|
||||
for p in pths:
|
||||
for el in p:
|
||||
p.remove(el)
|
||||
a = [e for e in p if e.uid == el.uid]
|
||||
if a :
|
||||
if a:
|
||||
test = False
|
||||
break
|
||||
assert test
|
||||
|
||||
assert test
|
||||
|
||||
# TODO : test that identical requests are correctly agregated
|
||||
# TODO : test that identical requests are correctly agregated
|
||||
# and reproduce disjunction vector as well as route constraints
|
||||
# check that requests with different parameters are not aggregated
|
||||
|
||||
# check that the total agregated bandwidth is the same after aggregation
|
||||
|
||||
#
|
||||
|
||||
def create_rq(equipment, srce, dest, bdir, nd_list, ls_list):
|
||||
""" co
|
||||
""" create the usual request list according to parameters
|
||||
"""
|
||||
requests_list = []
|
||||
params = {}
|
||||
@@ -158,9 +158,6 @@ def test_include_constraints(test_setup, srce, dest, result, pth, nd_list, ls_li
|
||||
"""
|
||||
network, equipment = test_setup
|
||||
dsjn = []
|
||||
# case 1 wrong source or dest
|
||||
# srce = 'a'
|
||||
# dest = 'trx h'
|
||||
bdir = False
|
||||
rqs = create_rq(equipment, srce, dest, bdir, nd_list, ls_list)
|
||||
print(rqs)
|
||||
|
||||
Reference in New Issue
Block a user