• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2#
3#   Copyright 2020 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17from datetime import timedelta
18
19from mobly.asserts import assert_true
20from mobly.asserts import assert_false
21from mobly import signals
22
23from blueberry.tests.gd.cert.event_stream import IEventStream
24from blueberry.tests.gd.cert.event_stream import NOT_FOR_YOU_assert_event_occurs
25from blueberry.tests.gd.cert.event_stream import NOT_FOR_YOU_assert_all_events_occur
26from blueberry.tests.gd.cert.event_stream import NOT_FOR_YOU_assert_none_matching
27from blueberry.tests.gd.cert.event_stream import NOT_FOR_YOU_assert_none
28
29
30class ObjectSubject(object):
31
32    def __init__(self, value):
33        self._value = value
34
35    def isEqualTo(self, other):
36        if self._value != other:
37            raise signals.TestFailure("Expected \"%s\" to be equal to \"%s\"" % (self._value, other), extras=None)
38
39    def isNotEqualTo(self, other):
40        if self._value == other:
41            raise signals.TestFailure("Expected \"%s\" not to be equal to \"%s\"" % (self._value, other), extras=None)
42
43    def isGreaterThan(self, other):
44        if self._value <= other:
45            raise signals.TestFailure("Expected \"%s\" to be greater than \"%s\"" % (self._value, other), extras=None)
46
47    def isLessThan(self, other):
48        if self._value >= other:
49            raise signals.TestFailure("Expected \"%s\" to be less than \"%s\"" % (self._value, other), extras=None)
50
51    def isNone(self):
52        if self._value is not None:
53            raise signals.TestFailure("Expected \"%s\" to be None" % self._value, extras=None)
54
55    def isNotNone(self):
56        if self._value is None:
57            raise signals.TestFailure("Expected \"%s\" to not be None" % self._value, extras=None)
58
59
60DEFAULT_TIMEOUT = timedelta(seconds=30)
61
62
63class EventStreamSubject(ObjectSubject):
64
65    def __init__(self, value):
66        super().__init__(value)
67
68    def emits(self, *match_fns, at_least_times=1, timeout=DEFAULT_TIMEOUT):
69        if len(match_fns) == 0:
70            raise signals.TestFailure("Must specify a match function")
71        elif len(match_fns) == 1:
72            NOT_FOR_YOU_assert_event_occurs(self._value, match_fns[0], at_least_times=at_least_times, timeout=timeout)
73            return EventStreamContinuationSubject(self._value)
74        else:
75            return MultiMatchStreamSubject(self._value, match_fns, timeout)
76
77    def emitsNone(self, *match_fns, timeout):
78        if len(match_fns) == 0:
79            NOT_FOR_YOU_assert_none(self._value, timeout=timeout)
80            return EventStreamContinuationSubject(self._value)
81        elif len(match_fns) == 1:
82            NOT_FOR_YOU_assert_none_matching(self._value, match_fns[0], timeout=timeout)
83            return EventStreamContinuationSubject(self._value)
84        else:
85            raise signals.TestFailure("Cannot specify multiple match functions")
86
87
88class MultiMatchStreamSubject(object):
89
90    def __init__(self, stream, match_fns, timeout):
91        self._stream = stream
92        self._match_fns = match_fns
93        self._timeout = timeout
94
95    def inAnyOrder(self):
96        NOT_FOR_YOU_assert_all_events_occur(self._stream, self._match_fns, order_matters=False, timeout=self._timeout)
97        return EventStreamContinuationSubject(self._stream)
98
99    def inOrder(self):
100        NOT_FOR_YOU_assert_all_events_occur(self._stream, self._match_fns, order_matters=True, timeout=self._timeout)
101        return EventStreamContinuationSubject(self._stream)
102
103
104class EventStreamContinuationSubject(ObjectSubject):
105
106    def __init__(self, value):
107        super().__init__(value)
108
109    def then(self, *match_fns, at_least_times=1, timeout=DEFAULT_TIMEOUT):
110        if len(match_fns) == 0:
111            raise signals.TestFailure("Must specify a match function")
112        elif len(match_fns) == 1:
113            NOT_FOR_YOU_assert_event_occurs(self._value, match_fns[0], at_least_times=at_least_times, timeout=timeout)
114            return EventStreamContinuationSubject(self._value)
115        else:
116            return MultiMatchStreamSubject(self._value, match_fns, timeout)
117
118    def thenNone(self, *match_fns, timeout):
119        if len(match_fns) == 0:
120            NOT_FOR_YOU_assert_none(self._value, timeout=timeout)
121            return EventStreamContinuationSubject(self._value)
122        elif len(match_fns) == 1:
123            NOT_FOR_YOU_assert_none_matching(self._value, match_fns[0], timeout=timeout)
124            return EventStreamContinuationSubject(self._value)
125        else:
126            raise signals.TestFailure("Cannot specify multiple match functions")
127
128
129class BooleanSubject(ObjectSubject):
130
131    def __init__(self, value):
132        super().__init__(value)
133
134    def isTrue(self):
135        assert_true(self._value, "")
136
137    def isFalse(self):
138        assert_false(self._value, "")
139
140
141class TimeDeltaSubject(ObjectSubject):
142
143    def __init__(self, value):
144        super().__init__(value)
145
146    def isWithin(self, time_bound):
147        assert_true(self._value < time_bound, "")
148
149
150def assertThat(subject):
151    if type(subject) is bool:
152        return BooleanSubject(subject)
153    elif isinstance(subject, IEventStream):
154        return EventStreamSubject(subject)
155    elif isinstance(subject, timedelta):
156        return TimeDeltaSubject(subject)
157    else:
158        return ObjectSubject(subject)
159