tests: add pytest testrunners

Signed-off-by: Christian Franke <chris@opensourcerouting.org>

tests/.gitignore

@@ -3,6 +3,7 @@ Makefile.in
 *.o
 tags
 TAGS
+.cache
 .deps
 .nfs*
 *~
@@ -14,8 +15,11 @@ TAGS
 *.log
 *.sum
 *.xml
+*.pyc
 .arch-inventory
 .arch-ids
+__pycache__
+.dirstamp
 /bgpd/test_aspath
 /bgpd/test_capability
 /bgpd/test_ecommunity

tests/Makefile.am

@@ -1,3 +1,5 @@
+PYTHON ?= python
+
 AUTOMAKE_OPTIONS = subdir-objects
 AM_CPPFLAGS = \
 	-I.. \
@@ -113,3 +115,29 @@ bgpd_test_capability_LDADD = $(BGP_TEST_LDADD)
 bgpd_test_ecommunity_LDADD = $(BGP_TEST_LDADD)
 bgpd_test_mp_attr_LDADD = $(BGP_TEST_LDADD)
 bgpd_test_mpath_LDADD = $(BGP_TEST_LDADD)
+
+EXTRA_DIST = \
+    runtests.py \
+    bgpd/test_aspath.py \
+    bgpd/test_capability.py \
+    bgpd/test_ecommunity.py \
+    bgpd/test_mp_attr.py \
+    bgpd/test_mpath.py \
+    helpers/python/frrsix.py \
+    helpers/python/frrtest.py \
+    lib/cli/test_commands.in \
+    lib/cli/test_commands.py \
+    lib/cli/test_commands.refout \
+    lib/cli/test_cli.in \
+    lib/cli/test_cli.py \
+    lib/cli/test_cli.refout \
+    lib/test_nexthop_iter.py \
+    lib/test_stream.py \
+    lib/test_stream.refout \
+    lib/test_table.py \
+    lib/test_timer_correctness.py
+
+.PHONY: tests.xml
+tests.xml: $(check_PROGRAMS)
+	$(PYTHON) runtests.py --junitxml=$@ -v
+check: tests.xml

tests/bgpd/test_aspath.py

@@ -0,0 +1,79 @@
+import frrtest
+import re
+
+re_okfail = re.compile(r'^(?:\x1b\[3[12]m)?(?P<ret>OK|failed)'.encode('utf8'),
+                       re.MULTILINE)
+
+class TestAspath(frrtest.TestMultiOut):
+    program = './test_aspath'
+
+    def _parsertest(self, line):
+        if not hasattr(self, 'parserno'):
+            self.parserno = -1
+        self.parserno += 1
+
+        self._onesimple("test %d" % self.parserno)
+        self._okfail("%s:" % line, okfail=re_okfail)
+        self._okfail("empty prepend %s:" % line, okfail=re_okfail)
+
+    def _attrtest(self, line):
+        if not hasattr(self, 'attrno'):
+            self.attrno = -1
+        self.attrno += 1
+
+        self._onesimple("aspath_attr test %d" % self.attrno)
+        self._okfail(line, okfail=re_okfail)
+
+TestAspath.parsertest("seq1")
+TestAspath.parsertest("seq2")
+TestAspath.parsertest("seq3")
+TestAspath.parsertest("seqset")
+TestAspath.parsertest("seqset2")
+TestAspath.parsertest("multi")
+TestAspath.parsertest("confed")
+TestAspath.parsertest("confed2")
+TestAspath.parsertest("confset")
+TestAspath.parsertest("confmulti")
+TestAspath.parsertest("seq4")
+TestAspath.parsertest("tripleseq1")
+TestAspath.parsertest("someprivate")
+TestAspath.parsertest("allprivate")
+TestAspath.parsertest("long")
+TestAspath.parsertest("seq1extra")
+TestAspath.parsertest("empty")
+TestAspath.parsertest("redundantset")
+TestAspath.parsertest("reconcile_lead_asp")
+TestAspath.parsertest("reconcile_new_asp")
+TestAspath.parsertest("reconcile_confed")
+TestAspath.parsertest("reconcile_start_trans")
+TestAspath.parsertest("reconcile_start_trans4")
+TestAspath.parsertest("reconcile_start_trans_error")
+TestAspath.parsertest("redundantset2")
+TestAspath.parsertest("zero-size overflow")
+TestAspath.parsertest("zero-size overflow + valid segment")
+TestAspath.parsertest("invalid segment type")
+
+for i in range(10):
+    TestAspath.okfail("prepend test %d" % i)
+for i in range(5):
+    TestAspath.okfail("aggregate test %d" % i)
+for i in range(5):
+    TestAspath.okfail("reconcile test %d" % i)
+for _ in range(22):
+    TestAspath.okfail("left cmp ")
+
+TestAspath.okfail("empty_get_test")
+
+TestAspath.attrtest("basic test")
+TestAspath.attrtest("length too short")
+TestAspath.attrtest("length too long")
+TestAspath.attrtest("incorrect flag")
+TestAspath.attrtest("as4_path, with as2 format data")
+TestAspath.attrtest("as4, with incorrect attr length")
+TestAspath.attrtest("basic 4-byte as-path")
+TestAspath.attrtest("4b AS_PATH: too short")
+TestAspath.attrtest("4b AS_PATH: too long")
+TestAspath.attrtest("4b AS_PATH: too long2")
+TestAspath.attrtest("4b AS_PATH: bad flags")
+TestAspath.attrtest("4b AS4_PATH w/o AS_PATH")
+TestAspath.attrtest("4b AS4_PATH: confed")

tests/bgpd/test_capability.py

@@ -0,0 +1,47 @@
+import frrtest
+
+class TestCapability(frrtest.TestMultiOut):
+    program = './test_capability'
+
+TestCapability.okfail("MP4: MP IP/Uni")
+TestCapability.okfail("MPv6: MP IPv6/Uni")
+TestCapability.okfail("MP2: MP IP/Multicast")
+TestCapability.okfail("MP3: MP IP6/MPLS-labeled VPN")
+TestCapability.okfail("MP5: MP IP6/MPLS-VPN")
+TestCapability.okfail("MP6: MP IP4/MPLS-laveled VPN")
+TestCapability.okfail("MP8: MP unknown AFI/SAFI")
+TestCapability.okfail("MP-short: MP IP4/Unicast, length too short (< minimum)")
+TestCapability.okfail("MP-overflow: MP IP4/Unicast, length too long")
+TestCapability.okfail("caphdr: capability header, and no more")
+TestCapability.okfail("nodata: header, no data but length says there is")
+TestCapability.okfail("padded: valid, with padding")
+TestCapability.okfail("minsize: violates minsize requirement")
+TestCapability.okfail("ORF: ORF, simple, single entry, single tuple")
+TestCapability.okfail("ORF-many: ORF, multi entry/tuple")
+TestCapability.okfail("ORFlo: ORF, multi entry/tuple, hdr length too short")
+TestCapability.okfail("ORFlu: ORF, multi entry/tuple, length too long")
+TestCapability.okfail("ORFnu: ORF, multi entry/tuple, entry number too long")
+TestCapability.okfail("ORFno: ORF, multi entry/tuple, entry number too short")
+TestCapability.okfail("ORFpad: ORF, multi entry/tuple, padded to align")
+TestCapability.okfail("AS4: AS4 capability")
+TestCapability.okfail("GR: GR capability")
+TestCapability.okfail("GR-short: GR capability, but header length too short")
+TestCapability.okfail("GR-long: GR capability, but header length too long")
+TestCapability.okfail("GR-trunc: GR capability, but truncated")
+TestCapability.okfail("GR-empty: GR capability, but empty.")
+TestCapability.okfail("MP-empty: MP capability, but empty.")
+TestCapability.okfail("ORF-empty: ORF capability, but empty.")
+TestCapability.okfail("AS4-empty: AS4 capability, but empty.")
+TestCapability.okfail("dyn-empty: Dynamic capability, but empty.")
+TestCapability.okfail("dyn-old: Dynamic capability (deprecated version)")
+TestCapability.okfail("Cap-singlets: One capability per Optional-Param")
+TestCapability.okfail("Cap-series: Series of capability, one Optional-Param")
+TestCapability.okfail("AS4more: AS4 capability after other caps (singlets)")
+TestCapability.okfail("AS4series: AS4 capability, in series of capabilities")
+TestCapability.okfail("AS4real: AS4 capability, in series of capabilities")
+TestCapability.okfail("AS4real2: AS4 capability, in series of capabilities")
+TestCapability.okfail("DynCap: Dynamic Capability Message, IP/Multicast")
+TestCapability.okfail("DynCapLong: Dynamic Capability Message, IP/Multicast, truncated")
+TestCapability.okfail("DynCapPadded: Dynamic Capability Message, IP/Multicast, padded")
+TestCapability.okfail("DynCapMPCpadded: Dynamic Capability Message, IP/Multicast, cap data padded")
+TestCapability.okfail("DynCapMPCoverflow: Dynamic Capability Message, IP/Multicast, cap data != length")

tests/bgpd/test_ecommunity.py

@@ -0,0 +1,9 @@
+import frrtest
+
+class TestEcommunity(frrtest.TestMultiOut):
+    program = './test_ecommunity'
+
+TestEcommunity.okfail('ipaddr')
+TestEcommunity.okfail('ipaddr-so')
+TestEcommunity.okfail('asn')
+TestEcommunity.okfail('asn4')

tests/bgpd/test_mp_attr.py

@@ -0,0 +1,33 @@
+import frrtest
+
+class TestMpAttr(frrtest.TestMultiOut):
+    program = './test_mp_attr'
+
+TestMpAttr.okfail("IPv6: IPV6 MP Reach, global nexthop, 1 NLRI")
+TestMpAttr.okfail("IPv6-2: IPV6 MP Reach, global nexthop, 2 NLRIs")
+TestMpAttr.okfail("IPv6-default: IPV6 MP Reach, global nexthop, 2 NLRIs + default")
+TestMpAttr.okfail("IPv6-lnh: IPV6 MP Reach, global+local nexthops, 2 NLRIs + default")
+TestMpAttr.okfail("IPv6-nhlen: IPV6 MP Reach, inappropriate nexthop length")
+TestMpAttr.okfail("IPv6-nhlen2: IPV6 MP Reach, invalid nexthop length")
+TestMpAttr.okfail("IPv6-nhlen3: IPV6 MP Reach, nexthop length overflow")
+TestMpAttr.okfail("IPv6-nhlen4: IPV6 MP Reach, nexthop length short")
+TestMpAttr.okfail("IPv6-nlri: IPV6 MP Reach, NLRI bitlen overflow")
+TestMpAttr.okfail("IPv4: IPv4 MP Reach, 2 NLRIs + default")
+TestMpAttr.okfail("IPv4-nhlen: IPv4 MP Reach, nexthop lenth overflow")
+TestMpAttr.okfail("IPv4-nlrilen: IPv4 MP Reach, nlri lenth overflow")
+TestMpAttr.okfail("IPv4-VPNv4: IPv4/VPNv4 MP Reach, RD, Nexthop, 2 NLRIs")
+TestMpAttr.okfail("IPv4-VPNv4-bogus-plen: IPv4/MPLS-labeled VPN MP Reach, RD, Nexthop, NLRI / bogus p'len")
+TestMpAttr.okfail("IPv4-VPNv4-plen1-short: IPv4/VPNv4 MP Reach, RD, Nexthop, 2 NLRIs, 1st plen short")
+TestMpAttr.okfail("IPv4-VPNv4-plen1-long: IPv4/VPNv4 MP Reach, RD, Nexthop, 2 NLRIs, 1st plen long")
+TestMpAttr.okfail("IPv4-VPNv4-plenn-long: IPv4/VPNv4 MP Reach, RD, Nexthop, 3 NLRIs, last plen long")
+TestMpAttr.okfail("IPv4-VPNv4-plenn-short: IPv4/VPNv4 MP Reach, RD, Nexthop, 2 NLRIs, last plen short")
+TestMpAttr.okfail("IPv4-VPNv4-bogus-rd-type: IPv4/VPNv4 MP Reach, RD, NH, 2 NLRI, unknown RD in 1st (log, but parse)")
+TestMpAttr.okfail("IPv4-VPNv4-0-nlri: IPv4/VPNv4 MP Reach, RD, Nexthop, 3 NLRI, 3rd 0 bogus")
+TestMpAttr.okfail("IPv6-bug: IPv6, global nexthop, 1 default NLRI")
+TestMpAttr.okfail("IPv6-unreach: IPV6 MP Unreach, 1 NLRI")
+TestMpAttr.okfail("IPv6-unreach2: IPV6 MP Unreach, 2 NLRIs")
+TestMpAttr.okfail("IPv6-unreach-default: IPV6 MP Unreach, 2 NLRIs + default")
+TestMpAttr.okfail("IPv6-unreach-nlri: IPV6 MP Unreach, NLRI bitlen overflow")
+TestMpAttr.okfail("IPv4-unreach: IPv4 MP Unreach, 2 NLRIs + default")
+TestMpAttr.okfail("IPv4-unreach-nlrilen: IPv4 MP Unreach, nlri length overflow")
+TestMpAttr.okfail("IPv4-unreach-VPNv4: IPv4/MPLS-labeled VPN MP Unreach, RD, 3 NLRIs")

tests/bgpd/test_mpath.py

@@ -0,0 +1,9 @@
+import frrtest
+
+class TestMpath(frrtest.TestMultiOut):
+    program = './test_mpath'
+
+TestMpath.okfail("bgp maximum-paths config")
+TestMpath.okfail("bgp_mp_list")
+TestMpath.okfail("bgp_info_mpath_update")
+

tests/helpers/python/frrsix.py

@@ -0,0 +1,80 @@
+#
+# Copyright (c) 2010-2017 Benjamin Peterson
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+#
+
+#
+# This code is taken from the six python2 to python3 compatibility module
+#
+
+import sys
+
+PY2 = sys.version_info[0] == 2
+PY3 = sys.version_info[0] == 3
+
+def add_metaclass(metaclass):
+    """Class decorator for creating a class with a metaclass."""
+    def wrapper(cls):
+        orig_vars = cls.__dict__.copy()
+        slots = orig_vars.get('__slots__')
+        if slots is not None:
+            if isinstance(slots, str):
+                slots = [slots]
+            for slots_var in slots:
+                orig_vars.pop(slots_var)
+        orig_vars.pop('__dict__', None)
+        orig_vars.pop('__weakref__', None)
+        return metaclass(cls.__name__, cls.__bases__, orig_vars)
+    return wrapper
+
+if PY3:
+    import builtins
+    exec_ = getattr(builtins,'exec')
+
+    def reraise(tp, value, tb=None):
+        try:
+            if value is None:
+                value = tp()
+            if value.__traceback__ is not tb:
+                raise value.with_traceback(tb)
+            raise value
+        finally:
+            value = None
+            tb = None
+
+else:
+    def exec_(_code_, _globs_=None, _locs_=None):
+        """Execute code in a namespace."""
+        if _globs_ is None:
+            frame = sys._getframe(1)
+            _globs_ = frame.f_globals
+            if _locs_ is None:
+                _locs_ = frame.f_locals
+            del frame
+        elif _locs_ is None:
+            _locs_ = _globs_
+        exec("""exec _code_ in _globs_, _locs_""")
+
+    exec_("""def reraise(tp, value, tb=None):
+    try:
+        raise tp, value, tb
+    finally:
+        tb = None
+""")

tests/helpers/python/frrtest.py

@@ -0,0 +1,177 @@
+#
+# Test helpers for FRR
+#
+# Copyright (C) 2017 by David Lamparter & Christian Franke,
+#                       Open Source Routing / NetDEF Inc.
+#
+# This file is part of FreeRangeRouting (FRR)
+#
+# FRR is free software; you can redistribute it and/or modify it
+# under the terms of the GNU General Public License as published by the
+# Free Software Foundation; either version 2, or (at your option) any
+# later version.
+#
+# FRR is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with FRR; see the file COPYING.  If not, write to the Free
+# Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
+# 02111-1307, USA.
+#
+
+import subprocess
+import sys
+import re
+import inspect
+import os
+
+import frrsix
+
+#
+# These are the gritty internals of the TestMultiOut implementation.
+# See below for the definition of actual TestMultiOut tests.
+#
+
+class MultiTestFailure(Exception):
+    pass
+
+class MetaTestMultiOut(type):
+    def __getattr__(cls, name):
+        if name.startswith('_'):
+            raise AttributeError
+
+        internal_name = '_{}'.format(name)
+        if internal_name not in dir(cls):
+            raise AttributeError
+
+        def registrar(*args, **kwargs):
+            cls._add_test(getattr(cls,internal_name), *args, **kwargs)
+        return registrar
+
+@frrsix.add_metaclass(MetaTestMultiOut)
+class _TestMultiOut(object):
+    def _run_tests(self):
+        if 'tests_run' in dir(self.__class__) and self.tests_run:
+            return
+        self.__class__.tests_run = True
+        basedir = os.path.dirname(inspect.getsourcefile(type(self)))
+        program = os.path.join(basedir, self.program)
+        proc = subprocess.Popen([program], stdout=subprocess.PIPE)
+        self.output,_ = proc.communicate('')
+        self.exitcode = proc.wait()
+
+        self.__class__.testresults = {}
+        for test in self.tests:
+            try:
+                test(self)
+            except MultiTestFailure:
+                self.testresults[test] = sys.exc_info()
+            else:
+                self.testresults[test] = None
+
+    def _exit_cleanly(self):
+        if self.exitcode != 0:
+            raise MultiTestFailure("Program did not terminate with exit code 0")
+
+    @classmethod
+    def _add_test(cls, method, *args, **kwargs):
+        if 'tests' not in dir(cls):
+            setattr(cls,'tests',[])
+            cls._add_test(cls._exit_cleanly)
+
+        def matchfunction(self):
+            method(self, *args, **kwargs)
+        cls.tests.append(matchfunction)
+
+        def testfunction(self):
+            self._run_tests()
+            result = self.testresults[matchfunction]
+            if result is not None:
+                frrsix.reraise(*result)
+
+        testname = re.sub(r'[^A-Za-z0-9]', '_', '%r%r' % (args, kwargs))
+        testname = re.sub(r'__*', '_', testname)
+        testname = testname.strip('_')
+        if not testname:
+            testname = method.__name__.strip('_')
+        if "test_%s" % testname in dir(cls):
+            index = 2
+            while "test_%s_%d" % (testname,index) in dir(cls):
+                index += 1
+            testname = "%s_%d" % (testname, index)
+        setattr(cls,"test_%s" % testname, testfunction)
+
+#
+# This class houses the actual TestMultiOut tests types.
+# If you want to add a new test type, you probably do it here.
+#
+# Say you want to add a test type called foobarlicious. Then define
+# a function _foobarlicious here that takes self and the test arguments
+# when called. That function should check the output in self.output
+# to see whether it matches the expectation of foobarlicious with the
+# given arguments and should then adjust self.output according to how
+# much output it consumed.
+# If the output doesn't meet the expectations, MultiTestFailure can be
+# raised, however that should only be done after self.output has been
+# modified according to consumed content.
+#
+
+re_okfail = re.compile(r'(?:[3[12]m|^)?(?P<ret>OK|failed)'.encode('utf8'),
+                       re.MULTILINE)
+class TestMultiOut(_TestMultiOut):
+    def _onesimple(self, line):
+        if type(line) is str:
+            line = line.encode('utf8')
+        idx = self.output.find(line)
+        if idx != -1:
+            self.output = self.output[idx+len(line):]
+        else:
+            raise MultiTestFailure("%r could not be found" % line)
+
+    def _okfail(self, line, okfail=re_okfail):
+        self._onesimple(line)
+
+        m = okfail.search(self.output)
+        if m is None:
+            raise MultiTestFailure('OK/fail not found')
+        self.output = self.output[m.end():]
+
+        if m.group('ret') != 'OK'.encode('utf8'):
+            raise MultiTestFailure('Test output indicates failure')
+
+#
+# This class implements a test comparing the output of a program against
+# an existing reference output
+#
+
+class TestRefMismatch(Exception):
+    pass
+class TestExitNonzero(Exception):
+    pass
+
+class TestRefOut(object):
+    def test_refout(self):
+        basedir = os.path.dirname(inspect.getsourcefile(type(self)))
+        program = os.path.join(basedir, self.program)
+
+        refin = program + '.in'
+        refout = program + '.refout'
+
+        intext = ''
+        if os.path.exists(refin):
+            with open(refin, 'rb') as f:
+                intext = f.read()
+        with open(refout, 'rb') as f:
+            reftext = f.read()
+
+        proc = subprocess.Popen([program],
+                                stdin=subprocess.PIPE,
+                                stdout=subprocess.PIPE)
+        outtext,_ = proc.communicate(intext)
+        if outtext != reftext:
+            raise TestRefMismatch(self, outtext, reftext)
+        if proc.wait() != 0:
+            raise TestExitNonzero(self)

tests/lib/cli/test_cli.py

@@ -0,0 +1,4 @@
+import frrtest
+
+class TestCli(frrtest.TestRefOut):
+    program = './test_cli'

tests/lib/cli/test_commands.py

@@ -0,0 +1,8 @@
+import frrtest
+import pytest
+import os
+
+@pytest.mark.skipif('QUAGGA_TEST_COMMANDS' not in os.environ,
+                    reason='QUAGGA_TEST_COMMANDS not set')
+class TestCommands(frrtest.TestRefOut):
+    program = './test_commands'

tests/lib/test_nexthop_iter.py

@@ -0,0 +1,7 @@
+import frrtest
+
+class TestNexthopIter(frrtest.TestMultiOut):
+    program = './test_nexthop_iter'
+
+TestNexthopIter.onesimple('Simple test passed.')
+TestNexthopIter.onesimple('PRNG test passed.')

tests/lib/test_stream.py

@@ -0,0 +1,4 @@
+import frrtest
+
+class TestStream(frrtest.TestRefOut):
+    program = './test_stream'

tests/lib/test_table.py

@@ -0,0 +1,10 @@
+import frrtest
+
+class TestTable(frrtest.TestMultiOut):
+    program = './test_table'
+
+for i in range(6):
+    TestTable.onesimple('Verifying cmp')
+for i in range(11):
+    TestTable.onesimple('Verifying successor')
+TestTable.onesimple('Verified pausing')

tests/lib/test_timer_correctness.py

@@ -0,0 +1,6 @@
+import frrtest
+
+class TestTimerCorrectness(frrtest.TestMultiOut):
+    program = './test_timer_correctness'
+
+TestTimerCorrectness.onesimple('Expected output and actual output match.')

tests/runtests.py

@@ -0,0 +1,6 @@
+import pytest
+import sys
+import os
+
+sys.path.append(os.path.join(os.path.dirname(__file__), 'helpers','python'))
+raise SystemExit(pytest.main(sys.argv[1:]))