tests/helpers/python/frrtest.py

   1 # SPDX-License-Identifier: GPL-2.0-or-later
   2 #
   3 # Test helpers for FRR
   4 #
   5 # Copyright (C) 2017 by David Lamparter & Christian Franke,
   6 #                       Open Source Routing / NetDEF Inc.
   7 #
   8 # This file is part of FRRouting (FRR)
   9 #
  10
  11 import subprocess
  12 import sys
  13 import re
  14 import inspect
  15 import os
  16 import difflib
  17
  18 import frrsix
  19
  20 #
  21 # These are the gritty internals of the TestMultiOut implementation.
  22 # See below for the definition of actual TestMultiOut tests.
  23 #
  24
  25 srcbase = os.path.abspath(inspect.getsourcefile(frrsix))
  26 for i in range(0, 3):
  27     srcbase = os.path.dirname(srcbase)
  28
  29
  30 def binpath(srcpath):
  31     return os.path.relpath(os.path.abspath(srcpath), srcbase)
  32
  33
  34 class MultiTestFailure(Exception):
  35     pass
  36
  37
  38 class MetaTestMultiOut(type):
  39     def __getattr__(cls, name):
  40         if name.startswith("_"):
  41             raise AttributeError
  42
  43         internal_name = "_{}".format(name)
  44         if internal_name not in dir(cls):
  45             raise AttributeError
  46
  47         def registrar(*args, **kwargs):
  48             cls._add_test(getattr(cls, internal_name), *args, **kwargs)
  49
  50         return registrar
  51
  52
  53 @frrsix.add_metaclass(MetaTestMultiOut)
  54 class _TestMultiOut(object):
  55     def _run_tests(self):
  56         if "tests_run" in dir(self.__class__) and self.tests_run:
  57             return
  58         self.__class__.tests_run = True
  59         basedir = os.path.dirname(inspect.getsourcefile(type(self)))
  60         program = os.path.join(basedir, self.program)
  61         proc = subprocess.Popen([binpath(program)], stdout=subprocess.PIPE)
  62         self.output, _ = proc.communicate("")
  63         self.exitcode = proc.wait()
  64
  65         self.__class__.testresults = {}
  66         for test in self.tests:
  67             try:
  68                 test(self)
  69             except MultiTestFailure:
  70                 self.testresults[test] = sys.exc_info()
  71             else:
  72                 self.testresults[test] = None
  73
  74     def _exit_cleanly(self):
  75         if self.exitcode != 0:
  76             raise MultiTestFailure("Program did not terminate with exit code 0")
  77
  78     @classmethod
  79     def _add_test(cls, method, *args, **kwargs):
  80         if "tests" not in dir(cls):
  81             setattr(cls, "tests", [])
  82             if method is not cls._exit_cleanly:
  83                 cls._add_test(cls._exit_cleanly)
  84
  85         def matchfunction(self):
  86             method(self, *args, **kwargs)
  87
  88         cls.tests.append(matchfunction)
  89
  90         def testfunction(self):
  91             self._run_tests()
  92             result = self.testresults[matchfunction]
  93             if result is not None:
  94                 frrsix.reraise(*result)
  95
  96         testname = re.sub(r"[^A-Za-z0-9]", "_", "%r%r" % (args, kwargs))
  97         testname = re.sub(r"__*", "_", testname)
  98         testname = testname.strip("_")
  99         if not testname:
 100             testname = method.__name__.strip("_")
 101         if "test_%s" % testname in dir(cls):
 102             index = 2
 103             while "test_%s_%d" % (testname, index) in dir(cls):
 104                 index += 1
 105             testname = "%s_%d" % (testname, index)
 106         setattr(cls, "test_%s" % testname, testfunction)
 107
 108
 109 #
 110 # This class houses the actual TestMultiOut tests types.
 111 # If you want to add a new test type, you probably do it here.
 112 #
 113 # Say you want to add a test type called foobarlicious. Then define
 114 # a function _foobarlicious here that takes self and the test arguments
 115 # when called. That function should check the output in self.output
 116 # to see whether it matches the expectation of foobarlicious with the
 117 # given arguments and should then adjust self.output according to how
 118 # much output it consumed.
 119 # If the output doesn't meet the expectations, MultiTestFailure can be
 120 # raised, however that should only be done after self.output has been
 121 # modified according to consumed content.
 122 #
 123
 124 re_okfail = re.compile(r"(?:[3[12]m|^)?(?P<ret>OK|failed)".encode("utf8"), re.MULTILINE)
 125
 126
 127 class TestMultiOut(_TestMultiOut):
 128     def _onesimple(self, line):
 129         if type(line) is str:
 130             line = line.encode("utf8")
 131         idx = self.output.find(line)
 132         if idx != -1:
 133             self.output = self.output[idx + len(line) :]
 134         else:
 135             raise MultiTestFailure("%r could not be found" % line)
 136
 137     def _okfail(self, line, okfail=re_okfail):
 138         self._onesimple(line)
 139
 140         m = okfail.search(self.output)
 141         if m is None:
 142             raise MultiTestFailure("OK/fail not found")
 143         self.output = self.output[m.end() :]
 144
 145         if m.group("ret") != "OK".encode("utf8"):
 146             raise MultiTestFailure("Test output indicates failure")
 147
 148
 149 #
 150 # This class implements a test comparing the output of a program against
 151 # an existing reference output
 152 #
 153
 154
 155 class TestRefMismatch(Exception):
 156     def __init__(self, _test, outtext, reftext):
 157         self.outtext = outtext
 158         self.reftext = reftext
 159
 160     def __str__(self):
 161         rv = "Expected output and actual output differ:\n"
 162         rv += "\n".join(
 163             difflib.unified_diff(
 164                 self.reftext.splitlines(),
 165                 self.outtext.splitlines(),
 166                 "outtext",
 167                 "reftext",
 168                 lineterm="",
 169             )
 170         )
 171         return rv
 172
 173
 174 class TestExitNonzero(Exception):
 175     pass
 176
 177
 178 class TestRefOut(object):
 179     def test_refout(self):
 180         basedir = os.path.dirname(inspect.getsourcefile(type(self)))
 181         program = os.path.join(basedir, self.program)
 182
 183         if getattr(self, "built_refin", False):
 184             refin = binpath(program) + ".in"
 185         else:
 186             refin = program + ".in"
 187         if getattr(self, "built_refout", False):
 188             refout = binpath(program) + ".refout"
 189         else:
 190             refout = program + ".refout"
 191
 192         intext = ""
 193         if os.path.exists(refin):
 194             with open(refin, "rb") as f:
 195                 intext = f.read()
 196         with open(refout, "rb") as f:
 197             reftext = f.read()
 198
 199         proc = subprocess.Popen(
 200             [binpath(program)], stdin=subprocess.PIPE, stdout=subprocess.PIPE
 201         )
 202         outtext, _ = proc.communicate(intext)
 203
 204         # Get rid of newline problems (Windows vs Unix Style)
 205         outtext_str = outtext.decode("utf8").replace("\r\n", "\n").replace("\r", "\n")
 206         reftext_str = reftext.decode("utf8").replace("\r\n", "\n").replace("\r", "\n")
 207
 208         if outtext_str != reftext_str:
 209             raise TestRefMismatch(self, outtext_str, reftext_str)
 210         if proc.wait() != 0:
 211             raise TestExitNonzero(self)