test_sensu_checks.py

import copy

import jsonschema
import responses

from brian_polling_manager import inventory, sensu, interfaces

checks_match = sensu.AbstractCheck.match_check_dicts


@responses.activate
def test_load_checks(config, mocked_sensu):
    checks = sensu.load_all_checks(config["sensu"])
    assert len(checks) > 0  # test data has checks in it


@responses.activate
def test_check_lifecycle(config, mocked_sensu, mocked_inventory):
    new_check = interfaces.NetconfRouterCheck(
        config["sensu"]["interface-check"], "xyz"
    ).to_dict()

    check_name = new_check["metadata"]["name"]
    assert check_name not in mocked_sensu

    # create the new check
    sensu.create_check(config["sensu"], new_check)
    assert check_name in mocked_sensu

    # modify interval & update, then verify the correct call was made
    updated_check = copy.copy(new_check)
    updated_check["interval"] += 1
    sensu.update_check(config["sensu"], updated_check)
    assert mocked_sensu[check_name]["interval"] == updated_check["interval"]

    # delete the check and confirm the correct call was made
    sensu.delete_check(config["sensu"], check_name)
    assert check_name not in mocked_sensu


def test_router_check(config):
    check = interfaces.NetconfRouterCheck(
        config["sensu"]["interface-check"], "xyz"
    ).to_dict()
    assert check["command"] == (
        "/home/brian_checks/venv/get-interface-stats "
        "--config /var/lib/sensu/conf/get-interface-stats.config.json "
        "--juniper xyz --all"
    )
    assert check["proxy_entity_name"] == "xyz"
    assert check["metadata"]["name"] == "rtr-xyz"
    assert check["output_metric_format"] == ""
    assert check["output_metric_handlers"] == []


def test_snmp_interface_check(config):
    check = interfaces.SNMPInterfaceCheck(
        config["sensu"]["snmp-interface-check"],
        {"router": "bogus.router", "name": "bogus/1/1", "snmp-index": 123},
    ).to_dict()
    assert check["command"] == (
        "/var/lib/sensu/bin/counter2influx-v6.sh counters 0pBiFbD bogus.router "
        "bogus/1/1 123"
    )
    assert check["proxy_entity_name"] == "bogus.router"
    assert check["metadata"]["name"] == "ifc-bogus.router-bogus-1-1"
    assert check["output_metric_format"] == "influxdb_line"
    assert check["output_metric_handlers"] == ["influx-db-handler"]


@responses.activate
def test_snmp_interface_refresh(config, mocked_sensu, mocked_inventory):
    routers = inventory.load_interfaces("http://inventory1")
    result = interfaces.snmp_interfaces_refresh(
        config["sensu"], inventory_interfaces=routers
    )
    assert result == {"checks": 3, "input": 1, "created": 1, "updated": 0, "deleted": 3}


def test_nokia_router_check(config):
    check = interfaces.NetconfRouterCheck(
        config["sensu"]["interface-check"], "xyz", vendor="nokia"
    ).to_dict()
    assert check["command"] == (
        "/home/brian_checks/venv/get-interface-stats "
        "--config /var/lib/sensu/conf/get-interface-stats.config.json "
        "--nokia xyz --all"
    )
    assert check["proxy_entity_name"] == "xyz"
    assert check["metadata"]["name"] == "rtr-xyz"
    assert check["output_metric_format"] == ""
    assert check["output_metric_handlers"] == []


@responses.activate
def test_runs_idempotent(config, mocked_sensu, mocked_inventory):
    routers = inventory.load_interfaces("http://inventory1")
    interfaces.netconf_router_refresh(config["sensu"], inventory_interfaces=routers)
    sensu.clear_cached_values()
    result = interfaces.netconf_router_refresh(
        config["sensu"], inventory_interfaces=routers
    )
    assert result == {"checks": 3, "input": 3, "created": 0, "updated": 0, "deleted": 0}


class DummyCheck(sensu.AbstractCheck):

    def __init__(self, name, command, proxy_entity_name):
        super().__init__()
        self._name = name
        self._command = command
        self._proxy_entity_name = proxy_entity_name

    @sensu.AbstractCheck.name.getter
    def name(self):
        return self._name

    @sensu.AbstractCheck.command.getter
    def command(self):
        return self._command

    @sensu.AbstractCheck.proxy_entity_name.getter
    def proxy_entity_name(self):
        return self._proxy_entity_name


def test_check_dict_schema():
    c = DummyCheck(name="x", command="y", proxy_entity_name="z")
    jsonschema.validate(c.to_dict(), sensu.AbstractCheck.CHECK_DICT_SCHEMA)


def test_has_handler_and_metric_format():
    c = DummyCheck(name="x", command="y", proxy_entity_name="z")
    assert c.to_dict()["output_metric_handlers"] == ["influx-db-handler"]
    assert c.to_dict()["output_metric_format"] == "influxdb_line"


def test_has_subscriptions():
    c = DummyCheck(name="x", command="y", proxy_entity_name="z")
    assert c.to_dict()["subscriptions"] == ["interfacecounters"]


def test_check_compare():
    a = DummyCheck(name="x", command="y", proxy_entity_name="z")
    b = DummyCheck(name="x", command="y", proxy_entity_name="z")
    assert checks_match(a.to_dict(), b.to_dict())


def test_check_compare_with_none_handlers_and_subscriptions():
    a = DummyCheck(name="x", command="y", proxy_entity_name="z").to_dict()
    b = DummyCheck(name="x", command="y", proxy_entity_name="z").to_dict()
    a["subscriptions"] = b["subscriptions"] = None
    a["output_metric_handlers"] = b["output_metric_handlers"] = None
    assert checks_match(a, b)


def test_checks_differ():
    a = DummyCheck(name="x", command="x", proxy_entity_name="1")
    b = DummyCheck(name="x", command="x", proxy_entity_name="2")
    assert not checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name="x", command="1", proxy_entity_name="x")
    b = DummyCheck(name="x", command="2", proxy_entity_name="x")
    assert not checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name="1", command="x", proxy_entity_name="x")
    b = DummyCheck(name="2", command="x", proxy_entity_name="x")
    assert not checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name="x", command="x", proxy_entity_name="x")
    b = DummyCheck(name="x", command="x", proxy_entity_name="x")
    a.publish = not a.publish
    assert not checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name="x", command="x", proxy_entity_name="x")
    b = DummyCheck(name="x", command="x", proxy_entity_name="x")
    a.INTERVAL_S += 1
    assert not checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name="x", command="x", proxy_entity_name="x")
    b = DummyCheck(name="x", command="x", proxy_entity_name="x")
    a.round_robin = not a.round_robin
    assert not checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name="x", command="x", proxy_entity_name="x")
    b = DummyCheck(name="x", command="x", proxy_entity_name="x")
    a.METRIC_FORMAT = a.METRIC_FORMAT + "x"
    assert not checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name="x", command="x", proxy_entity_name="x")
    b = DummyCheck(name="x", command="x", proxy_entity_name="x")
    a.SUBSCRIPTIONS = ["x"]
    assert not checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name="x", command="x", proxy_entity_name="x")
    b = DummyCheck(name="x", command="x", proxy_entity_name="x")
    a.METRIC_HANDLERS = ["a"]
    assert not checks_match(a.to_dict(), b.to_dict())

    a = DummyCheck(name="x", command="x", proxy_entity_name="x")
    b = DummyCheck(name="x", command="x", proxy_entity_name="x")
    a.NAMESPACE = a.NAMESPACE + "x"
    assert not checks_match(a.to_dict(), b.to_dict())