test_snmp_handling.py

import json
import os
from io import StringIO

import jsonschema
import pytest

from inventory_provider import snmp
from inventory_provider import config

OID_TEST_CONFIG = """#
#   This file is located in dbupdates/conf and is used by scripts under dbupdates/scripts.
#   It holds OID values for retrieving details of a router.
#

## IPv4
v4Address=.1.3.6.1.2.1.4.20.1.1
v4InterfaceOID=.1.3.6.1.2.1.4.20.1.2
v4InterfaceName=.1.3.6.1.2.1.31.1.1.1.1
v4Mask=.1.3.6.1.2.1.4.20.1.3

## IPv6
v6AddressAndMask=.1.3.6.1.2.1.55.1.8.1.2
v6InterfaceName=.1.3.6.1.2.1.55.1.5.1.2
"""  # noqa E501


@pytest.fixture
def snmp_walk_responses():
    test_data_dir = os.path.join(os.path.dirname(__file__), "data")
    test_data_filename = os.path.join(test_data_dir, "snmp-walk.json")
    with open(test_data_filename) as f:
        return json.loads(f.read())


@pytest.fixture
def oid_config():
    with StringIO(OID_TEST_CONFIG) as s:
        return config._load_oids(s)


def test_snmp_interfaces(mocker, oid_config, snmp_walk_responses):

    expected_result_schema = {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "type": "array",
        "items": {
            "type": "object",
            "properties": {
                "v4Address": {"type": "string"},
                "v4Mask": {"type": "string"},
                "v4InterfaceName": {"type": "string"},
                "v6Address": {"type": "string"},
                "v6Mask": {"type": "string"},
                "v6InterfaceName": {"type": "string"},
                "index": {"type": "string"}
            },
            "required": ["index"],
            "additionalProperties": False
        }
    }

    def _mocked_walk(agent_hostname, community, base_oid):
        return [e for e in snmp_walk_responses
                if e['oid'].startswith(base_oid)]

    mocker.patch(
        'inventory_provider.snmp.walk',
        _mocked_walk)

    interfaces = snmp.get_router_interfaces(
        'ignored', 'ignored', {'oids': oid_config})
    interfaces = list(interfaces)

    jsonschema.validate(interfaces, expected_result_schema)
    assert interfaces, "interface list isn't empty"
    for ifc in interfaces:
        if 'v4Address' in ifc \
                and 'v4Mask' in ifc \
                and 'v4InterfaceName' in ifc:
            continue
        if 'v6Address' in ifc \
                and 'v6Mask' in ifc \
                and 'v6InterfaceName' in ifc:
            continue
        assert False, "address details not found in interface dict"