tests: add topotest for LDP SNMP

[mirror_frr.git] / tests / topotests / ldp-snmp / test_ldp_snmp_topo1.py

#!/usr/bin/env python

#
# test_ldp_isis_topo1.py
# Part of NetDEF Topology Tests
#
# Copyright (c) 2020 by Volta Networks
#
# Permission to use, copy, modify, and/or distribute this software
# for any purpose with or without fee is hereby granted, provided
# that the above copyright notice and this permission notice appear
# in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
# DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
# WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
# OF THIS SOFTWARE.
#

"""
test_ldp_vpls_topo1.py:

                   +---------+                +---------+
                   |         |                |         |
                   |   CE1   |                |   CE2   |
                   |         |                |         |
                   +---------+                +---------+
ce1-eth0 (172.16.1.1/24)|                          |ce2-eth0 (172.16.1.2/24)
                        |                          |
                        |                          |
                rt1-eth0|                          |rt2-eth0
                   +---------+  10.0.1.0/24   +---------+
                   |         |rt1-eth1        |         |
                   |   RT1   +----------------+   RT2   |
                   | 1.1.1.1 |        rt2-eth1| 2.2.2.2 |
                   |         |                |         |
                   +---------+                +---------+
                rt1-eth2|                          |rt2-eth2
                        |                          |
                        |                          |
             10.0.2.0/24|        +---------+       |10.0.3.0/24
                        |        |         |       |
                        |        |   RT3   |       |
                        +--------+ 3.3.3.3 +-------+
                         rt3-eth2|         |rt3-eth1
                                 +---------+
                                      |rt3-eth0
                                      |
                                      |
              ce3-eth0 (172.16.1.3/24)|
                                 +---------+
                                 |         |
                                 |   CE3   |
                                 |         |
                                 +---------+
"""

import os
import re
import sys
import pytest
import json
from time import sleep
from functools import partial

# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))

# pylint: disable=C0413
# Import topogen and topotest helpers
from lib import topotest
from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger
from lib.snmptest import SnmpTester

# Required to instantiate the topology builder class.
from mininet.topo import Topo


class TemplateTopo(Topo):
    "Test topology builder"

    def build(self, *_args, **_opts):
        "Build function"
        tgen = get_topogen(self)

        #
        # Define FRR Routers
        #
        for router in ["ce1", "ce2", "ce3", "r1", "r2", "r3"]:
            tgen.add_router(router)

        #
        # Define connections
        #
        switch = tgen.add_switch("s1")
        switch.add_link(tgen.gears["ce1"])
        switch.add_link(tgen.gears["r1"])

        switch = tgen.add_switch("s2")
        switch.add_link(tgen.gears["ce2"])
        switch.add_link(tgen.gears["r2"])

        switch = tgen.add_switch("s3")
        switch.add_link(tgen.gears["ce3"])
        switch.add_link(tgen.gears["r3"])

        switch = tgen.add_switch("s4")
        switch.add_link(tgen.gears["r1"])
        switch.add_link(tgen.gears["r2"])

        switch = tgen.add_switch("s5")
        switch.add_link(tgen.gears["r1"])
        switch.add_link(tgen.gears["r3"])

        switch = tgen.add_switch("s6")
        switch.add_link(tgen.gears["r2"])
        switch.add_link(tgen.gears["r3"])


def setup_module(mod):
    "Sets up the pytest environment"
    tgen = Topogen(TemplateTopo, mod.__name__)
    tgen.start_topology()

    router_list = tgen.routers()

    # For all registered routers, load the zebra configuration file
    for rname, router in router_list.items():
        router.load_config(
            TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
        )
        # Don't start isisd and ldpd in the CE nodes
        if router.name[0] == "r":
            router.load_config(
                TopoRouter.RD_ISIS, os.path.join(CWD, "{}/isisd.conf".format(rname))
            )
            router.load_config(
                TopoRouter.RD_LDP, os.path.join(CWD, "{}/ldpd.conf".format(rname)),
                "-M snmp"
            )
            router.load_config(
                TopoRouter.RD_SNMP, os.path.join(CWD, "{}/snmpd.conf".format(rname)),
                "-Le -Ivacm_conf,usmConf,iquery -V -DAgentX,trap"
            )


    tgen.start_router()


def teardown_module(mod):
    "Teardown the pytest environment"
    tgen = get_topogen()

    # This function tears down the whole topology.
    tgen.stop_topology()


def router_compare_json_output(rname, command, reference):
    "Compare router JSON output"

    logger.info('Comparing router "%s" "%s" output', rname, command)

    tgen = get_topogen()
    filename = "{}/{}/{}".format(CWD, rname, reference)
    expected = json.loads(open(filename).read())

    # Run test function until we get an result.
    test_func = partial(topotest.router_json_cmp, tgen.gears[rname], command, expected)
    _, diff = topotest.run_and_expect(test_func, None, count=320, wait=0.5)
    assertmsg = '"{}" JSON output mismatches the expected result'.format(rname)
    assert diff is None, assertmsg


def test_isis_convergence():
    logger.info("Test: check ISIS adjacencies")
    tgen = get_topogen()

    # Skip if previous fatal error condition is raised
    if tgen.routers_have_failure():
        pytest.skip(tgen.errors)

    for rname in ["r1", "r2", "r3"]:
        router_compare_json_output(
            rname,
            "show yang operational-data /frr-interface:lib isisd",
            "show_yang_interface_isis_adjacencies.ref",
        )


def test_rib():
    logger.info("Test: verify RIB")
    tgen = get_topogen()

    # Skip if previous fatal error condition is raised
    # TODO: disabling this check to avoid 'snmpd not running' errors
    #if tgen.routers_have_failure():
    #    pytest.skip(tgen.errors)

    for rname in ["r1", "r2", "r3"]:
        router_compare_json_output(rname, "show ip route json", "show_ip_route.ref")


def test_ldp_adjacencies():
    logger.info("Test: verify LDP adjacencies")
    tgen = get_topogen()

    # Skip if previous fatal error condition is raised
    # TODO: disabling this check to avoid 'snmpd not running' errors
    #if tgen.routers_have_failure():
    #    pytest.skip(tgen.errors)

    for rname in ["r1", "r2", "r3"]:
        router_compare_json_output(
            rname, "show mpls ldp discovery json", "show_ldp_discovery.ref"
        )


def test_ldp_neighbors():
    logger.info("Test: verify LDP neighbors")
    tgen = get_topogen()

    # Skip if previous fatal error condition is raised
    #if tgen.routers_have_failure():
    #    pytest.skip(tgen.errors)

    for rname in ["r1", "r2", "r3"]:
        router_compare_json_output(
            rname, "show mpls ldp neighbor json", "show_ldp_neighbor.ref"
        )


def test_r1_ldp_lsr_objects():
    "Test mplsLdpLsrObjects objects"
    tgen = get_topogen()

    r1 = tgen.net.get("r1")
    r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")

    assert r1_snmp.test_oid('mplsLdpLsrId', "01 01 01 01")
    assert r1_snmp.test_oid('mplsLdpLsrLoopDetectionCapable', 'none(1)')


def test_r1_ldp_entity_table():
    "Test mplsLdpEntityTable"
    tgen = get_topogen()

    r1 = tgen.net.get("r1")
    r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")

    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityLdpId', ['1.1.1.1:0'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityIndex', ['1'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityProtocolVersion', ['1'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityAdminStatus', ['enable(1)'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityOperStatus', ['enabled(2)'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityTcpPort', ['646'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityUdpDscPort', ['646'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityMaxPduLength', ['4096 octets'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityKeepAliveHoldTimer', ['180 seconds'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityHelloHoldTimer', ['0 seconds'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityInitSessionThreshold', ['0'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityLabelDistMethod', ['downstreamUnsolicited(2)'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityLabelRetentionMode', ['liberal(2)'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityPathVectorLimit', ['0'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityHopCountLimit', ['0'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityTransportAddrKind', ['loopback(2)'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityTargetPeer', ['true(1)'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityTargetPeerAddrType', ['ipv4(1)'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityTargetPeerAddr', ['01 01 01 01'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityLabelType', ['generic(1)'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityDiscontinuityTime', ['(0) 0:00:00.00'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityStorageType', ['nonVolatile(3)'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpEntityRowStatus', ['createAndGo(4)'])


def test_r1_ldp_peer_table():
    "Test mplsLdpPeerTable"
    tgen = get_topogen()

    r1 = tgen.net.get("r1")
    r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")

    assert r1_snmp.test_oid_walk(
        'mplsLdpPeerLdpId', ['2.2.2.2:0', '3.3.3.3:0'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpPeerLabelDistMethod',
        ['downstreamUnsolicited(2)', 'downstreamUnsolicited(2)'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpPeerPathVectorLimit', ['0', '0'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpPeerTransportAddrType', ['ipv4(1)', 'ipv4(1)'])
    assert r1_snmp.test_oid_walk(
        'mplsLdpPeerTransportAddr', ['02 02 02 02', '03 03 03 03'])


def test_r1_ldp_session_table():
    "Test mplsLdpSessionTable"
    tgen = get_topogen()

    r1 = tgen.net.get("r1")
    r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")

    assert r1_snmp.test_oid_walk('mplsLdpSessionState',
        ['operational(5)', 'operational(5)'])
    assert r1_snmp.test_oid_walk('mplsLdpSessionRole',
        ['passive(3)', 'passive(3)'])
    assert r1_snmp.test_oid_walk('mplsLdpSessionProtocolVersion',
        ['1', '1'])
    assert r1_snmp.test_oid_walk('mplsLdpSessionKeepAliveTime',
        ['180 seconds', '180 seconds'])
    assert r1_snmp.test_oid_walk('mplsLdpSessionMaxPduLength',
        ['4096 octets', '4096 octets'])
    assert r1_snmp.test_oid_walk('mplsLdpSessionDiscontinuityTime',
        ['(0) 0:00:00.00', '(0) 0:00:00.00'])


def test_r1_ldp_hello_adjacency_table():
    "Test mplsLdpHelloAdjacencyTable"
    tgen = get_topogen()

    r1 = tgen.net.get("r1")
    r1_snmp = SnmpTester(r1, "1.1.1.1", "public", "2c")

    assert r1_snmp.test_oid_walk('mplsLdpHelloAdjacencyIndex',
        ['1', '2', '1'])
    assert r1_snmp.test_oid_walk('mplsLdpHelloAdjacencyHoldTime',
        ['15', '45', '15'])
    assert r1_snmp.test_oid_walk('mplsLdpHelloAdjacencyType',
        ['link(1)', 'targeted(2)', 'link(1)'])


# Memory leak test template
# disabling memory leak
def test_memory_leak():
    "Run the memory leak test and report results."
    tgen = get_topogen()
    if not tgen.is_memleak_enabled():
        pytest.skip("Memory leak test/report is disabled")

    tgen.report_memory_leaks()


if __name__ == "__main__":
    args = ["-s"] + sys.argv[1:]
    sys.exit(pytest.main(args))