test/ext/conftest.py

   1
   2 from subprocess import CalledProcessError
   3 import pytest, subprocess, sys
   4
   5 from bup.compat import fsdecode
   6 from bup.io import byte_stream
   7
   8 # Handle all test-* files as wvtest protocol subprocesses
   9 # cf. https://docs.pytest.org/en/latest/example/nonpython.html
  10
  11 class BupSubprocFailure(Exception):
  12     def __init__(self, msg, cmd, status, failures):
  13         super(BupSubprocFailure, self).__init__(msg)
  14         self.cmd = cmd
  15         self.status = status
  16         self.failures = failures
  17
  18 class BupSubprocTestRunner(pytest.Item):
  19
  20     def __init__(self, name, parent):
  21         super(BupSubprocTestRunner, self).__init__(name, parent)
  22
  23     def runtest(self):
  24         cmd = str(self.fspath)
  25         p = subprocess.Popen(cmd,
  26                              stdout=subprocess.PIPE,
  27                              stderr=subprocess.STDOUT)
  28         out = p.communicate()[0]
  29         sys.stdout.flush()
  30         byte_stream(sys.stdout).write(out)
  31         lines = out.splitlines()
  32         for line in lines:
  33             if line.startswith(b'!') and line.lower().endswith(b' skip ok'):
  34                 pytest.skip(line.decode('ascii'))
  35                 return
  36         failures = [line for line in lines
  37                     if (line.startswith(b'!')
  38                         and line.lower().endswith(b' failed'))]
  39         if b'AssertionError' in out:
  40             raise BupSubprocFailure('AssertionError detected')
  41         if failures or p.returncode != 0:
  42             raise BupSubprocFailure('%s failed (exit %d, %d failures)'
  43                                     % (cmd, p.returncode, len(failures)),
  44                                     cmd, p.returncode, failures)
  45
  46     def repr_failure(self, excinfo):
  47         ex = excinfo.value
  48         if isinstance(ex, BupSubprocFailure):
  49             msg = ['Exit status: %d' % ex.status,
  50                    'Failures:']
  51             msg.extend(fsdecode(s) for s in ex.failures)
  52             return '\n'.join(msg)
  53
  54     def reportinfo(self):
  55         # This does not appear to be documented, but is in the
  56         # example, and sets the final report header line (at least)
  57         # for failures.
  58         test_name = str(self.fspath)
  59         linenum = None
  60         return self.fspath, linenum, test_name
  61
  62 class BupSubprocTestFile(pytest.File):
  63     def collect(self):
  64         name = self.fspath.basename
  65         # name='' because there's only one test: running the command.
  66         # i.e there are no sub-tests.  Otherwise the status messages
  67         # duplicate the test name like this:
  68         #   test/ext/test-cat-file.sh::test-cat-file.sh PASSED ...
  69         try:
  70             yield BupSubprocTestRunner.from_parent(self, name='')
  71         except AttributeError:
  72             yield BupSubprocTestRunner('', self)
  73
  74 def pytest_collect_file(parent, path):
  75     base = path.basename
  76     if base.startswith('test-') and not base.endswith('~'):
  77         try:
  78             item = BupSubprocTestFile.from_parent(parent, fspath=path)
  79         except AttributeError:
  80             item = BupSubprocTestFile(path, parent)
  81         if base == 'test-release-archive':
  82             item.add_marker(pytest.mark.release)
  83         return item