wehub-resource-sync
241 lines
7.4 KiB
Python
241 lines
7.4 KiB
Python
# Copyright (c) 2021 PaddlePaddle Authors. All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
import collections
|
|
import functools
|
|
import inspect
|
|
import re
|
|
import sys
|
|
from unittest import SkipTest
|
|
|
|
import numpy as np
|
|
from distribution import config
|
|
|
|
TEST_CASE_NAME = 'suffix'
|
|
|
|
|
|
def xrand(shape=(10, 10, 10), dtype=config.DEFAULT_DTYPE, min=1.0, max=10.0):
|
|
return (np.random.rand(*shape).astype(dtype)) * (max - min) + min
|
|
|
|
|
|
def place(devices, key='place'):
|
|
def decorate(cls):
|
|
module = sys.modules[cls.__module__].__dict__
|
|
raw_classes = {
|
|
k: v for k, v in module.items() if k.startswith(cls.__name__)
|
|
}
|
|
|
|
for raw_name, raw_cls in raw_classes.items():
|
|
for d in devices:
|
|
test_cls = dict(raw_cls.__dict__)
|
|
test_cls.update({key: d})
|
|
new_name = raw_name + '.' + d.__class__.__name__
|
|
module[new_name] = type(new_name, (raw_cls,), test_cls)
|
|
del module[raw_name]
|
|
return cls
|
|
|
|
return decorate
|
|
|
|
|
|
def parameterize_cls(fields, values=None, test_pir=False):
|
|
fields = [fields] if isinstance(fields, str) else fields
|
|
params = [dict(zip(fields, vals)) for vals in values]
|
|
|
|
def decorate(cls):
|
|
test_cls_module = sys.modules[cls.__module__].__dict__
|
|
for k, v in enumerate(params):
|
|
test_cls = dict(cls.__dict__)
|
|
test_cls.update(v)
|
|
test_cls["test_pir"] = False
|
|
name = cls.__name__ + str(k)
|
|
name = name + '.' + v.get('suffix') if v.get('suffix') else name
|
|
test_cls_module[name] = type(name, (cls,), test_cls)
|
|
if test_pir:
|
|
name = name + ".pir"
|
|
test_cls["test_pir"] = True
|
|
pir_type = type(name, (cls,), test_cls)
|
|
test_cls_module[name] = pir_type
|
|
|
|
for m in list(cls.__dict__):
|
|
if m.startswith("test"):
|
|
delattr(cls, m)
|
|
return cls
|
|
|
|
return decorate
|
|
|
|
|
|
def parameterize_func(
|
|
input, name_func=None, doc_func=None, skip_on_empty=False
|
|
):
|
|
name_func = name_func or default_name_func
|
|
|
|
def wrapper(f, instance=None):
|
|
frame_locals = inspect.currentframe().f_back.f_locals
|
|
|
|
parameters = input_as_callable(input)()
|
|
|
|
if not parameters:
|
|
if not skip_on_empty:
|
|
raise ValueError(
|
|
"Parameters iterable is empty (hint: use "
|
|
"`parameterized.expand([], skip_on_empty=True)` to skip "
|
|
"this test when the input is empty)"
|
|
)
|
|
return functools.wraps(f)(skip_on_empty_helper)
|
|
|
|
digits = len(str(len(parameters) - 1))
|
|
for num, p in enumerate(parameters):
|
|
name = name_func(
|
|
f, "{num:0>{digits}}".format(digits=digits, num=num), p
|
|
)
|
|
# If the original function has patches applied by 'mock.patch',
|
|
# re-construct all patches on the just former decoration layer
|
|
# of param_as_standalone_func so as not to share
|
|
# patch objects between new functions
|
|
nf = reapply_patches_if_need(f)
|
|
frame_locals[name] = param_as_standalone_func(p, nf, name)
|
|
frame_locals[name].__doc__ = f.__doc__
|
|
|
|
# Delete original patches to prevent new function from evaluating
|
|
# original patching object as well as re-constrfucted patches.
|
|
delete_patches_if_need(f)
|
|
|
|
f.__test__ = False
|
|
|
|
return wrapper
|
|
|
|
|
|
def reapply_patches_if_need(func):
|
|
def dummy_wrapper(orgfunc):
|
|
@functools.wraps(orgfunc)
|
|
def dummy_func(*args, **kwargs):
|
|
return orgfunc(*args, **kwargs)
|
|
|
|
return dummy_func
|
|
|
|
if hasattr(func, 'patchings'):
|
|
func = dummy_wrapper(func)
|
|
tmp_patchings = func.patchings
|
|
delattr(func, 'patchings')
|
|
for patch_obj in tmp_patchings:
|
|
func = patch_obj.decorate_callable(func)
|
|
return func
|
|
|
|
|
|
def delete_patches_if_need(func):
|
|
if hasattr(func, 'patchings'):
|
|
func.patchings[:] = []
|
|
|
|
|
|
def default_name_func(func, num, p):
|
|
base_name = func.__name__
|
|
name_suffix = f"_{num}"
|
|
|
|
if len(p.args) > 0 and isinstance(p.args[0], str):
|
|
name_suffix += "_" + to_safe_name(p.args[0])
|
|
return base_name + name_suffix
|
|
|
|
|
|
def param_as_standalone_func(p, func, name):
|
|
@functools.wraps(func)
|
|
def standalone_func(*a):
|
|
return func(*(a + p.args), **p.kwargs)
|
|
|
|
standalone_func.__name__ = name
|
|
|
|
# place_as is used by py.test to determine what source file should be
|
|
# used for this test.
|
|
standalone_func.place_as = func
|
|
|
|
# Remove __wrapped__ because py.test will try to look at __wrapped__
|
|
# to determine which parameters should be used with this test case,
|
|
# and obviously we don't need it to do any parameterization.
|
|
try:
|
|
del standalone_func.__wrapped__
|
|
except AttributeError:
|
|
pass
|
|
return standalone_func
|
|
|
|
|
|
def input_as_callable(input):
|
|
if callable(input):
|
|
return lambda: check_input_values(input())
|
|
input_values = check_input_values(input)
|
|
return lambda: input_values
|
|
|
|
|
|
def check_input_values(input_values):
|
|
if not isinstance(input_values, list):
|
|
input_values = list(input_values)
|
|
return [param.from_decorator(p) for p in input_values]
|
|
|
|
|
|
def skip_on_empty_helper(*a, **kw):
|
|
raise SkipTest("parameterized input is empty")
|
|
|
|
|
|
_param = collections.namedtuple("param", "args kwargs")
|
|
|
|
|
|
class param(_param):
|
|
def __new__(cls, *args, **kwargs):
|
|
return _param.__new__(cls, args, kwargs)
|
|
|
|
@classmethod
|
|
def explicit(cls, args=None, kwargs=None):
|
|
"""Creates a ``param`` by explicitly specifying ``args`` and
|
|
``kwargs``::
|
|
>>> param.explicit([1,2,3])
|
|
param(*(1, 2, 3))
|
|
>>> param.explicit(kwargs={"foo": 42})
|
|
param(*(), **{"foo": "42"})
|
|
"""
|
|
args = args or ()
|
|
kwargs = kwargs or {}
|
|
return cls(*args, **kwargs)
|
|
|
|
@classmethod
|
|
def from_decorator(cls, args):
|
|
"""Returns an instance of ``param()`` for ``@parameterized`` argument
|
|
``args``::
|
|
>>> param.from_decorator((42, ))
|
|
param(args=(42, ), kwargs={})
|
|
>>> param.from_decorator("foo")
|
|
param(args=("foo", ), kwargs={})
|
|
"""
|
|
if isinstance(args, param):
|
|
return args
|
|
elif isinstance(args, str):
|
|
args = (args,)
|
|
try:
|
|
return cls(*args)
|
|
except TypeError as e:
|
|
if "after * must be" not in str(e):
|
|
raise
|
|
raise TypeError(
|
|
f"Parameters must be tuples, but {args!r} is not (hint: use '({args!r}, )')",
|
|
)
|
|
|
|
def __repr__(self):
|
|
return "param(*{!r}, **{!r})".format(*self)
|
|
|
|
|
|
def to_safe_name(s):
|
|
return str(re.sub("[^a-zA-Z0-9_]+", "_", s))
|
|
|
|
|
|
# alias
|
|
parameterize = parameterize_func
|
|
param_cls = parameterize_cls
|
|
param_func = parameterize_func
|