import ipaddress
import os
import pathlib
from copy import deepcopy
from functools import lru_cache
import pytest
from lxml import etree
import jsonschema
from inventory_provider import nokia
PORT_SCHEMA = {
'$schema': 'https://json-schema.org/draft/2020-12/schema',
'type': 'object',
'properties': {
'port-id': {'type': 'string'},
'admin-state': {'type': 'string'},
'description': {'type': 'string'},
'speed': {'type': 'integer'},
'speed-unit': {'type': 'string', 'enum': ['Gbps', 'Unknown']}
},
'required': ['port-id', 'description', 'admin-state'],
'additionalProperties': True
}
PEERS_SCHEMA = {
'$schema': 'https://json-schema.org/draft/2020-12/schema',
'type': 'object',
'properties': {
'address': {'type': 'string'},
'remote-asn': {'type': 'integer'},
'local-asn': {'type': 'integer'},
'description': {'type': 'string'},
'group': {'type': 'string'},
'instance': {'type': 'string'}
},
'required': ['address', 'group'],
'additionalProperties': False
}
@lru_cache
def _load_xml_doc(filename):
full_filename = os.path.join(
os.path.dirname(__file__),
'data',
'nokia',
filename)
return nokia.remove_xml_namespaces(etree.parse(full_filename))
def _load_netconf_config(hostname):
return _load_xml_doc(f'{hostname}-netconf.xml')
def _load_state_doc(hostname):
return _load_xml_doc(f'{hostname}-state.xml')
state_doc = nokia.remove_xml_namespaces(etree.parse(pathlib.Path(__file__).parent.joinpath(
'data/nokia/rt0.lon.uk.lab.office.geant.net-netconf-nokia-state.xml')))
@pytest.mark.parametrize('hostname,expected_bundles', [
('rt0.ams.nl.lab.office.geant.net', {
'lag-1': ['1/1/c2/1'],
'lag-2': ['1/1/c5/1'],
'lag-4': ['1/1/c8/1'],
'lag-5': ['1/1/c11/1', '2/1/c11/1'],
'lag-6': ['2/1/c8/1'],
'lag-11': ['1/1/c2/3']}),
('rt0.ams.nl.geant.net', {
'lag-5': ['1/1/c19/1', '2/1/c19/1'],
'lag-8': ['1/1/c1/1', '1/1/c5/1'],
'lag-RL': ['9/1/c1/1']}),
('rt0.fra.de.geant.net', {'lag-1': ['1/1/c2/1', '1/1/c2/2']}),
('rt0.gen.ch.geant.net', {'lag-1': ['1/1/c2/1', '1/1/c2/2']}),
('rt0.lon.uk.geant.net', {'lag-1': ['1/1/c2/1', '1/1/c2/2']}),
('rt0.lon2.uk.geant.net', {'lag-1': ['2/1/c2/1', '2/1/c2/2']}),
])
def test_get_lags(hostname, expected_bundles):
netconf_doc = _load_netconf_config(hostname)
lags = {lag['name']: lag for lag in nokia.get_lags_config(netconf_doc)}
assert set(lags.keys()) == set(expected_bundles.keys())
for ex_name, ex_ports in expected_bundles.items():
assert lags[ex_name]['ports'] == ex_ports
@pytest.mark.parametrize(
'hostname,all_expected_data', [
('rt0.ams.nl.lab.office.geant.net', (
('lag-1.0', {'62.40.119.88/31'}, {'2001:799:1ab:2::31/126'}),
('lag-2.0', {'62.40.119.90/31'}, {'2001:799:1ab:2::41/126'}),
('lag-4.0', {'62.40.119.98/31'}, {'2001:799:1ab:2::55/126'}),
('lag-5.0', {'62.40.119.100/31'}, {'2001:799:1ab:2::59/126'}),
('lag-6.0', {'62.40.119.102/31'}, {'2001:799:1ab:2::5d/126'}),
('system', {'62.40.119.6/32'}, {'2001:799:1ab::6/128'}),
('lag-11:100', {'62.40.124.217/30'}, {'2001:798:14:10aa::9/126'}),
('LAG-11.333', {'83.97.88.245/30'}, {'2001:7f8:3c::35/126'}),
('LAG-11.111', {'62.40.126.184/31'}, {'2001:798:111:1::31/126'}),
)),
('rt0.ams.nl.geant.net', (
('lag-5.0', {'62.40.98.39/31'}, {'2001:798:cc::62/126'}),
('lag-8.0', {'62.40.98.20/31'}, {'2001:798:cc::21/126'}),
('system', {'62.40.96.16/32'}, {'2001:798:aa:1::8/128'}),
)),
('rt0.fra.de.geant.net', (
('lag-1.0', {'62.40.98.73/31'}, {'2001:798:cc::72/126'}),
('system', {'62.40.96.26/32'}, {'2001:798:aa:1::a/128'}),
)),
('rt0.gen.ch.geant.net', (
('lag-1.0', {'62.40.98.77/31'}, {'2001:798:cc::7a/126'}),
('system', {'62.40.96.29/32'}, {'2001:798:aa:1::20/128'}),
)),
('rt0.lon.uk.geant.net', (
('lag-1.0', {'62.40.98.60/31'}, {'2001:798:cc::65/126'}),
('system', {'62.40.96.21/32'}, {'2001:798:aa:1::9/128'}),
)),
('rt0.lon2.uk.geant.net', (
('lag-1.0', {'62.40.98.62/31'}, {'2001:798:cc::69/126'}),
('system', {'62.40.96.25/32'}, {'2001:798:aa:1::b/128'}),
)),
])
def test_interface_info(hostname, all_expected_data):
netconf_doc = _load_netconf_config(hostname=hostname)
interfaces_by_id = {ifc['interface-name']: ifc for ifc in
nokia.get_interfaces_config(netconf_doc)}
assert len(interfaces_by_id) == len(all_expected_data)
for expected_data in all_expected_data:
ifc = interfaces_by_id[expected_data[0]]
assert ifc['interface-name'] == expected_data[0]
assert set(ifc['ipv4']) == expected_data[1]
assert set(ifc['ipv6']) == expected_data[2]
@pytest.mark.parametrize('hostname,port_count', [
('rt0.ams.nl.lab.office.geant.net', 65),
('rt0.ams.nl.geant.net', 66),
('rt0.fra.de.geant.net', 58),
('rt0.gen.ch.geant.net', 141),
('rt0.lon.uk.geant.net', 58),
('rt0.lon2.uk.geant.net', 139),
])
def test_get_ports(hostname, port_count):
netconf_doc = _load_netconf_config(hostname=hostname)
ports = list(nokia.get_ports_config(netconf_doc))
assert len(ports) == port_count
for port in ports:
jsonschema.validate(port, PORT_SCHEMA)
def test_get_port_state():
ports = {p['port-id']: p for p in nokia.get_ports_state(state_doc)}
assert len(ports) == 197
found_port_ids = {p_id for p_id in ports}
ports_sample = {
'1/1/c2', '1/1/c2/1', '1/1/c2/2', '1/1/c2/3', '1/1/c7', '1/1/c7/1',
'1/1/c8', '1/1/c8/1', '1/1/c9', '1/1/c9/1', '1/1/c13', '1/1/c13/1',
'2/1/c7', '2/1/c7/1', '2/1/c8', '2/1/c8/1', '2/1/c13', '2/1/c13/1',
}
assert found_port_ids > ports_sample
expected_up_ports = {
'1/1/c2', '1/1/c5', '1/1/c5/1', '1/1/c7', '1/1/c8', '1/1/c9', '1/1/c13',
'1/1/c13/1', '1/1/c19', '1/1/c26', '1/1/c26/1', '1/1/c27', '1/1/c27/1',
'1/1/c30', '2/1/c7', '2/1/c8', '2/1/c13', '2/1/c13/1', 'A/1', 'B/1'
}
assert {k for k, v in ports.items() if v['oper-state'] == 'up'} == expected_up_ports
def test_get_lag_state():
lags = {x['name']: x for x in nokia.get_lags_state(state_doc)}
found_names = {lag_name for lag_name in lags}
expected_names = {'lag-1', 'lag-2', 'lag-3'}
assert found_names == expected_names
expected_up_lags = {'lag-2'}
assert {k for k, v in lags.items() if v['oper-state'] == 'up'} == expected_up_lags
def test_get_interface_state():
interfaces = {x['interface-name']: x for x in nokia.get_interfaces_state(state_doc)}
found_names = {interface_name for interface_name in interfaces}
expected_names = {'system', 'lag-1.0', 'lag-2.0', 'lag-3.0', 'exfo400', 'guy',
'exfo400-100', 'management'}
assert found_names == expected_names
expected_up_interfaces = {'system', 'lag-2.0', 'guy', 'exfo400-100', 'management'}
assert {k for k, v in interfaces.items() if
v['oper-state'] == 'up'} == expected_up_interfaces
def test_snmp_index():
interfaces = nokia.get_interfaces_state(state_doc)
expected = {
"system": 1,
"lag-1.0": 2,
"lag-2.0": 3,
"lag-3.0": 4,
"exfo400": 5,
"guy": 6,
"exfo400-100": 7,
"management": 1280,
}
assert {ifc["interface-name"]: ifc["if-index"] for ifc in interfaces} == expected
@pytest.mark.parametrize('hostname', [
'rt0.ams.nl.lab.office.geant.net',
'rt0.ams.nl.geant.net',
'rt0.fra.de.geant.net',
'rt0.gen.ch.geant.net',
'rt0.lon.uk.geant.net',
'rt0.lon2.uk.geant.net'
])
def test_get_peers(hostname):
netconf_doc = _load_netconf_config(hostname)
all_peers_from_doc = set()
all_peers_from_call = set()
for neighbor_element in netconf_doc.xpath('//bgp/neighbor'):
address = neighbor_element.find('ip-address').text
all_peers_from_doc.add(ipaddress.ip_address(address).exploded)
for peer in nokia.get_all_vprn_peers(netconf_doc):
jsonschema.validate(peer, PEERS_SCHEMA)
all_peers_from_call.add(peer['address'])
router_peers_schema = deepcopy(PEERS_SCHEMA)
del router_peers_schema['properties']['instance']
for peer in nokia.get_router_peers(netconf_doc):
jsonschema.validate(peer, PEERS_SCHEMA)
all_peers_from_call.add(peer['address'])
assert all_peers_from_doc == all_peers_from_call