test_juniper.py

from datetime import datetime
from unittest.mock import call, patch

import pytest
from brian_polling_manager import influx
from brian_polling_manager.interface_stats.vendors import common, juniper
from lxml import etree
from ncclient.operations.rpc import RPCReply


@pytest.fixture
def get_netconf(data_dir):
    def _get_netconf(router_name):
        return juniper.get_netconf_interface_info_from_source_dir(
            router_name, source_dir=data_dir
        )

    return _get_netconf


@pytest.fixture
def juniper_router_doc_containing_every_field():
    doc = """\
<interface-information>
    <physical-interface>
        <name>ae12</name>
        <admin-status>up</admin-status>
        <oper-status>up</oper-status>
        <traffic-statistics>
            <input-bytes>1</input-bytes>
            <input-packets>2</input-packets>
            <output-bytes>3</output-bytes>
            <output-packets>4</output-packets>
            <ipv6-transit-statistics>
                <input-bytes>5</input-bytes>
                <input-packets>6</input-packets>
                <output-bytes>7</output-bytes>
                <output-packets>8</output-packets>
            </ipv6-transit-statistics>
        </traffic-statistics>
        <input-error-list>
            <input-errors>11</input-errors>
            <input-discards>12</input-discards>
            <input-fifo-errors>13</input-fifo-errors>
            <input-drops>14</input-drops>
            <framing-errors>15</framing-errors>
            <input-resource-errors>16</input-resource-errors>
        </input-error-list>
        <output-error-list>
            <output-errors>21</output-errors>
            <output-drops>22</output-drops>
            <output-resource-errors>23</output-resource-errors>
            <output-fifo-errors>24</output-fifo-errors>
            <output-collisions>25</output-collisions>
            <output-discards>26</output-discards>
        </output-error-list>
        <ethernet-mac-statistics>
            <input-crc-errors>31</input-crc-errors>
            <output-crc-errors>32</output-crc-errors>
            <input-total-errors>33</input-total-errors>
            <output-total-errors>34</output-total-errors>
        </ethernet-mac-statistics>
        <ethernet-pcs-statistics>
            <bit-error-seconds>41</bit-error-seconds>
            <errored-blocks-seconds>42</errored-blocks-seconds>
        </ethernet-pcs-statistics>
        <logical-interface>
            <name>ae12.1</name>
            <traffic-statistics>
                <input-bytes>51</input-bytes>
                <input-packets>52</input-packets>
                <output-bytes>53</output-bytes>
                <output-packets>54</output-packets>
                <ipv6-transit-statistics>
                    <input-bytes>55</input-bytes>
                    <input-packets>56</input-packets>
                    <output-bytes>57</output-bytes>
                    <output-packets>58</output-packets>
                </ipv6-transit-statistics>
            </traffic-statistics>
        </logical-interface>
    </physical-interface>
</interface-information>
"""
    return etree.fromstring(doc)


def test_verify_all_interfaces_present(
    juniper_router_fqdn, juniper_inventory, get_netconf
):
    """
    verify that all the interfaces we expect to poll are available in the netconf data
    compares a snapshot of all netconf docs with a a snapshot of inventory
    /poller/interfaces (the snapshots were all taken around the same time)
    """

    if juniper_router_fqdn not in juniper_inventory:
        pytest.skip(f"{juniper_router_fqdn} has no expected polled interfaces")
    all_interfaces = juniper_inventory[juniper_router_fqdn]
    doc = get_netconf(juniper_router_fqdn)
    counters = juniper.interface_counters(doc, interfaces=all_interfaces)
    interfaces = {x["name"] for x in counters}
    missing = juniper_inventory[juniper_router_fqdn] - interfaces
    assert not missing


def test_interface_counters(juniper_router_doc_containing_every_field):
    result = list(
        juniper.interface_counters(
            juniper_router_doc_containing_every_field, interfaces=["ae12", "ae12.1"]
        )
    )
    assert result == [
        {
            "name": "ae12",
            "brian": {
                "ingressOctets": 1,
                "ingressPackets": 2,
                "egressOctets": 3,
                "egressPackets": 4,
                "ingressOctetsv6": 5,
                "ingressPacketsv6": 6,
                "egressOctetsv6": 7,
                "egressPacketsv6": 8,
                "ingressErrors": 11,
                "ingressDiscards": 12,
                "egressErrors": 21,
            },
            "errors": {
                "input_discards": 12,
                "input_fifo_errors": 13,
                "input_drops": 14,
                "input_framing_errors": 15,
                "input_resource_errors": 16,
                "output_drops": 22,
                "output_resource_errors": 23,
                "output_fifo_errors": 24,
                "output_collisions": 25,
                "output_discards": 26,
                "input_crc_errors": 31,
                "output_crc_errors": 32,
                "input_total_errors": 33,
                "output_total_errors": 34,
                "bit_error_seconds": 41,
                "errored_blocks_seconds": 42,
            },
        },
        {
            "name": "ae12.1",
            "brian": {
                "ingressOctets": 51,
                "ingressPackets": 52,
                "egressOctets": 53,
                "egressPackets": 54,
                "ingressOctetsv6": 55,
                "ingressPacketsv6": 56,
                "egressOctetsv6": 57,
                "egressPacketsv6": 58,
            },
        },
    ]


def test_juniper_router_docs_do_not_generate_errors(
    get_netconf, juniper_router_fqdn, caplog, juniper_inventory
):
    doc = get_netconf(juniper_router_fqdn)
    counters = list(
        juniper.interface_counters(
            doc, interfaces=juniper_inventory[juniper_router_fqdn]
        )
    )
    assert counters
    assert not [r for r in caplog.records if r.levelname in ("ERROR", "WARNING")]


def test_validate_interface_counters_and_influx_points_for_all_juniper_routers(
    juniper_router_fqdn, get_netconf, juniper_inventory, schemavalidate
):
    doc = get_netconf(juniper_router_fqdn)

    interfaces = list(
        juniper.interface_counters(
            doc, interfaces=juniper_inventory[juniper_router_fqdn]
        )
    )
    assert interfaces
    for ifc in interfaces:
        schemavalidate(ifc, common.INTERFACE_COUNTER_SCHEMA)
        schemavalidate(ifc["brian"], common.BRIAN_POINT_FIELDS_SCHEMA)

    bpoints = list(
        common.brian_points(
            juniper_router_fqdn,
            interfaces,
            timestamp=datetime.now(),
            measurement_name="blah",
        )
    )
    assert bpoints
    for point in bpoints:
        schemavalidate(point, influx.INFLUX_POINT)
        schemavalidate(point["fields"], common.BRIAN_POINT_FIELDS_SCHEMA)
        for value in point['fields'].values():
            assert isinstance(value, float)

    epoints = list(
        common.error_points(
            juniper_router_fqdn,
            interfaces,
            timestamp=datetime.now(),
            measurement_name="blah",
        )
    )
    assert epoints
    for point in epoints:
        schemavalidate(point, influx.INFLUX_POINT)
        schemavalidate(point["fields"], common.ERROR_POINT_FIELDS_SCHEMA)


class TestGetJuniperNetConf:
    RAW_RESPONSE_FILE = "raw-response-juniper-sample.xml"

    @pytest.fixture(autouse=True)
    def mocked_connect(self, data_dir):
        raw_response = data_dir.joinpath(self.RAW_RESPONSE_FILE).read_text()
        with patch.object(juniper, "netconf_connect") as mock:
            mock().__enter__().rpc.return_value = RPCReply(raw_response)
            mock.reset_mock()
            yield mock

    def test_calls_ncclient_with_params(self, mocked_connect):
        router_name = "some-router"
        juniper.get_netconf_interface_info(router_name, ssh_params={"ssh": "param"})

        mocked_connect.call_args = call(hostname="some-router", ssh="param", timeout=5)

        assert (
            mocked_connect().__enter__().rpc.call_args[0][0].tag
            == "get-interface-information"
        )

    def test_converts_rpc_response_to_xml(self):
        router_name = "some-router"

        doc = juniper.get_netconf_interface_info(
            router_name, ssh_params={"some": "param"}
        )

        assert doc.tag == "rpc-reply"