git.samba.org - third_party/subunit/blob - python/subunit/tests/test_output_filter.py

   1 #
   2 #  subunit: extensions to python unittest to get test results from subprocesses.
   3 #  Copyright (C) 2005  Thomi Richards <thomi.richards@canonical.com>
   4 #
   5 #  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
   6 #  license at the users choice. A copy of both licenses are available in the
   7 #  project source as Apache-2.0 and BSD. You may not use this file except in
   8 #  compliance with one of these two licences.
   9 #
  10 #  Unless required by applicable law or agreed to in writing, software
  11 #  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
  12 #  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
  13 #  license you chose for the specific language governing permissions and
  14 #  limitations under that license.
  15 #
  16
  17
  18 from io import BytesIO
  19
  20 from collections import namedtuple
  21 from testtools import TestCase
  22 from testtools.matchers import (
  23     Equals,
  24     Matcher,
  25     Mismatch,
  26 )
  27 from testtools.testresult.doubles import StreamResult
  28
  29 from subunit.v2 import StreamResultToBytes, ByteStreamToStreamResult
  30 from subunit._output import (
  31     generate_bytestream,
  32     parse_arguments,
  33     translate_command_name,
  34 )
  35
  36 class OutputFilterArgumentTests(TestCase):
  37
  38     """Tests for the command line argument parser."""
  39
  40     def _test_command(self, command, test_id):
  41         args = parse_arguments(args=[command, test_id])
  42
  43         self.assertThat(args.action, Equals(command))
  44         self.assertThat(args.test_id, Equals(test_id))
  45
  46     def test_can_parse_start_test(self):
  47         self._test_command('start', self.getUniqueString())
  48
  49     def test_can_parse_pass_test(self):
  50         self._test_command('pass', self.getUniqueString())
  51
  52     def test_can_parse_fail_test(self):
  53         self._test_command('fail', self.getUniqueString())
  54
  55     def test_can_parse_skip_test(self):
  56         self._test_command('skip', self.getUniqueString())
  57
  58     def test_command_translation(self):
  59         self.assertThat(translate_command_name('start'), Equals('inprogress'))
  60         self.assertThat(translate_command_name('pass'), Equals('success'))
  61         for command in ('fail', 'skip'):
  62             self.assertThat(translate_command_name(command), Equals(command))
  63
  64
  65 class ByteStreamCompatibilityTests(TestCase):
  66
  67     def _get_result_for(self, *commands):
  68         """Get a result object from *commands.
  69
  70         Runs the 'generate_bytestream' function from subunit._output after
  71         parsing *commands as if they were specified on the command line. The
  72         resulting bytestream is then converted back into a result object and
  73         returned.
  74
  75         """
  76         stream = BytesIO()
  77
  78         for command_list in commands:
  79             args = parse_arguments(command_list)
  80             output_writer = StreamResultToBytes(output_stream=stream)
  81             generate_bytestream(args, output_writer)
  82
  83         stream.seek(0)
  84
  85         case = ByteStreamToStreamResult(source=stream)
  86         result = StreamResult()
  87         case.run(result)
  88         return result
  89
  90     def test_start_generates_inprogress(self):
  91         result = self._get_result_for(
  92             ['start', 'foo'],
  93         )
  94
  95         self.assertThat(
  96             result._events[0],
  97             MatchesCall(call='status', test_id='foo', test_status='inprogress')
  98         )
  99
 100     def test_pass_generates_success(self):
 101         result = self._get_result_for(
 102             ['pass', 'foo'],
 103         )
 104
 105         self.assertThat(
 106             result._events[0],
 107             MatchesCall(call='status', test_id='foo', test_status='success')
 108         )
 109
 110     def test_fail_generates_fail(self):
 111         result = self._get_result_for(
 112             ['fail', 'foo'],
 113         )
 114
 115         self.assertThat(
 116             result._events[0],
 117             MatchesCall(call='status', test_id='foo', test_status='fail')
 118         )
 119
 120     def test_skip_generates_skip(self):
 121         result = self._get_result_for(
 122             ['skip', 'foo'],
 123         )
 124
 125         self.assertThat(
 126             result._events[0],
 127             MatchesCall(call='status', test_id='foo', test_status='skip')
 128         )
 129
 130
 131 class MatchesCall(Matcher):
 132
 133     _position_lookup = {
 134             'call': 0,
 135             'test_id': 1,
 136             'test_status': 2,
 137             'test_tags': 3,
 138             'runnable': 4,
 139             'file_name': 5,
 140             'file_bytes': 6,
 141             'eof': 7,
 142             'mime_type': 8,
 143             'route_code': 9,
 144             'timestamp': 10,
 145         }
 146
 147     def __init__(self, **kwargs):
 148         unknown_kwargs = filter(
 149             lambda k: k not in self._position_lookup,
 150             kwargs
 151         )
 152         if unknown_kwargs:
 153             raise ValueError("Unknown keywords: %s" % ','.join(unknown_kwargs))
 154         self._filters = kwargs
 155
 156     def match(self, call_tuple):
 157         for k,v in self._filters.items():
 158             try:
 159                 if call_tuple[self._position_lookup[k]] != v:
 160                     return Mismatch("Value for key is %r, not %r" % (self._position_lookup[k], v))
 161             except IndexError:
 162                 return Mismatch("Key %s is not present." % k)
 163
 164     def __str__(self):
 165         return "<MatchesCall %r>" % self._filters
 166