6755749014

Committed 04 Nov 2023 03:39PM UTC coverage: 97.65% (+0.4%) from 97.266%

Build # 6755749014

Build Type

Pull #10

github

Committed by

web-flow

Commit Message

Merge c8a234d82 into db5669aca

Pull Request Pull Request #10: Support async functions natively

Run Details

21 of 23 new or added lines in 1 file covered. (91.3%)

3 existing lines in 1 file now uncovered.

748 of 766 relevant lines covered (97.65%)

0.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.86

/dobles/allowance.py

import functools
import inspect
from typing import Any

import dobles.lifecycle
from dobles.call_count_accumulator import CallCountAccumulator
from dobles.exceptions import MockExpectationError, VerifyingBuiltinDoubleArgumentError
from dobles.verification import verify_arguments

_any = object()


async def _async_return_value(*args, **kwargs):
    return None


def _maybe_async(is_async: bool, return_value: Any):
    if is_async:
        return make_it_async(return_value)
    return return_value


def make_it_async(
    return_value=None,
    exception=None,
):
    async def inner(*args, **kwargs):
        if exception:
            raise exception
        return return_value

    return inner()


def verify_count_is_non_negative(func):
    @functools.wraps(func)
    def inner(self, arg):
        if arg < 0:
            raise TypeError(func.__name__ + " requires one positive integer argument")
        return func(self, arg)

    return inner


def check_func_takes_args(func):
    arg_spec = inspect.getfullargspec(func)
    return any([arg_spec.args, arg_spec.varargs, arg_spec.varkw, arg_spec.defaults])


def build_argument_repr_string(args, kwargs):
    args = [repr(x) for x in args]
    kwargs = ["{}={!r}".format(k, v) for k, v in kwargs.items()]
    return "({})".format(", ".join(args + kwargs))


class Allowance(object):
    """An individual method allowance (stub)."""

    def __init__(self, target, method_name, caller):
        """
        :param Target target: The object owning the method to stub.
        :param str method_name: The name of the method to stub.
        """

        self._target = target
        self._method_name = method_name
        self._caller = caller
        self.args = _any
        self.kwargs = _any
        self._custom_matcher = None
        self._is_satisfied = True
        self._call_counter = CallCountAccumulator()
        self._is_async = inspect.iscoroutinefunction(
            target.get_attr(method_name).object
        )

        if self._is_async:
            self._return_value = _async_return_value
        else:
            self._return_value = lambda *args, **kwargs: None

    def and_raise(self, exception, *args, **kwargs):
        """Causes the double to raise the provided exception when called.

        If provided, additional arguments (positional and keyword) passed to
        `and_raise` are used in the exception instantiation.

        :param Exception exception: The exception to raise.
        """

        def proxy_exception(*proxy_args, **proxy_kwargs):
            raise exception

        async def async_proxy_exception(*proxy_args, **proxy_kwargs):
            raise exception

        self._return_value = (
            async_proxy_exception if self._is_async else proxy_exception
        )
        return self

    def and_return(self, *return_values):
        """Set a return value for an allowance

        Causes the double to return the provided values in order.  If multiple
        values are provided, they are returned one at a time in sequence as the double is called.
        If the double is called more times than there are return values, it should continue to
        return the last value in the list.


        :param object return_values: The values the double will return when called,
        """

        if not return_values:
            raise TypeError("and_return() expected at least 1 return value")

        return_values = list(return_values)
        final_value = return_values.pop()

        self.and_return_result_of(
            lambda: return_values.pop(0) if return_values else final_value
        )
        return self

    def and_return_result_of(self, return_value):
        """Causes the double to return the result of calling the provided value.

        :param return_value: A callable that will be invoked to determine the double's return value.
        :type return_value: any callable object
        """

        if not check_func_takes_args(return_value):
            self._return_value = lambda *args, **kwargs: _maybe_async(
                self._is_async,
                return_value(),
            )
        else:
            self._return_value = _maybe_async(self._is_async, return_value)

        return self

    def is_satisfied(self):
        """Returns a boolean indicating whether or not the double has been satisfied.

        Stubs are always satisfied, but mocks are only satisfied if they've been
        called as was declared.

        :return: Whether or not the double is satisfied.
        :rtype: bool
        """
        return self._is_satisfied

    def with_args(self, *args, **kwargs):
        """Declares that the double can only be called with the provided arguments.

        :param args: Any positional arguments required for invocation.
        :param kwargs: Any keyword arguments required for invocation.
        """

        self.args = args
        self.kwargs = kwargs
        self.verify_arguments()
        return self

    def with_args_validator(self, matching_function):
        """Define a custom function for testing arguments

        :param func matching_function:  The function used to test arguments passed to the stub.
        """
        self.args = None
        self.kwargs = None
        self._custom_matcher = matching_function
        return self

    def __call__(self, *args, **kwargs):
        """A short hand syntax for with_args

        Allows callers to do:
            allow(module).foo.with_args(1, 2)
        With:
            allow(module).foo(1, 2)

        :param args: Any positional arguments required for invocation.
        :param kwargs: Any keyword arguments required for invocation.
        """
        return self.with_args(*args, **kwargs)

    def with_no_args(self):
        """Declares that the double can only be called with no arguments."""

        self.args = ()
        self.kwargs = {}
        self.verify_arguments()
        return self

    def satisfy_any_args_match(self):
        """Returns a boolean indicating whether or not the stub will accept arbitrary arguments.

        This will be true unless the user has specified otherwise using ``with_args`` or
        ``with_no_args``.

        :return: Whether or not the stub accepts arbitrary arguments.
        :rtype: bool
        """

        return self.args is _any and self.kwargs is _any

    def satisfy_exact_match(self, args, kwargs):
        """Returns a boolean indicating whether or not the stub will accept the provided arguments.

        :return: Whether or not the stub accepts the provided arguments.
        :rtype: bool
        """

        if self.args is None and self.kwargs is None:
            return False
        elif self.args is _any and self.kwargs is _any:
            return True
        elif args == self.args and kwargs == self.kwargs:
            return True
        elif len(args) != len(self.args) or len(kwargs) != len(self.kwargs):
            return False

        if not all(x == y or y == x for x, y in zip(args, self.args)):
            return False

        for key, value in self.kwargs.items():
            if key not in kwargs:
                return False
            elif not (kwargs[key] == value or value == kwargs[key]):
                return False

        return True

    def satisfy_custom_matcher(self, args, kwargs):
        """Return a boolean indicating if the args satisfy the stub

        :return: Whether or not the stub accepts the provided arguments.
        :rtype: bool
        """
        if not self._custom_matcher:
            return False
        try:
            return self._custom_matcher(*args, **kwargs)
        except Exception:
            return False

    def return_value(self, *args, **kwargs):
        """Extracts the real value to be returned from the wrapping callable.

        :return: The value the double should return when called.
        """

        self._called()
        return self._return_value(*args, **kwargs)

    def verify_arguments(self, args=None, kwargs=None):
        """Ensures that the arguments specified match the signature of the real method.

        :raise: ``VerifyingDoubleError`` if the arguments do not match.
        """

        args = self.args if args is None else args
        kwargs = self.kwargs if kwargs is None else kwargs

        try:
            verify_arguments(self._target, self._method_name, args, kwargs)
        except VerifyingBuiltinDoubleArgumentError:
            if dobles.lifecycle.ignore_builtin_verification():
                raise

    @verify_count_is_non_negative
    def exactly(self, n):
        """Set an exact call count allowance

        :param integer n:
        """

        self._call_counter.set_exact(n)
        return self

    @verify_count_is_non_negative
    def at_least(self, n):
        """Set a minimum call count allowance

        :param integer n:
        """

        self._call_counter.set_minimum(n)
        return self

    @verify_count_is_non_negative
    def at_most(self, n):
        """Set a maximum call count allowance

        :param integer n:
        """

        self._call_counter.set_maximum(n)
        return self

    def never(self):
        """Set an expected call count allowance of 0"""

        self.exactly(0)
        return self

    def once(self):
        """Set an expected call count allowance of 1"""

        self.exactly(1)
        return self

    def twice(self):
        """Set an expected call count allowance of 2"""

        self.exactly(2)
        return self

    @property
    def times(self):
        return self

    time = times

    def _called(self):
        """Indicate that the allowance was called

        :raise MockExpectationError if the allowance has been called too many times
        """

        if self._call_counter.called().has_too_many_calls():
            self.raise_failure_exception()

    def raise_failure_exception(self, expect_or_allow="Allowed"):
        """Raises a ``MockExpectationError`` with a useful message.

        :raise: ``MockExpectationError``
        """

        raise MockExpectationError(
            "{} '{}' to be called {}on {!r} with {}, but was not. ({}:{})".format(
                expect_or_allow,
                self._method_name,
                self._call_counter.error_string(),
                self._target.obj,
                self._expected_argument_string(),
                self._caller.filename,
                self._caller.lineno,
            )
        )

    def _expected_argument_string(self):
        """Generates a string describing what arguments the double expected.

        :return: A string describing expected arguments.
        :rtype: str
        """

        if self.args is _any and self.kwargs is _any:
            return "any args"
        elif self._custom_matcher:
            return "custom matcher: '{}'".format(self._custom_matcher.__name__)
        else:
            return build_argument_repr_string(self.args, self.kwargs)

1	import functools	1✔
2	import inspect	1✔
3	from typing import Any	1✔
4
5	import dobles.lifecycle	1✔
6	from dobles.call_count_accumulator import CallCountAccumulator	1✔
7	from dobles.exceptions import MockExpectationError, VerifyingBuiltinDoubleArgumentError	1✔
8	from dobles.verification import verify_arguments	1✔
9
10	_any = object()	1✔
11
12
13	async def _async_return_value(args, *kwargs):	1✔
14	return None	1✔
15
16
17	def _maybe_async(is_async: bool, return_value: Any):	1✔
18	if is_async:	1✔
19	return make_it_async(return_value)	1✔
20	return return_value	1✔
21
22
23	def make_it_async(	1✔
24	return_value=None,
25	exception=None,
26	):
27	async def inner(args, *kwargs):	1✔
28	if exception:	1✔
NEW 29	raise exception	×
30	return return_value	1✔
31
32	return inner()	1✔
33
34
35	def verify_count_is_non_negative(func):	1✔
36	@functools.wraps(func)	1✔
37	def inner(self, arg):	1✔
38	if arg < 0:	1✔
39	raise TypeError(func.__name__ + " requires one positive integer argument")	1✔
40	return func(self, arg)	1✔
41
42	return inner	1✔
43
44
45	def check_func_takes_args(func):	1✔
46	arg_spec = inspect.getfullargspec(func)	1✔
47	return any([arg_spec.args, arg_spec.varargs, arg_spec.varkw, arg_spec.defaults])	1✔
48
49
50	def build_argument_repr_string(args, kwargs):	1✔
51	args = [repr(x) for x in args]	1✔
52	kwargs = ["{}={!r}".format(k, v) for k, v in kwargs.items()]	1✔
53	return "({})".format(", ".join(args + kwargs))	1✔
54
55
56	class Allowance(object):	1✔
57	"""An individual method allowance (stub)."""
58
59	def __init__(self, target, method_name, caller):	1✔
60	"""
61	:param Target target: The object owning the method to stub.
62	:param str method_name: The name of the method to stub.
63	"""
64
65	self._target = target	1✔
66	self._method_name = method_name	1✔
67	self._caller = caller	1✔
68	self.args = _any	1✔
69	self.kwargs = _any	1✔
70	self._custom_matcher = None	1✔
71	self._is_satisfied = True	1✔
72	self._call_counter = CallCountAccumulator()	1✔
73	self._is_async = inspect.iscoroutinefunction(	1✔
74	target.get_attr(method_name).object
75	)
76
77	if self._is_async:	1✔
78	self._return_value = _async_return_value	1✔
79	else:
80	self._return_value = lambda args, *kwargs: None	1✔
81
82	def and_raise(self, exception, args, *kwargs):	1✔
83	"""Causes the double to raise the provided exception when called.
84
85	If provided, additional arguments (positional and keyword) passed to
86	`and_raise` are used in the exception instantiation.
87
88	:param Exception exception: The exception to raise.
89	"""
90
91	def proxy_exception(proxy_args, *proxy_kwargs):	1✔
92	raise exception	1✔
93
94	async def async_proxy_exception(proxy_args, *proxy_kwargs):	1✔
NEW 95	raise exception	×
96
97	self._return_value = (	1✔
98	async_proxy_exception if self._is_async else proxy_exception
99	)
100	return self	1✔
101
102	def and_return(self, *return_values):	1✔
103	"""Set a return value for an allowance
104
105	Causes the double to return the provided values in order. If multiple
106	values are provided, they are returned one at a time in sequence as the double is called.
107	If the double is called more times than there are return values, it should continue to
108	return the last value in the list.
109
110
111	:param object return_values: The values the double will return when called,
112	"""
113
114	if not return_values:	1✔
115	raise TypeError("and_return() expected at least 1 return value")	1✔
116
117	return_values = list(return_values)	1✔
118	final_value = return_values.pop()	1✔
119
120	self.and_return_result_of(	1✔
121	lambda: return_values.pop(0) if return_values else final_value
122	)
123	return self	1✔
124
125	def and_return_result_of(self, return_value):	1✔
126	"""Causes the double to return the result of calling the provided value.
127
128	:param return_value: A callable that will be invoked to determine the double's return value.
129	:type return_value: any callable object
130	"""
131
132	if not check_func_takes_args(return_value):	1✔
133	self._return_value = lambda args, *kwargs: _maybe_async(	1✔
134	self._is_async,
135	return_value(),
136	)
137	else:
138	self._return_value = _maybe_async(self._is_async, return_value)	1✔
139
140	return self	1✔
141
142	def is_satisfied(self):	1✔
143	"""Returns a boolean indicating whether or not the double has been satisfied.
144
145	Stubs are always satisfied, but mocks are only satisfied if they've been
146	called as was declared.
147
148	:return: Whether or not the double is satisfied.
149	:rtype: bool
150	"""
UNCOV 151	return self._is_satisfied	×
152
153	def with_args(self, args, *kwargs):	1✔
154	"""Declares that the double can only be called with the provided arguments.
155
156	:param args: Any positional arguments required for invocation.
157	:param kwargs: Any keyword arguments required for invocation.
158	"""
159
160	self.args = args	1✔
161	self.kwargs = kwargs	1✔
162	self.verify_arguments()	1✔
163	return self	1✔
164
165	def with_args_validator(self, matching_function):	1✔
166	"""Define a custom function for testing arguments
167
168	:param func matching_function: The function used to test arguments passed to the stub.
169	"""
170	self.args = None	1✔
171	self.kwargs = None	1✔
172	self._custom_matcher = matching_function	1✔
173	return self	1✔
174
175	def __call__(self, args, *kwargs):	1✔
176	"""A short hand syntax for with_args
177
178	Allows callers to do:
179	allow(module).foo.with_args(1, 2)
180	With:
181	allow(module).foo(1, 2)
182
183	:param args: Any positional arguments required for invocation.
184	:param kwargs: Any keyword arguments required for invocation.
185	"""
186	return self.with_args(args, *kwargs)	1✔
187
188	def with_no_args(self):	1✔
189	"""Declares that the double can only be called with no arguments."""
190
191	self.args = ()	1✔
192	self.kwargs = {}	1✔
193	self.verify_arguments()	1✔
194	return self	1✔
195
196	def satisfy_any_args_match(self):	1✔
197	"""Returns a boolean indicating whether or not the stub will accept arbitrary arguments.
198
199	This will be true unless the user has specified otherwise using ``with_args`` or
200	``with_no_args``.
201
202	:return: Whether or not the stub accepts arbitrary arguments.
203	:rtype: bool
204	"""
205
206	return self.args is _any and self.kwargs is _any	1✔
207
208	def satisfy_exact_match(self, args, kwargs):	1✔
209	"""Returns a boolean indicating whether or not the stub will accept the provided arguments.
210
211	:return: Whether or not the stub accepts the provided arguments.
212	:rtype: bool
213	"""
214
215	if self.args is None and self.kwargs is None:	1✔
216	return False	1✔
217	elif self.args is _any and self.kwargs is _any:	1✔
218	return True	1✔
219	elif args == self.args and kwargs == self.kwargs:	1✔
220	return True	1✔
221	elif len(args) != len(self.args) or len(kwargs) != len(self.kwargs):	1✔
222	return False	1✔
223
224	if not all(x == y or y == x for x, y in zip(args, self.args)):	1✔
225	return False	1✔
226
227	for key, value in self.kwargs.items():	1✔
228	if key not in kwargs:	1✔
229	return False	1✔
UNCOV 230	elif not (kwargs[key] == value or value == kwargs[key]):	×
UNCOV 231	return False	×
232
233	return True	1✔
234
235	def satisfy_custom_matcher(self, args, kwargs):	1✔
236	"""Return a boolean indicating if the args satisfy the stub
237
238	:return: Whether or not the stub accepts the provided arguments.
239	:rtype: bool
240	"""
241	if not self._custom_matcher:	1✔
242	return False	1✔
243	try:	1✔
244	return self._custom_matcher(args, *kwargs)	1✔
245	except Exception:	1✔
246	return False	1✔
247
248	def return_value(self, args, *kwargs):	1✔
249	"""Extracts the real value to be returned from the wrapping callable.
250
251	:return: The value the double should return when called.
252	"""
253
254	self._called()	1✔
255	return self._return_value(args, *kwargs)	1✔
256
257	def verify_arguments(self, args=None, kwargs=None):	1✔
258	"""Ensures that the arguments specified match the signature of the real method.
259
260	:raise: ``VerifyingDoubleError`` if the arguments do not match.
261	"""
262
263	args = self.args if args is None else args	1✔
264	kwargs = self.kwargs if kwargs is None else kwargs	1✔
265
266	try:	1✔
267	verify_arguments(self._target, self._method_name, args, kwargs)	1✔
268	except VerifyingBuiltinDoubleArgumentError:	1✔
269	if dobles.lifecycle.ignore_builtin_verification():	1✔
270	raise	1✔
271
272	@verify_count_is_non_negative	1✔
273	def exactly(self, n):	1✔
274	"""Set an exact call count allowance
275
276	:param integer n:
277	"""
278
279	self._call_counter.set_exact(n)	1✔
280	return self	1✔
281
282	@verify_count_is_non_negative	1✔
283	def at_least(self, n):	1✔
284	"""Set a minimum call count allowance
285
286	:param integer n:
287	"""
288
289	self._call_counter.set_minimum(n)	1✔
290	return self	1✔
291
292	@verify_count_is_non_negative	1✔
293	def at_most(self, n):	1✔
294	"""Set a maximum call count allowance
295
296	:param integer n:
297	"""
298
299	self._call_counter.set_maximum(n)	1✔
300	return self	1✔
301
302	def never(self):	1✔
303	"""Set an expected call count allowance of 0"""
304
305	self.exactly(0)	1✔
306	return self	1✔
307
308	def once(self):	1✔
309	"""Set an expected call count allowance of 1"""
310
311	self.exactly(1)	1✔
312	return self	1✔
313
314	def twice(self):	1✔
315	"""Set an expected call count allowance of 2"""
316
317	self.exactly(2)	1✔
318	return self	1✔
319
320	@property	1✔
321	def times(self):	1✔
322	return self	1✔
323
324	time = times	1✔
325
326	def _called(self):	1✔
327	"""Indicate that the allowance was called
328
329	:raise MockExpectationError if the allowance has been called too many times
330	"""
331
332	if self._call_counter.called().has_too_many_calls():	1✔
333	self.raise_failure_exception()	1✔
334
335	def raise_failure_exception(self, expect_or_allow="Allowed"):	1✔
336	"""Raises a ``MockExpectationError`` with a useful message.
337
338	:raise: ``MockExpectationError``
339	"""
340
341	raise MockExpectationError(	1✔
342	"{} '{}' to be called {}on {!r} with {}, but was not. ({}:{})".format(
343	expect_or_allow,
344	self._method_name,
345	self._call_counter.error_string(),
346	self._target.obj,
347	self._expected_argument_string(),
348	self._caller.filename,
349	self._caller.lineno,
350	)
351	)
352
353	def _expected_argument_string(self):	1✔
354	"""Generates a string describing what arguments the double expected.
355
356	:return: A string describing expected arguments.
357	:rtype: str
358	"""
359
360	if self.args is _any and self.kwargs is _any:	1✔
361	return "any args"	1✔
362	elif self._custom_matcher:	1✔
363	return "custom matcher: '{}'".format(self._custom_matcher.__name__)	1✔
364	else:
365	return build_argument_repr_string(self.args, self.kwargs)	1✔

smartfastlabs / dobles / 6755749014

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous