scottk/oopt-gnpy
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/oopt-gnpy.git synced 2025-10-29 01:02:32 +00:00
Code Issues Packages Projects Releases 10 Wiki Activity
Files
master
oopt-gnpy/tests/test_logger.py
EstherLerouzic f2039fbe1c fix: use loaded json instead of Path for extra configs 
In order to be used by API.

Co-authored-by: Renato Ambrosone <renato.ambrosone@polito.it>

Signed-off-by: EstherLerouzic <esther.lerouzic@orange.com>
Change-Id: I12111427c8a90b85b3158cdd95f4ee771cb39316
2025-09-26 11:17:45 +02:00

595 lines
21 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# SPDX-License-Identifier: BSD-3-Clause
# test_logger
# Copyright (C) 2025 Telecom Infra Project and GNPy contributors
# see AUTHORS.rst for a list of contributors
"""
Checks that loggers works as expected
"""
from pathlib import Path
import re
import pytest
from gnpy.core.exceptions import ConfigurationError, ServiceError, EquipmentConfigError, ParametersError, \
NetworkTopologyError
from gnpy.tools.json_io import SI, Roadm, Amp, load_equipment, requests_from_json, network_from_json, \
load_network, load_requests, load_json
from gnpy.tools.convert import xls_to_json_data
TEST_DIR = Path(__file__).parent
EQPT_FILENAME = TEST_DIR / 'data/eqpt_config.json'
MULTIBAND_EQPT_FILENAME = TEST_DIR / 'data/eqpt_config_multiband.json'
DATA_DIR = TEST_DIR / 'data'
EXTRA_CONFIGS = {"std_medium_gain_advanced_config.json": load_json(DATA_DIR / "std_medium_gain_advanced_config.json")}
def test_jsonthing(caplog):
"""Check that a missing key correctly raises an info
"""
json_data = {
"baud_rate": 32e9,
"f_max": 196.1e12,
"spacing": 50e9,
"power_dbm": 0,
"power_range_db": [0, 0, 1],
"roll_off": 0.15,
"tx_osnr": 40,
"sys_margins": 2
}
_ = SI(**json_data)
expected_msg = '\n\tWARNING missing f_min attribute in eqpt_config.json[SI]\n' \
+ '\tdefault value is f_min = 191350000000000.0'
assert expected_msg in caplog.text
def wrong_equipment():
"""Creates list of malformed equipments
"""
data = []
data.append({
"error": EquipmentConfigError,
"equipment": Roadm,
"json_data": {
"target_pch_out_db": -20,
"target_out_mWperSlotWidth": 3.125e-4,
"add_drop_osnr": 38,
"pmd": 0,
"pdl": 0,
"restrictions": {
"preamp_variety_list": [],
"booster_variety_list": []
}
},
"expected_msg": "Only one equalization type should be set in ROADM, found: target_pch_out_db,"
+ " target_out_mWperSlotWidth"
})
data.append({
"error": EquipmentConfigError,
"equipment": Roadm,
"json_data": {
"add_drop_osnr": 38,
"pmd": 0,
"pdl": 0,
"restrictions": {
"preamp_variety_list": [],
"booster_variety_list": []
}
},
"expected_msg": "No equalization type set in ROADM"
})
return data
@pytest.mark.parametrize('error, equipment, json_data, expected_msg',
[(e['error'], e['equipment'], e['json_data'], e['expected_msg']) for e in wrong_equipment()])
def test_wrong_equipment(caplog, error, equipment, json_data, expected_msg):
"""
"""
with pytest.raises(EquipmentConfigError, match=expected_msg):
_ = equipment(**json_data)
@pytest.mark.parametrize('xls_service_filename, xls_topo_filename, expected_msg',
[('wrong_service.xlsx', 'testTopology.xls',
"Service error: Request Id: 0 - could not find tsp : 'Voyager' with mode: 'Mode 10' "
+ "in eqpt library \nComputation stopped."),
('wrong_service_type.xlsx', 'testTopology.xls',
"Service error: Request Id: 0 - could not find tsp : 'Galileo' with mode: 'mode 1' "
+ "in eqpt library \nComputation stopped.")])
def test_wrong_xls_service(xls_service_filename, xls_topo_filename, expected_msg):
"""
"""
equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
network = load_network(DATA_DIR / xls_topo_filename, equipment)
with pytest.raises(ServiceError, match=expected_msg):
_ = load_requests(DATA_DIR / xls_service_filename, equipment, False, network, DATA_DIR / xls_topo_filename)
def wrong_amp():
"""Creates list of malformed equipments
"""
data = []
data.append({
"error": EquipmentConfigError,
"json_data": {
"type_variety": "test_fixed_gain",
"type_def": "fixed_gain",
"gain_flatmax": 21,
"gain_min": 20,
"p_max": 21,
"allowed_for_design": True
},
"expected_msg": "missing nf0 value input for amplifier: test_fixed_gain in equipment config"
})
data.append({
"error": EquipmentConfigError,
"json_data": {
"type_variety": "test",
"type_def": "variable_gain",
"gain_flatmax": 25,
"gain_min": 15,
"p_max": 21,
"nf_min": 5.8,
"out_voa_auto": False,
"allowed_for_design": True
},
"expected_msg": "missing nf_min or nf_max value input for amplifier: test in equipment config"
})
data.append({
"error": EquipmentConfigError,
"json_data": {
"type_variety": "medium+high_power",
"type_def": "dual_stage",
"gain_min": 25,
"preamp_variety": "std_medium_gain",
"allowed_for_design": False
},
"expected_msg": "missing preamp/booster variety input for amplifier: medium+high_power in equipment config"
})
return data
@pytest.mark.parametrize('error, json_data, expected_msg',
[(e['error'], e['json_data'], e['expected_msg']) for e in wrong_amp()])
def test_wrong_amp(error, json_data, expected_msg):
"""
"""
with pytest.raises(error, match=re.escape(expected_msg)):
_ = Amp.from_json(EQPT_FILENAME, **json_data)
def wrong_requests():
"""Creates list of malformed requests
"""
data = []
data.append({
'error': ConfigurationError,
'json_data': {
"path-request": [{
"request-id": "imposed_mode",
"source": "trx Brest_KLA",
"destination": "trx Vannes_KBE",
"src-tp-id": "trx Brest_KLA",
"dst-tp-id": "trx Vannes_KBE",
"bidirectional": False,
"path-constraints": {
"te-bandwidth": {
"technology": "flexi-grid",
"trx_type": "test_offset",
"trx_mode": "mode 3",
"spacing": 75000000000.0,
"path_bandwidth": 100000000000.0
}
}
}]
},
'expected_msg': 'Equipment Config error in imposed_mode: '
+ 'Could not find transponder "test_offset" in equipment library'
})
data.append({
'error': ServiceError,
'json_data': {
"path-request": [{
"request-id": "Missing_type",
"source": "trx Brest_KLA",
"destination": "trx Vannes_KBE",
"src-tp-id": "trx Brest_KLA",
"dst-tp-id": "trx Vannes_KBE",
"bidirectional": False,
"path-constraints": {
"te-bandwidth": {
"technology": "flexi-grid",
"trx_type": None,
"spacing": 75000000000.0,
"path_bandwidth": 100000000000.0
}
}
}]},
'expected_msg': 'Request Missing_type has no transceiver type defined'
})
data.append({
'error': ServiceError,
'json_data': {
"path-request": [{
"request-id": "wrong_spacing",
"source": "trx Brest_KLA",
"destination": "trx Vannes_KBE",
"src-tp-id": "trx Brest_KLA",
"dst-tp-id": "trx Vannes_KBE",
"bidirectional": False,
"path-constraints": {
"te-bandwidth": {
"technology": "flexi-grid",
"trx_type": "Voyager",
"trx_mode": "mode 2",
"spacing": 50000000000.0,
"path_bandwidth": 100000000000.0
}
}
}]},
'expected_msg': 'Request wrong_spacing has spacing below transponder Voyager mode 2 min spacing'
+ ' value 75.0GHz.\nComputation stopped'
})
data.append({
'error': ServiceError,
'json_data': {
"path-request": [{
"request-id": "Wrong_nb_channel",
"source": "trx Brest_KLA",
"destination": "trx Vannes_KBE",
"src-tp-id": "trx Brest_KLA",
"dst-tp-id": "trx Vannes_KBE",
"bidirectional": False,
"path-constraints": {
"te-bandwidth": {
"technology": "flexi-grid",
"trx_type": "Voyager",
"trx_mode": "mode 2",
"spacing": 75000000000.0,
"max-nb-of-channel": 150,
"path_bandwidth": 100000000000.0
}
}
}]},
'expected_msg': 'Requested channel number 150, baud rate 66.0 GHz'
+ ' and requested spacing 75.0GHz is not consistent with frequency range'
+ ' 191.35 THz, 196.1 THz.'
+ ' Max recommanded nb of channels is 63.'
})
data.append({
'error': ServiceError,
'json_data': {
"path-request": [{
"request-id": "Wrong_M",
"source": "trx Brest_KLA",
"destination": "trx Vannes_KBE",
"src-tp-id": "trx Brest_KLA",
"dst-tp-id": "trx Vannes_KBE",
"bidirectional": False,
"path-constraints": {
"te-bandwidth": {
"technology": "flexi-grid",
"trx_type": "Voyager",
"trx_mode": "mode 2",
"spacing": 75000000000.0,
"effective-freq-slot": [
{
"N": -208,
"M": 4
}
],
"path_bandwidth": 100000000000.0
}
}
}]},
'expected_msg': 'Requested M [{\'N\': -208, \'M\': 4}] number of slots for request Wrong_M '
+ 'support 0 nb of channels while 1 are required to support request 100.0 Gbit/s'
+ ' with Voyager mode 2'
})
return data
@pytest.mark.parametrize('error, json_data, expected_msg',
[(e['error'], e['json_data'], e['expected_msg']) for e in wrong_requests()])
def test_json_request(error, json_data, expected_msg):
"""
Check that a missing key is correctly raisong the logger
"""
equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
with pytest.raises(error, match=re.escape(expected_msg)):
_ = requests_from_json(json_data, equipment)
def wrong_element():
"""
"""
data = []
data.append({
"error": ConfigurationError,
"json_data": {
"elements": [{
"uid": "roadm SITE2",
"type": "Roadm",
"params": {
"target_pch_out_db": -20,
"target_out_mWperSlotWidth": 3.125e-4,
},
"metadata": {
"location": {
"latitude": 2.0,
"longitude": 3.0,
"city": "SITE2",
"region": "RLD"
}
}
}],
"connections": []
},
"expected_msg": "ROADM roadm SITE2: invalid equalization settings"
})
data.append({
"error": ConfigurationError,
"json_data": {
"elements": [{
"uid": "east edfa in ILA2 to SITE2",
"type": "Edfa",
"type_variety": "not_valid_variety",
"metadata": {
"location": {
"latitude": 2.0,
"longitude": 0.0,
"city": "ILA2",
"region": "RLD"
}
}
}],
"connections": []
},
"expected_msg": "The Edfa of variety type not_valid_variety was not recognized:"
+ "\nplease check it is properly defined in the eqpt_config json file"
})
data.append({
"error": ParametersError,
"json_data": {
"elements": [{
"uid": "fiber (ILA2 → ILA1)",
"type": "Fiber",
"type_variety": "SSMF",
"params": {
"length": 100.0,
"loss_coef": 0.2,
"att_in": 0,
"con_in": 0,
"con_out": 0
},
"metadata": {
"location": {
"latitude": 2.0,
"longitude": 1.5,
"city": None,
"region": None
}
}
}],
"connections": []
},
"expected_msg": "Config error in fiber (ILA2 → ILA1): "
+ "Fiber configurations json must include \'length_units\'. Configuration: "
+ "{\'length\': 100.0, \'loss_coef\': 0.2, \'att_in\': 0, \'con_in\': 0, \'con_out\': 0, "
+ "\'pmd_coef_defined': False, \'pmd_coef\': 1.265e-15, "
+ "\'type_variety\': \'SSMF\', \'dispersion\': 1.67e-05, \'effective_area\': 8.3e-11}"
})
return data
@pytest.mark.parametrize('error, json_data, expected_msg',
[(e['error'], e['json_data'], e['expected_msg']) for e in wrong_element()])
def test_json_network(error, json_data, expected_msg):
"""
Check that a missing key is correctly raisong the logger
"""
equipment = load_equipment(EQPT_FILENAME, EXTRA_CONFIGS)
with pytest.raises(error, match=re.escape(expected_msg)):
_ = network_from_json(json_data, equipment)
@pytest.mark.parametrize('input_filename, expected_msg',
[(DATA_DIR / 'wrong_topo_node.xlsx', 'XLS error: The following nodes are not referenced from the Links sheet.'
+ ' If unused, remove them from the Nodes sheet:\n - toto'),
(DATA_DIR / 'wrong_topo_link.xlsx', 'XLS error: The Links sheet references nodes that are not defined in the '
+ 'Nodes sheet:\n - ALB -> toto'),
(DATA_DIR / 'wrong_topo_link_header.xlsx', 'missing header Node Z'),
(DATA_DIR / 'wrong_topo_eqpt.xlsx', 'XLS error: The Eqpt sheet refers to nodes that are not defined in the '
+ 'Nodes sheet:\n - toto'),
(DATA_DIR / 'wrong_topo_duplicate_node.xlsx', 'Duplicate city: Counter({\'ALB\': 2, \'CHA_3\': 1})'),
(DATA_DIR / 'wrong_topo_duplicate_eqpt.xlsx', 'XLS error: Duplicate lines in Eqpt sheet: - ALB -> CHA_3'),
(DATA_DIR / 'wrong_topo_bad_eqpt.xlsx', 'XLS error: The Eqpt sheet references links that are not defined '
+ 'in the Links sheet:\n - toto -> CHA_3'),
(DATA_DIR / 'wrong_duplicate_link_reverse.xlsx', 'XLS error: links - (\'ila\', \'siteb\') are duplicate'),
(DATA_DIR / 'wrong_duplicate_eqpt_ila_reverse.xlsx', 'XLS error: Duplicate ILA eqpt definition in Eqpt sheet:'
+ ' - ila')])
def test_wrong_xlsx(input_filename, expected_msg):
"""Check that error and logs are correctly working
"""
with pytest.raises(NetworkTopologyError, match=re.escape(expected_msg)):
_ = xls_to_json_data(input_filename)
@pytest.mark.parametrize('input_filename, expected_msg',
[(DATA_DIR / 'wrong_node_type.xlsx', 'invalid node type (ILA) specified in Lannion_CAS, replaced by ROADM\n')])
def test_log_wrong_xlsx(caplog, input_filename, expected_msg):
"""Check that logs are correctly working
"""
_ = xls_to_json_data(input_filename)
assert expected_msg in caplog.text
def wrong_configs():
wrong_config = [[{
"uid": "Edfa1",
"type": "Edfa",
"type_variety": "std_medium_gain_multiband",
"operational": {
"gain_target": 21,
"delta_p": 3.0,
"out_voa": 3.0,
"in_voa": 0.0,
"tilt_target": 0.0,
"f_min": 186.2,
"f_max": 190.2
}},
ParametersError]
]
wrong_config.append([{
"uid": "[83/WR-2-4-SIG=>930/WRT-1-2-SIG]-PhysicalConn/2056",
"type": "Fiber",
"type_variety": "SSMF",
"params": {
"dispersion_per_frequency": {
"frequency": [
185.49234135667396e12,
186.05251641137855e12,
188.01312910284463e12,
189.99124726477024e12],
"value": [
1.60e-05,
1.67e-05,
1.7e-05,
1.8e-05]
},
"loss_coef": 2.85,
"length_units": "km",
"att_in": 0.0,
"con_in": 0.0,
"con_out": 0.0
}},
ParametersError
])
wrong_config.append([{
"uid": "east edfa in Lannion_CAS to Stbrieuc",
"metadata": {
"location": {
"city": "Lannion_CAS",
"region": "RLD",
"latitude": 2.0,
"longitude": 0.0
}
},
"type": "Multiband_amplifier",
"type_variety": "std_low_gain_multiband",
"amplifiers": [{
"type_variety": "std_low_gain",
"operational": {
"gain_target": 20.0,
"delta_p": 0,
"tilt_target": 0,
"out_voa": 0
}
}, {
"type_variety": "std_low_gain_L_ter",
"operational": {
"gain_target": 20.0,
"delta_p": 1,
"tilt_target": 0,
"out_voa": 1
}
}]},
ConfigurationError])
wrong_config.append([{
"uid": "east edfa in Lannion_CAS to Stbrieuc",
"metadata": {
"location": {
"city": "Lannion_CAS",
"region": "RLD",
"latitude": 2.0,
"longitude": 0.0
}
},
"type": "Edfa",
"type_variety": "std_low_gain_multiband",
"amplifiers": [{
"type_variety": "std_low_gain",
"operational": {
"gain_target": 20.0,
"delta_p": 0,
"tilt_target": 0,
"out_voa": 0
}
}, {
"type_variety": "std_low_gain_L",
"operational": {
"gain_target": 20.0,
"delta_p": 1,
"tilt_target": 0,
"out_voa": 1
}
}]},
ParametersError])
wrong_config.append([{
"uid": "east edfa in Lannion_CAS to Stbrieuc",
"metadata": {
"location": {
"city": "Lannion_CAS",
"region": "RLD",
"latitude": 2.0,
"longitude": 0.0
}
},
"type": "Multiband_amplifier",
"type_variety": "std_low_gain_multiband",
"amplifiers": [{
"type_variety": "std_low_gain",
"operational": {
"gain_target": 20.0,
"delta_p": 0,
"tilt_target": 0,
"out_voa": 0
}
}, {
"type_variety": "std_low_gain_L",
"operational": {
"gain_target": 20.0,
"delta_p": 1,
"tilt_target": 0,
"out_voa": 1
}
}, {
"type_variety": "std_low_gain",
"operational": {
"gain_target": 20.0,
"delta_p": 1,
"tilt_target": 0,
"out_voa": 1
}
}]},
ParametersError])
return wrong_config
@pytest.mark.parametrize('el_config, error_type', wrong_configs())
def test_wrong_multiband(el_config, error_type):
equipment = load_equipment(MULTIBAND_EQPT_FILENAME, EXTRA_CONFIGS)
fused_config = {
"uid": "[83/WR-2-4-SIG=>930/WRT-1-2-SIG]-Tl/9300",
"type": "Fused",
"params": {
"loss": 20
}
}
json_data = {
"elements": [
el_config,
fused_config
],
"connections": []
}
with pytest.raises(error_type):
_ = network_from_json(json_data, equipment)
Reference in New Issue View Git Blame Copy Permalink