#!/usr/bin/env python3 # -*- coding: utf-8 -*- # SPDX-License-Identifier: BSD-3-Clause # test_disjunction # Copyright (C) 2025 Telecom Infra Project and GNPy contributors # see AUTHORS.rst for a list of contributors """ 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 import pytest from gnpy.core.equipment import trx_mode_params from gnpy.core.network import build_network from gnpy.core.exceptions import ServiceError, DisjunctionError from gnpy.core.utils import automatic_nch, dbm2watt from gnpy.core.elements import Roadm from gnpy.topology.request import (compute_path_dsjctn, isdisjoint, find_reversed_path, PathRequest, correct_json_route_list, requests_aggregation, Disjunction) from gnpy.topology.spectrum_assignment import build_oms_list from gnpy.tools.json_io import requests_from_json, load_requests, load_network, load_equipment, \ disjunctions_from_json, load_json DATA_DIR = Path(__file__).parent.parent / 'tests/data' NETWORK_FILE_NAME = DATA_DIR / 'testTopology_expected.json' SERVICE_FILE_NAME = DATA_DIR / 'testTopology_testservices.json' RESULT_FILE_NAME = DATA_DIR / 'testTopology_testresults.json' EQPT_LIBRARY_NAME = DATA_DIR / 'eqpt_config.json' EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": load_json(DATA_DIR / "std_medium_gain_advanced_config.json")} @pytest.fixture() def serv(test_setup): """common setup for service list""" network, equipment = test_setup data = load_requests(SERVICE_FILE_NAME, equipment, bidir=False, network=network, network_filename=NETWORK_FILE_NAME) rqs = requests_from_json(data, equipment) rqs = correct_json_route_list(network, rqs) dsjn = disjunctions_from_json(data) return network, equipment, rqs, dsjn @pytest.fixture() def test_setup(): """common setup for tests: builds network, equipment and oms only once""" equipment = load_equipment(EQPT_LIBRARY_NAME, EXTRA_CONFIGS) 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 p_db = equipment['SI']['default'].power_dbm params = { "power": dbm2watt(p_db), "tx_power": dbm2watt(p_db), "nb_channel": automatic_nch(equipment['SI']['default'].f_min, equipment['SI']['default'].f_max, equipment['SI']['default'].spacing), 'request_id': None, 'trx_type': None, 'trx_mode': None, 'source': None, 'destination': None, 'bidir': False, 'nodes_list': [], 'loose_list': [], 'format': '', 'baud_rate': None, 'bit_rate': None, 'roll_off': None, 'OSNR': None, 'penalties': None, 'path_bandwidth': None, 'effective_freq_slot': None, 'f_min': None, 'f_max': None, 'spacing': None, 'min_spacing': None, 'cost': None, 'equalization_offset_db': None, 'tx_osnr': None } build_network(network, equipment, PathRequest(**params)) build_oms_list(network, equipment) return network, equipment def test_disjunction(serv): """service_file contains sevaral combination of disjunction constraint The test checks that computed paths with disjunction constraint are effectively disjoint. """ network, equipment, rqs, dsjn = serv pths = compute_path_dsjctn(network, equipment, rqs, dsjn) print(dsjn) dsjn_list = [d.disjunctions_req for d in dsjn] # assumes only pairs in dsjn list test = True for e in dsjn_list: rqs_id_list = [r.request_id for r in rqs] p1 = pths[rqs_id_list.index(e[0])][1:-1] p2 = pths[rqs_id_list.index(e[1])][1:-1] if isdisjoint(p1, p2) + isdisjoint(p1, find_reversed_path(p2)) > 0: test = False print(f'Computed path (roadms):{[e.uid for e in p1 if isinstance(e, Roadm)]}\n') print(f'Computed path (roadms):{[e.uid for e in p2 if isinstance(e, Roadm)]}\n') break print(dsjn_list) assert test def test_does_not_loop_back(serv): """check that computed paths do not loop back ie each element appears only once""" network, equipment, rqs, dsjn = serv pths = compute_path_dsjctn(network, equipment, rqs, dsjn) test = True for p in pths: for el in p: p.remove(el) a = [e for e in p if e.uid == el.uid] if a: test = False break assert test # 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, node_list, loose_list, rqid='test_request'): """create the usual request list according to parameters""" requests_list = [] params = { 'request_id': rqid, 'source': srce, 'bidir': bdir, 'destination': dest, 'trx_type': 'Voyager', 'trx_mode': 'mode 1', 'spacing': 50000000000.0, 'nodes_list': node_list, 'loose_list': loose_list, 'path_bandwidth': 100.0e9, 'power': 1.0e-3, 'tx_power': 1.0e-3, 'effective_freq_slot': None } params['format'] = params['trx_mode'] trx_params = trx_mode_params(equipment, params['trx_type'], params['trx_mode'], True) params.update(trx_params) f_min = params['f_min'] f_max_from_si = params['f_max'] params['nb_channel'] = automatic_nch(f_min, f_max_from_si, params['spacing']) requests_list.append(PathRequest(**params)) return requests_list @pytest.mark.parametrize('srce, dest, result, pth, node_list, loose_list', [ ['a', 'trx h', 'fail', 'no_path', [], []], ['trx a', 'h', 'fail', 'no_path', [], []], ['trx a', 'trx h', 'pass', 'found_path', [], []], ['trx a', 'trx h', 'pass', 'found_path', ['roadm b', 'roadm a'], ['LOOSE', 'LOOSE']], ['trx a', 'trx h', 'pass', 'no_path', ['roadm b', 'roadm a'], ['STRICT', 'STRICT']], ['trx a', 'trx h', 'pass', 'found_path', ['roadm b', 'roadm c'], ['STRICT', 'STRICT']], ['trx a', 'trx h', 'fail', 'no_path', ['Lorient_KMA', 'roadm c'], ['STRICT', 'STRICT']], ['trx a', 'trx h', 'pass', 'no_path', ['roadm Lorient_KMA', 'roadm c'], ['LOOSE', 'STRICT']], ['trx a', 'trx h', 'pass', 'found_path', ['roadm c', 'roadm c'], ['LOOSE', 'LOOSE']], ['trx a', 'trx h', 'pass', 'found_path', ['roadm c', 'roadm c'], ['STRICT', 'STRICT']], ['trx a', 'trx h', 'pass', 'found_path', ['roadm c', 'roadm g'], ['STRICT', 'STRICT']], ['trx a', 'trx h', 'pass', 'found_path', ['trx a', 'roadm g'], ['STRICT', 'STRICT']], ['trx a', 'trx h', 'pass', 'found_path', ['trx h'], ['STRICT']], ['trx a', 'trx h', 'pass', 'found_path', ['roadm a'], ['STRICT']]]) def test_include_constraints(test_setup, srce, dest, result, pth, node_list, loose_list): """check that all combinations of constraints are correctly handled: - STRICT/LOOSE - correct names/incorrect names -> pass/fail - possible include/impossible include if incorrect name -> fail else: constraint |one or more STRICT | all LOOSE ---------------------------------------------------------------------------------- >1 path from s to d | can be applied | found_path | found_path | cannot be applied | no_path | found_path ---------------------------------------------------------------------------------- 0 | | computation stops """ network, equipment = test_setup dsjn = [] bdir = False rqs = create_rq(equipment, srce, dest, bdir, node_list, loose_list) print(rqs) if result == 'fail': with pytest.raises(ServiceError): rqs = correct_json_route_list(network, rqs) else: rqs = correct_json_route_list(network, rqs) paths = compute_path_dsjctn(network, equipment, rqs, dsjn) # if loose, one path can be returned if paths[0]: assert pth == 'found_path' else: assert pth == 'no_path' @pytest.mark.parametrize('dis1, dis2, node_list1, loose_list1, result, expected_paths', [ [['1', '2', '3'], ['2', '3'], [], [], 'pass', [['roadm a', 'roadm c', 'roadm d', 'roadm e', 'roadm g'], ['roadm c', 'roadm f'], ['roadm a', 'roadm b', 'roadm f', 'roadm h']]], [['1', '2', '3'], ['2', '3'], ['b'], ['STRICT'], 'fail', []], [['1', '2'], ['2', '3'], [], [], 'pass', [['roadm a', 'roadm c', 'roadm d', 'roadm e', 'roadm g'], ['roadm c', 'roadm f'], ['roadm a', 'roadm b', 'roadm f', 'roadm h']]], [['1', '2'], ['2', '3'], ['roadm e'], ['LOOSE'], 'pass', [['roadm a', 'roadm c', 'roadm d', 'roadm e', 'roadm g'], ['roadm c', 'roadm f'], ['roadm a', 'roadm b', 'roadm f', 'roadm h']]], [['1', '2'], ['2', '3'], ['roadm c | roadm f'], ['LOOSE'], 'pass', [['roadm a', 'roadm c', 'roadm d', 'roadm e', 'roadm g'], ['roadm c', 'roadm f'], ['roadm a', 'roadm b', 'roadm f', 'roadm h']]]]) def test_create_disjunction(test_setup, dis1, dis2, node_list1, loose_list1, result, expected_paths): """verifies that the expected result is obtained for a set of particular constraints: in particular, verifies that: - multiple disjunction constraints are correcly handled - in case a loose constraint can not be met, the first alternate candidate is selected instead of the last one (last case). """ network, equipment = test_setup json_data = { 'synchronization': [{ 'synchronization-id': 'x', 'svec': { 'relaxable': 'false', 'disjointness': 'node link', 'request-id-number': dis1 } }, { 'synchronization-id': 'y', 'svec': { 'relaxable': 'false', 'disjointness': 'node link', 'request-id-number': dis2 } }]} dsjn = disjunctions_from_json(json_data) bdir = False rqs = create_rq(equipment, 'trx a', 'trx g', bdir, node_list1, loose_list1, '1') +\ create_rq(equipment, 'trx c', 'trx f', bdir, [], [], '2') +\ create_rq(equipment, 'trx a', 'trx h', bdir, [], [], '3') if result == 'fail': with pytest.raises(DisjunctionError): paths = compute_path_dsjctn(network, equipment, rqs, dsjn) else: paths = compute_path_dsjctn(network, equipment, rqs, dsjn) path_names = [] for path in paths: roadm_names = [e.uid for e in path if isinstance(e, Roadm)] path_names.append(roadm_names) assert path_names == expected_paths # if loose, one path can be returned @pytest.fixture() def request_set(): """ creates default request dict """ return { # 'request_id': '0', 'source': 'trx a', 'bidir': False, 'destination': 'trx g', 'trx_type': 'Voyager', 'spacing': 50e9, 'nodes_list': [], 'loose_list': [], 'f_min': 191.1e12, 'f_max': 196.3e12, 'nb_channel': None, 'power': 1e-3, 'tx_power': 1e-3, 'path_bandwidth': 200e9} @pytest.mark.parametrize( 'ids, modes, req_n, req_m, disjunction, final_ids, final_ns, final_ms, final_path_bandwidths', # requests that should be correctly aggregated: [(['a', 'b', 'c', 'd'], ['mode 1', 'mode 1', 'mode 1', 'mode 1'], [[0], [16], [32], [48]], [[8], [8], [8], [8]], [[]], ['d | c | b | a'], [[48, 32, 16, 0]], [[8, 8, 8, 8]], [800e9]), (['a', 'b', 'c', 'd'], ['mode 1', 'mode 1', 'mode 1', 'mode 1'], [[0, 8], [16, 24], [32, 40], [48]], [[4, 4], [4, 4], [4, 4], [8]], [[]], ['d | c | b | a'], [[48, 32, 40, 16, 24, 0, 8]], [[8, 4, 4, 4, 4, 4, 4]], [800e9]), (['a', 'b', 'c', 'd'], ['mode 1', 'mode 1', 'mode 1', 'mode 1'], [[0, 8], [None, 24], [32, 40], [None]], [[4, 4], [4, 4], [4, 4], [None]], [[]], ['d | c | b | a'], [[None, 32, 40, None, 24, 0, 8]], [[None, 4, 4, 4, 4, 4, 4]], [800e9]), # 'a' and 'b' have same constraint and can be aggregated (['a', 'b', 'c', 'd'], ['mode 1', 'mode 1', 'mode 1', 'mode 1'], [[0], [16], [32], [48]], [[8], [8], [8], [8]], [['c', 'd']], ['b | a', 'c', 'd'], [[16, 0], [32], [48]], [[8, 8], [8], [8]], [400e9, 200e9, 200e9]), (['a', 'b', 'c', 'd'], ['mode 1', 'mode 1', 'mode 1', 'mode 1'], [[0], [16], [32], [48]], [[8], [8], [8], [8]], [['a', 'd'], ['b', 'd']], ['b | a', 'c', 'd'], [[16, 0], [32], [48]], [[8, 8], [8], [8]], [400e9, 200e9, 200e9]), # requests that should not be aggregated: (['a', 'b', 'c', 'd'], [None, None, None, 'mode 1'], [[0, 8], [None, 24], [32, 40], [None]], [[4, 4], [4, 4], [4, 4], [None]], [[]], ['a', 'b', 'c', 'd'], [[0, 8], [None, 24], [32, 40], [None]], [[4, 4], [4, 4], [4, 4], [None]], [200e9, 200e9, 200e9, 200e9]), (['a', 'b', 'c', 'd'], ['mode 1', 'mode 1', 'mode 1', 'mode 1'], [[0], [16], [32], [48]], [[8], [8], [8], [8]], [['c', 'd', 'a']], ['a', 'b', 'c', 'd'], [[0], [16], [32], [48]], [[8], [8], [8], [8]], [200e9, 200e9, 200e9, 200e9]), ]) def test_aggregation(ids, modes, req_n, req_m, disjunction, final_ids, final_ns, final_ms, final_path_bandwidths, request_set): """ tests that identical requests are correctly aggregated (included frequency slots merging) if mode is not defined, requests must not be merged, if requests are in a synchronization vector, they should not be merged """ equipment = load_equipment(EQPT_LIBRARY_NAME, EXTRA_CONFIGS) requests = [] for request_id, mode, req_n, req_m in zip(ids, modes, req_n, req_m): params = request_set params['request_id'] = request_id params['trx_mode'] = mode params['effective_freq_slot'] = [{'N': n, 'M': m} for n, m in zip(req_n, req_m)] trx_params = trx_mode_params(equipment, params['trx_type'], params['trx_mode'], True) params.update(trx_params) requests.append(PathRequest(**params)) params = { 'relaxable': False, 'link_diverse': True, 'node_diverse': True } disjunctions = [] i = 0 for vector in disjunction: params['disjunctions_req'] = vector params['disjunction_id'] = i disjunctions.append(Disjunction(**params)) i += 1 requests, disjunctions = requests_aggregation(requests, disjunctions) print(disjunctions) print(requests) i = 0 for final_id, final_n, final_m, final_path_bandwidth in zip(final_ids, final_ns, final_ms, final_path_bandwidths): assert requests[i].request_id == final_id assert requests[i].N == final_n assert requests[i].M == final_m assert requests[i].path_bandwidth == final_path_bandwidth i += 1