Skip to content
Snippets Groups Projects
Commit d3f3a9c9 authored by Sam Roberts's avatar Sam Roberts
Browse files

add unit tests for parsing speeds from interface-info documents

parent 841a124a
No related branches found
No related tags found
1 merge request!22Feature/pol1 744 fix speed issues
...@@ -2,9 +2,13 @@ import glob ...@@ -2,9 +2,13 @@ import glob
import json import json
import os import os
import re import re
from unittest.mock import patch
from lxml import etree from lxml import etree
import pytest import pytest
from ncclient.manager import make_device_handler, Manager
from ncclient.transport import SSHSession
from ncclient.xml_ import NCElement
import inventory_provider import inventory_provider
from inventory_provider import juniper from inventory_provider import juniper
...@@ -64,3 +68,41 @@ def netconf_doc(mocker, router, data_config): ...@@ -64,3 +68,41 @@ def netconf_doc(mocker, router, data_config):
'inventory_provider.juniper.Device', 'inventory_provider.juniper.Device',
MockedJunosDevice) MockedJunosDevice)
return juniper.load_config(router, data_config['ssh']) return juniper.load_config(router, data_config['ssh'])
# different library requires different mocking
@pytest.fixture
def interface_info_response(router):
filename = os.path.join(TEST_DATA_DIRNAME, 'interface_info', f'{router}.xml')
try:
with open(filename, 'r') as file:
data = file.read()
return data
except FileNotFoundError:
pytest.skip(f'no corresponding interface_info doc for {router}, skipping')
@pytest.fixture
def mocked_router(interface_info_response):
def _mocked_manager_connect(*args, **kwargs):
assert kwargs
device_params = kwargs["device_params"]
device_handler = make_device_handler(device_params)
session = SSHSession(device_handler)
return Manager(session, device_handler)
def _mocked_rpc_reply(*args, **kwargs):
device_handler = make_device_handler({'name': 'junos'})
rpc_reply = NCElement(
interface_info_response,
device_handler.transform_reply()
)
return rpc_reply
with patch('ncclient.manager.connect') as manager:
manager.side_effect = _mocked_manager_connect
with patch('ncclient.operations.third_party.juniper.rpc.ExecuteRpc._request') as rpc:
rpc.side_effect = _mocked_rpc_reply
with patch('ncclient.operations.session.CloseSession.request'):
yield
...@@ -2,6 +2,8 @@ import copy ...@@ -2,6 +2,8 @@ import copy
import ipaddress import ipaddress
import jsonschema import jsonschema
import pytest import pytest
from lxml import etree
from inventory_provider import juniper from inventory_provider import juniper
from inventory_provider.routes import msr from inventory_provider.routes import msr
...@@ -143,3 +145,11 @@ def test_interface_addresses_list(netconf_doc): ...@@ -143,3 +145,11 @@ def test_interface_addresses_list(netconf_doc):
addresses = list(juniper.interface_addresses(netconf_doc)) addresses = list(juniper.interface_addresses(netconf_doc))
jsonschema.validate(addresses, schema) jsonschema.validate(addresses, schema)
def test_interface_speeds(mocked_router, router, data_config):
interface_info = juniper.get_interface_info_for_router(router, data_config['ssh'])
speeds = juniper.get_interface_speeds(etree.fromstring(interface_info))
assert len(speeds) > 0 # should have any speeds at all
for speed in speeds:
assert len(speed) > 0 # no empty strings, every found speed should have a value
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment