# Copyright (c) 2013 Potential Ventures Ltd
# Copyright (c) 2013 SolarFlare Communications Inc
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# * Neither the name of Potential Ventures Ltd,
# SolarFlare Communications Inc nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL POTENTIAL VENTURES LTD BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import functools
import sys
import typing
import warnings
import cocotb
import cocotb.triggers
from cocotb.log import SimLog
from cocotb.task import Task as _Task # noqa: F401
from cocotb.task import _RunningCoroutine, _RunningTest
from cocotb.utils import lazy_property
def public(f):
"""Use a decorator to avoid retyping function/class names.
* Based on an idea by Duncan Booth:
http://groups.google.com/group/comp.lang.python/msg/11cbb03e09611b8a
* Improved via a suggestion by Dave Angel:
http://groups.google.com/group/comp.lang.python/msg/3d400fb22d8a42e1
"""
all = sys.modules[f.__module__].__dict__.setdefault("__all__", [])
if f.__name__ not in all: # Prevent duplicates if run from an IDE.
all.append(f.__name__)
return f
public(public) # Emulate decorating ourself
[docs]
class coroutine:
"""Decorator class that allows us to provide common coroutine mechanisms:
``log`` methods will log to ``cocotb.coroutine.name``.
:meth:`~cocotb.task.Task.join` method returns an event which will fire when the coroutine exits.
Used as ``@cocotb.coroutine``.
"""
def __init__(self, func):
self._func = func
functools.update_wrapper(self, func)
@lazy_property
def log(self):
return SimLog(f"cocotb.coroutine.{self._func.__qualname__}.{id(self)}")
def __call__(self, *args, **kwargs):
return _RunningCoroutine(self._func(*args, **kwargs), self)
def __get__(self, obj, owner=None):
"""Permit the decorator to be used on class methods
and standalone functions"""
return type(self)(self._func.__get__(obj, owner))
def __iter__(self):
return self
def __str__(self):
return str(self._func.__qualname__)
[docs]
@public
class function:
"""Decorator class that allows a function to block.
This allows a coroutine that consumes simulation time
to be called by a thread started with :class:`cocotb.external`;
in other words, to internally block while externally
appear to yield.
"""
def __init__(self, func):
self._coro = cocotb.coroutine(func)
@lazy_property
def log(self):
return SimLog(f"cocotb.function.{self._coro.__qualname__}.{id(self)}")
def __call__(self, *args, **kwargs):
return cocotb.scheduler._queue_function(self._coro(*args, **kwargs))
def __get__(self, obj, owner=None):
"""Permit the decorator to be used on class methods
and standalone functions"""
return type(self)(self._coro._func.__get__(obj, owner))
[docs]
@public
class external:
"""Decorator to apply to an external function to enable calling from cocotb.
This turns a normal function that isn't a coroutine into a blocking coroutine.
Currently, this creates a new execution thread for each function that is
called.
Scope for this to be streamlined to a queue in future.
"""
def __init__(self, func):
self._func = func
self._log = SimLog(f"cocotb.external.{self._func.__qualname__}.{id(self)}")
def __call__(self, *args, **kwargs):
return cocotb.scheduler._run_in_executor(self._func, *args, **kwargs)
def __get__(self, obj, owner=None):
"""Permit the decorator to be used on class methods
and standalone functions"""
return type(self)(self._func.__get__(obj, owner))
class _decorator_helper(type):
"""
Metaclass that allows a type to be constructed using decorator syntax,
passing the decorated function as the first argument. Supports
construction with or without having the type called.
So:
@MyClass(construction, args='go here')
def this_is_passed_as_f(...):
pass
ends up calling
MyClass.__init__(this_is_passed_as_f, construction, args='go here')
"""
def __call__(cls, *args, **kwargs):
if len(args) == 1 and callable(args[0]): # case without parenthesis
f = args[0]
return type.__call__(cls, f, **kwargs)
# case with parenthesis
def decorator(f):
# fall back to the normal way of constructing an object, now that
# we have all the arguments
return type.__call__(cls, f, *args, **kwargs)
return decorator
[docs]
@public
class test(coroutine, metaclass=_decorator_helper):
"""
Decorator to mark a Callable which returns a Coroutine as a test.
The test decorator provides a test timeout, and allows us to mark tests as skipped
or expecting errors or failures.
Tests are evaluated in the order they are defined in a test module.
Used as ``@cocotb.test(...)``.
Args:
timeout_time (numbers.Real or decimal.Decimal, optional):
Simulation time duration before timeout occurs.
.. versionadded:: 1.3
.. note::
Test timeout is intended for protection against deadlock.
Users should use :class:`~cocotb.triggers.with_timeout` if they require a
more general-purpose timeout mechanism.
timeout_unit (str, optional):
Units of timeout_time, accepts any units that :class:`~cocotb.triggers.Timer` does.
.. versionadded:: 1.3
.. deprecated:: 1.5
Using ``None`` as the *timeout_unit* argument is deprecated, use ``'step'`` instead.
expect_fail (bool, optional):
If ``True`` and the test fails a functional check via an ``assert`` statement, :class:`pytest.raises`,
:class:`pytest.warns`, or :class:`pytest.deprecated_call` the test is considered to have passed.
If ``True`` and the test passes successfully, the test is considered to have failed.
expect_error (exception type or tuple of exception types, optional):
Mark the result as a pass only if one of the exception types is raised in the test.
This is primarily for cocotb internal regression use for when a simulator error is expected.
Users are encouraged to use the following idiom instead::
@cocotb.test()
async def my_test(dut):
try:
await thing_that_should_fail()
except ExceptionIExpect:
pass
else:
assert False, "Exception did not occur"
.. versionchanged:: 1.3
Specific exception types can be expected
.. deprecated:: 1.5
Passing a :class:`bool` value is now deprecated.
Pass a specific :class:`Exception` or a tuple of Exceptions instead.
skip (bool, optional):
Don't execute this test as part of the regression. Test can still be run
manually by setting :make:var:`TESTCASE`.
stage (int)
Order tests logically into stages, where multiple tests can share a stage.
Defaults to 0.
"""
_id_count = 0 # used by the RegressionManager to sort tests in definition order
def __init__(
self,
f,
timeout_time=None,
timeout_unit="step",
expect_fail=False,
expect_error=(),
skip=False,
stage=0,
):
if timeout_unit is None:
warnings.warn(
'Using timeout_unit=None is deprecated, use timeout_unit="step" instead.',
DeprecationWarning,
stacklevel=2,
)
timeout_unit = "step" # don't propagate deprecated value
self._id = self._id_count
type(self)._id_count += 1
if timeout_time is not None:
co = coroutine(f)
@functools.wraps(f)
async def f(*args, **kwargs):
running_co = co(*args, **kwargs)
try:
res = await cocotb.triggers.with_timeout(
running_co, self.timeout_time, self.timeout_unit
)
except cocotb.result.SimTimeoutError:
running_co.kill()
raise
else:
return res
super().__init__(f)
self.timeout_time = timeout_time
self.timeout_unit = timeout_unit
self.expect_fail = expect_fail
if isinstance(expect_error, bool):
warnings.warn(
"Passing bool values to `except_error` option of `cocotb.test` is deprecated. "
"Pass a specific Exception type instead",
DeprecationWarning,
stacklevel=2,
)
if expect_error is True:
expect_error = (BaseException,)
elif expect_error is False:
expect_error = ()
self.expect_error = expect_error
self.skip = skip
self.stage = stage
self.im_test = True # For auto-regressions
self.name = self._func.__name__
def __call__(self, *args, **kwargs):
inst = self._func(*args, **kwargs)
coro = _RunningTest(inst, self)
return coro
if sys.version_info < (3, 7):
Task = _Task
RunningTask = _Task
RunningCoroutine = _RunningCoroutine
RunningTest = _RunningTest
else:
def __getattr__(attr: str) -> typing.Any:
if attr in ("Task", "RunningTask"):
warnings.warn(
f"The class {attr} has been renamed to cocotb.task.Task.",
DeprecationWarning,
stacklevel=2,
)
attr = "_Task"
elif attr in ("RunningCoroutine", "RunningTest"):
warnings.warn(
f"The class {attr} is now private. Update all uses to the parent class cocotb.task.Task.",
DeprecationWarning,
stacklevel=2,
)
attr = f"_{attr}"
try:
return globals()[attr]
except KeyError:
raise AttributeError(
f"module {__name__!r} has no attribute {attr!r}"
) from None