summaryrefslogtreecommitdiff
path: root/testrepository/results.py
blob: 47f427c5c69e70eca9846f055b1b79a18d442945 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
from subunit import test_results

from testtools import TestResult

from testrepository.utils import timedelta_to_seconds

# Subunit 0.0.8 has the time forwarding fix built-in. 0.0.8 can be detected by
# looking for _PredicateFilter.
if getattr(test_results, '_PredicateFilter', None) is None:
    class TestResultFilter(test_results.TestResultFilter):
        """Test result filter."""

        def _get_concrete_result(self):
            # XXX: This is really crappy. It assumes that the test result we
            # actually care about is decorated and that we can find our way to the
            # one we care about. We want to report counts before filtering, so we
            # should actually use two result objects - one to count and one to
            # show. Arguably also the testsRun incrementing facility should be in
            # testtools / subunit
            concrete = self
            while True:
                next = getattr(concrete, 'decorated', None)
                if next is None:
                    return concrete
                concrete = next

        def _filtered(self):
            super(TestResultFilter, self)._filtered()
            concrete = self._get_concrete_result()
            concrete.testsRun += 1

        def stopTest(self, test):
            # Filter out 'time' calls, because we want to forward those events
            # regardless of whether the test is filtered.
            #
            # XXX: Should this be pushed into subunit?
            buffered_calls = []
            for method, args, kwargs in self._buffered_calls:
                if method == 'time':
                    self.decorated.time(*args, **kwargs)
                else:
                    buffered_calls.append((method, args, kwargs))
            self._buffered_calls = buffered_calls
            super(TestResultFilter, self).stopTest(test)
else:
    TestResultFilter = test_results.TestResultFilter


class SummarizingResult(TestResult):

    def __init__(self):
        super(SummarizingResult, self).__init__()
        self._first_time = None

    def startTestRun(self):
        super(SummarizingResult, self).startTestRun()
        self._first_time = None

    def get_num_failures(self):
        return len(self.failures) + len(self.errors)

    def time(self, a_time):
        if self._first_time is None:
            self._first_time = a_time
        super(SummarizingResult, self).time(a_time)

    def get_time_taken(self):
        now = self._now()
        if None in (self._first_time, now):
            return None
        return timedelta_to_seconds(now - self._first_time)