Files
2026-07-13 12:40:42 +08:00

727 lines
29 KiB
Python

# Copyright (c) 2018 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 unittest
import numpy
import numpy as np
import op_test
from op_test import get_device_place, is_custom_device
import paddle
from paddle import base
from paddle.base import core
from paddle.framework import in_pir_mode
def create_test_class(op_type, typename, callback, check_pir=False):
class Cls(op_test.OpTest):
def setUp(self):
a = numpy.random.random(size=(10, 7)).astype(typename)
b = numpy.random.random(size=(10, 7)).astype(typename)
c = callback(a, b)
self.python_api = eval("paddle." + op_type)
self.inputs = {'X': a, 'Y': b}
self.outputs = {'Out': c}
self.op_type = op_type
def test_output(self):
self.check_output(check_cinn=True, check_pir=check_pir)
def test_int16_support(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
a = paddle.static.data(name='a', shape=[-1, 2], dtype='int16')
b = paddle.static.data(name='b', shape=[-1, 2], dtype='int16')
op = eval(f"paddle.{self.op_type}")
try:
result = op(x=a, y=b)
except TypeError:
self.fail("TypeError should not be raised for int16 inputs")
cls_name = f"{op_type}_{typename}"
Cls.__name__ = cls_name
globals()[cls_name] = Cls
for _type_name in {
'float32',
'float64',
'uint8',
'int8',
'int16',
'int32',
'int64',
'float16',
}:
if _type_name == 'float64' and core.is_compiled_with_rocm():
_type_name = 'float32'
if _type_name == 'float16' and (
not (core.is_compiled_with_cuda() or is_custom_device())
):
continue
create_test_class('less_than', _type_name, lambda _a, _b: _a < _b, True)
create_test_class('less_equal', _type_name, lambda _a, _b: _a <= _b, True)
create_test_class('greater_than', _type_name, lambda _a, _b: _a > _b, True)
create_test_class(
'greater_equal', _type_name, lambda _a, _b: _a >= _b, True
)
create_test_class('equal', _type_name, lambda _a, _b: _a == _b, True)
create_test_class('not_equal', _type_name, lambda _a, _b: _a != _b, True)
def create_paddle_case(op_type, callback):
class PaddleCls(unittest.TestCase):
def setUp(self):
self.op_type = op_type
self.input_x = np.array([1, 2, 3, 4]).astype(np.int64)
self.input_y = np.array([1, 3, 2, 4]).astype(np.int64)
self.real_result = callback(self.input_x, self.input_y)
self.place = base.CPUPlace()
if core.is_compiled_with_cuda() or is_custom_device():
self.place = get_device_place()
def test_api(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
x = paddle.static.data(name='x', shape=[4], dtype='int64')
y = paddle.static.data(name='y', shape=[4], dtype='int64')
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
exe = base.Executor(self.place)
(res,) = exe.run(
feed={"x": self.input_x, "y": self.input_y},
fetch_list=[out],
)
self.assertEqual((res == self.real_result).all(), True)
def test_api_float(self):
if self.op_type == "equal":
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
x = paddle.static.data(name='x', shape=[4], dtype='int64')
y = paddle.static.data(name='y', shape=[], dtype='int64')
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
exe = base.Executor(self.place)
(res,) = exe.run(
feed={"x": self.input_x, "y": 1.0}, fetch_list=[out]
)
self.real_result = np.array([1, 0, 0, 0]).astype(np.int64)
self.assertEqual((res == self.real_result).all(), True)
def test_dynamic_api(self):
with paddle.base.dygraph.guard():
x = paddle.to_tensor(self.input_x)
y = paddle.to_tensor(self.input_y)
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
self.assertEqual((out.numpy() == self.real_result).all(), True)
def test_dynamic_api_int(self):
if self.op_type == "equal":
with paddle.base.dygraph.guard():
x = paddle.to_tensor(self.input_x)
op = eval(f"paddle.{self.op_type}")
out = op(x, 1)
self.real_result = np.array([1, 0, 0, 0]).astype(np.int64)
self.assertEqual(
(out.numpy() == self.real_result).all(), True
)
def test_dynamic_api_float(self):
if self.op_type == "equal":
with paddle.base.dygraph.guard():
x = paddle.to_tensor(self.input_x)
op = eval(f"paddle.{self.op_type}")
out = op(x, 1.0)
self.real_result = np.array([1, 0, 0, 0]).astype(np.int64)
self.assertEqual(
(out.numpy() == self.real_result).all(), True
)
def test_dynamic_api_float16(self):
with paddle.base.dygraph.guard():
x = paddle.to_tensor(self.input_x, dtype="float16")
y = paddle.to_tensor(self.input_y, dtype="float16")
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
self.assertEqual((out.numpy() == self.real_result).all(), True)
def test_dynamic_api_inf_1(self):
if self.op_type == "equal":
with paddle.base.dygraph.guard():
x1 = np.array([1, float('inf'), float('inf')]).astype(
np.int64
)
x = paddle.to_tensor(x1)
y1 = np.array([1, float('-inf'), float('inf')]).astype(
np.int64
)
y = paddle.to_tensor(y1)
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
self.real_result = (x1 == y1).astype(np.int64)
self.assertEqual(
(
out.numpy().astype(np.int64) == self.real_result
).all(),
True,
)
def test_dynamic_api_inf_2(self):
if self.op_type == "equal":
with paddle.base.dygraph.guard():
x1 = np.array([1, float('inf'), float('inf')]).astype(
np.float32
)
x = paddle.to_tensor(x1)
y1 = np.array([1, float('-inf'), float('inf')]).astype(
np.float32
)
y = paddle.to_tensor(y1)
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
self.real_result = (x1 == y1).astype(np.int64)
self.assertEqual(
(
out.numpy().astype(np.int64) == self.real_result
).all(),
True,
)
def test_dynamic_api_inf_3(self):
if self.op_type == "equal":
with paddle.base.dygraph.guard():
x1 = np.array([1, float('inf'), float('-inf')]).astype(
np.float32
)
x = paddle.to_tensor(x1)
y1 = np.array([1, 2, 3]).astype(np.float32)
y = paddle.to_tensor(y1)
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
self.real_result = (x1 == y1).astype(np.int64)
self.assertEqual(
(
out.numpy().astype(np.int64) == self.real_result
).all(),
True,
)
def test_dynamic_api_nan_1(self):
if self.op_type == "equal":
with paddle.base.dygraph.guard():
x1 = np.array([1, float('nan'), float('nan')]).astype(
np.int64
)
x = paddle.to_tensor(x1)
y1 = np.array([1, float('-nan'), float('nan')]).astype(
np.int64
)
y = paddle.to_tensor(y1)
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
self.real_result = (x1 == y1).astype(np.int64)
self.assertEqual(
(
out.numpy().astype(np.int64) == self.real_result
).all(),
True,
)
def test_dynamic_api_nan_2(self):
if self.op_type == "equal":
with paddle.base.dygraph.guard():
x1 = np.array([1, float('nan'), float('nan')]).astype(
np.float32
)
x = paddle.to_tensor(x1)
y1 = np.array([1, float('-nan'), float('nan')]).astype(
np.float32
)
y = paddle.to_tensor(y1)
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
self.real_result = (x1 == y1).astype(np.int64)
self.assertEqual(
(
out.numpy().astype(np.int64) == self.real_result
).all(),
True,
)
def test_dynamic_api_nan_3(self):
if self.op_type == "equal":
with paddle.base.dygraph.guard():
x1 = np.array([1, float('-nan'), float('nan')]).astype(
np.float32
)
x = paddle.to_tensor(x1)
y1 = np.array([1, 2, 1]).astype(np.float32)
y = paddle.to_tensor(y1)
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
self.real_result = (x1 == y1).astype(np.int64)
self.assertEqual(
(
out.numpy().astype(np.int64) == self.real_result
).all(),
True,
)
def test_not_equal(self):
if self.op_type == "not_equal":
with paddle.base.dygraph.guard():
x = paddle.to_tensor(
np.array([1.2e-15, 2, 2, 1]), dtype="float32"
)
y = paddle.to_tensor(
np.array([1.1e-15, 2, 2, 1]), dtype="float32"
)
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
self.real_result = np.array([0, 0, 0, 0]).astype(np.int64)
self.assertEqual(
(out.numpy() == self.real_result).all(), True
)
def test_assert(self):
def test_dynamic_api_string(self):
if self.op_type == "equal":
with paddle.base.dygraph.guard():
x = paddle.to_tensor(self.input_x)
op = eval(f"paddle.{self.op_type}")
out = op(x, "1.0")
self.assertRaises(TypeError, test_dynamic_api_string)
def test_dynamic_api_bool(self):
if self.op_type == "equal":
with paddle.base.dygraph.guard():
x = paddle.to_tensor(self.input_x)
op = eval(f"paddle.{self.op_type}")
out = op(x, True)
self.real_result = np.array([1, 0, 0, 0]).astype(np.int64)
self.assertEqual(
(out.numpy() == self.real_result).all(), True
)
def test_broadcast_api_1(self):
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
x = paddle.static.data(
name='x', shape=[1, 2, 1, 3], dtype='int32'
)
y = paddle.static.data(name='y', shape=[1, 2, 3], dtype='int32')
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
exe = paddle.static.Executor(self.place)
input_x = np.arange(1, 7).reshape((1, 2, 1, 3)).astype(np.int32)
input_y = np.arange(0, 6).reshape((1, 2, 3)).astype(np.int32)
real_result = callback(input_x, input_y)
(res,) = exe.run(
feed={"x": input_x, "y": input_y}, fetch_list=[out]
)
self.assertEqual((res == real_result).all(), True)
def test_broadcast_api_2(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
x = paddle.static.data(name='x', shape=[1, 2, 3], dtype='int32')
y = paddle.static.data(
name='y', shape=[1, 2, 1, 3], dtype='int32'
)
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
exe = paddle.static.Executor(self.place)
input_x = np.arange(0, 6).reshape((1, 2, 3)).astype(np.int32)
input_y = np.arange(1, 7).reshape((1, 2, 1, 3)).astype(np.int32)
real_result = callback(input_x, input_y)
(res,) = exe.run(
feed={"x": input_x, "y": input_y}, fetch_list=[out]
)
self.assertEqual((res == real_result).all(), True)
def test_broadcast_api_3(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
x = paddle.static.data(name='x', shape=[5], dtype='int32')
y = paddle.static.data(name='y', shape=[3, 1], dtype='int32')
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
exe = paddle.static.Executor(self.place)
input_x = np.arange(0, 5).reshape(5).astype(np.int32)
input_y = np.array([5, 3, 2]).reshape((3, 1)).astype(np.int32)
real_result = callback(input_x, input_y)
(res,) = exe.run(
feed={"x": input_x, "y": input_y}, fetch_list=[out]
)
self.assertEqual((res == real_result).all(), True)
def test_zero_dim_api_1(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
x = paddle.randint(-3, 3, shape=[], dtype='int32')
y = paddle.randint(-3, 3, shape=[], dtype='int32')
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
exe = paddle.static.Executor(self.place)
(
x_np,
y_np,
res,
) = exe.run(fetch_list=[x, y, out])
real_result = callback(x_np, y_np)
self.assertEqual((res == real_result).all(), True)
def test_zero_dim_api_2(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
x = paddle.randint(-3, 3, shape=[2, 3, 4], dtype='int32')
y = paddle.randint(-3, 3, shape=[], dtype='int32')
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
exe = paddle.static.Executor(self.place)
(
x_np,
y_np,
res,
) = exe.run(fetch_list=[x, y, out])
real_result = callback(x_np, y_np)
self.assertEqual((res == real_result).all(), True)
def test_zero_dim_api_3(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
x = paddle.randint(-3, 3, shape=[], dtype='int32')
y = paddle.randint(-3, 3, shape=[2, 3, 4], dtype='int32')
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
exe = paddle.static.Executor(self.place)
(
x_np,
y_np,
res,
) = exe.run(fetch_list=[x, y, out])
real_result = callback(x_np, y_np)
self.assertEqual((res == real_result).all(), True)
def test_bool_api_4(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
x = paddle.static.data(name='x', shape=[3, 1], dtype='bool')
y = paddle.static.data(name='y', shape=[3, 1], dtype='bool')
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
exe = paddle.static.Executor(self.place)
input_x = np.array([True, False, True]).astype(np.bool_)
input_y = np.array([True, True, False]).astype(np.bool_)
real_result = callback(input_x, input_y)
(res,) = exe.run(
feed={"x": input_x, "y": input_y}, fetch_list=[out]
)
self.assertEqual((res == real_result).all(), True)
def test_bool_broadcast_api_4(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
x = paddle.static.data(name='x', shape=[3, 1], dtype='bool')
y = paddle.static.data(name='y', shape=[1], dtype='bool')
op = eval(f"paddle.{self.op_type}")
out = op(x, y)
exe = paddle.static.Executor(self.place)
input_x = np.array([True, False, True]).astype(np.bool_)
input_y = np.array([True]).astype(np.bool_)
real_result = callback(input_x, input_y)
(res,) = exe.run(
feed={"x": input_x, "y": input_y}, fetch_list=[out]
)
self.assertEqual((res == real_result).all(), True)
def test_attr_name(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
x = paddle.static.data(name='x', shape=[-1, 4], dtype='int32')
y = paddle.static.data(name='y', shape=[-1, 4], dtype='int32')
op = eval(f"paddle.{self.op_type}")
out = op(x=x, y=y, name=f"name_{self.op_type}")
if not in_pir_mode():
self.assertEqual(f"name_{self.op_type}" in out.name, True)
cls_name = f"TestCase_{op_type}"
PaddleCls.__name__ = cls_name
globals()[cls_name] = PaddleCls
create_paddle_case('less_than', lambda _a, _b: _a < _b)
create_paddle_case('less_equal', lambda _a, _b: _a <= _b)
create_paddle_case('greater_than', lambda _a, _b: _a > _b)
create_paddle_case('greater_equal', lambda _a, _b: _a >= _b)
create_paddle_case('equal', lambda _a, _b: _a == _b)
create_paddle_case('not_equal', lambda _a, _b: _a != _b)
# add bf16 tests
def create_bf16_case(op_type, callback, check_pir=False):
class TestCompareOpBF16Op(op_test.OpTest):
def setUp(self):
self.op_type = op_type
self.dtype = np.uint16
self.python_api = eval("paddle." + op_type)
x = np.random.uniform(0, 1, [5, 5]).astype(np.float32)
y = np.random.uniform(0, 1, [5, 5]).astype(np.float32)
real_result = callback(x, y)
self.inputs = {
'X': op_test.convert_float_to_uint16(x),
'Y': op_test.convert_float_to_uint16(y),
}
self.outputs = {'Out': real_result}
def test_check_output(self):
self.check_output(check_cinn=True, check_pir=check_pir)
cls_name = f"BF16TestCase_{op_type}"
TestCompareOpBF16Op.__name__ = cls_name
globals()[cls_name] = TestCompareOpBF16Op
create_bf16_case('less_than', lambda _a, _b: _a < _b, True)
create_bf16_case('less_equal', lambda _a, _b: _a <= _b, True)
create_bf16_case('greater_than', lambda _a, _b: _a > _b, True)
create_bf16_case('greater_equal', lambda _a, _b: _a >= _b, True)
create_bf16_case('equal', lambda _a, _b: _a == _b, True)
create_bf16_case('not_equal', lambda _a, _b: _a != _b, True)
class TestCompareOpError(unittest.TestCase):
def test_int16_support(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
# The input x and y of compare_op must be Variable.
x = paddle.static.data(name='x', shape=[-1, 1], dtype="float32")
y = base.create_lod_tensor(
numpy.array([[-1]]), [[1]], base.CPUPlace()
)
self.assertRaises(TypeError, paddle.greater_equal, x, y)
class API_TestElementwise_Equal(unittest.TestCase):
def test_api(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
label = paddle.assign(np.array([3, 3], dtype="int32"))
limit = paddle.assign(np.array([3, 2], dtype="int32"))
out = paddle.equal(x=label, y=limit)
place = base.CPUPlace()
exe = base.Executor(place)
(res,) = exe.run(fetch_list=[out])
self.assertEqual((res == np.array([True, False])).all(), True)
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
label = paddle.assign(np.array([3, 3], dtype="int32"))
limit = paddle.assign(np.array([3, 3], dtype="int32"))
out = paddle.equal(x=label, y=limit)
place = base.CPUPlace()
exe = base.Executor(place)
(res,) = exe.run(fetch_list=[out])
self.assertEqual((res == np.array([True, True])).all(), True)
def test_api_fp16(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
label = paddle.to_tensor([3, 3], dtype="float16")
limit = paddle.to_tensor([3, 2], dtype="float16")
out = paddle.equal(x=label, y=limit)
if core.is_compiled_with_cuda() or is_custom_device():
place = get_device_place()
exe = base.Executor(place)
(res,) = exe.run(fetch_list=[out])
self.assertEqual((res == np.array([True, False])).all(), True)
class API_TestElementwise_Greater_Than(unittest.TestCase):
def test_api_fp16(self):
paddle.enable_static()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
label = paddle.to_tensor([3, 3], dtype="float16")
limit = paddle.to_tensor([3, 2], dtype="float16")
out = paddle.greater_than(x=label, y=limit)
if core.is_compiled_with_cuda() or is_custom_device():
place = get_device_place()
exe = paddle.static.Executor(place)
(res,) = exe.run(fetch_list=[out])
self.assertEqual((res == np.array([False, True])).all(), True)
class TestCompareOpPlace(unittest.TestCase):
def test_place_1(self):
paddle.enable_static()
place = paddle.CPUPlace()
if core.is_compiled_with_cuda() or is_custom_device():
place = get_device_place()
with paddle.static.program_guard(
paddle.static.Program(), paddle.static.Program()
):
label = paddle.assign(np.array([3, 3], dtype="int32"))
limit = paddle.assign(np.array([3, 2], dtype="int32"))
out = paddle.less_than(label, limit)
exe = base.Executor(place)
(res,) = exe.run(fetch_list=[out])
self.assertEqual((res == np.array([False, False])).all(), True)
def test_place_2(self):
place = paddle.CPUPlace()
data_place = place
if core.is_compiled_with_cuda() or is_custom_device():
place = get_device_place()
data_place = paddle.CUDAPinnedPlace()
paddle.disable_static(place)
data = np.array([9], dtype="int64")
data_tensor = paddle.to_tensor(data, place=data_place)
result = data_tensor == 0
self.assertEqual((result.numpy() == np.array([False])).all(), True)
class TestCompareOutAndParamAlias(unittest.TestCase):
def setUp(self) -> None:
self.shape = [2, 3, 4, 5]
self.api_names = [
"equal", # eq
"equal",
"not_equal", # ne
"not_equal",
"less_than", # lt
"less_than", # less
"less_equal", # le
"less_equal",
"greater_than", # gt
"greater_than", # greater
"greater_equal", # ge
"greater_equal",
]
self.apis = [getattr(paddle, name) for name in self.api_names]
self.np_apis = [
np.equal,
np.equal,
np.not_equal,
np.not_equal,
np.less,
np.less,
np.less_equal,
np.less_equal,
np.greater,
np.greater,
np.greater_equal,
np.greater_equal,
]
self.input = np.random.rand(*self.shape).astype(np.float32)
self.other = np.random.rand(*self.shape).astype(np.float32)
self.other[0, 0, 3, 0] = self.input[0, 0, 3, 0]
def test_dygraph_out(self):
paddle.disable_static()
for api, np_api in zip(self.apis, self.np_apis):
x = paddle.to_tensor(self.input)
y = paddle.to_tensor(self.other)
out_holder = paddle.zeros_like(x)
out = api(x, y)
out_holder[:] = out
np.testing.assert_allclose(
out_holder.numpy(), np_api(self.input, self.other)
)
def test_dygraph_param_alias(self):
paddle.disable_static()
for api, np_api in zip(self.apis, self.np_apis):
x = paddle.to_tensor(self.input)
y = paddle.to_tensor(self.other)
out1 = api(x, y)
out2 = api(x, y)
out3 = api(x, y)
out4 = api(x, y)
for out in [out1, out2, out3, out4]:
np.testing.assert_allclose(
out.numpy(), np_api(self.input, self.other)
)
def test_dygraph_param_alias_out(self):
paddle.disable_static()
for api, np_api in zip(self.apis, self.np_apis):
x = paddle.to_tensor(self.input)
y = paddle.to_tensor(self.other)
out_holders = [paddle.zeros_like(x) for _ in range(4)]
out_holders[0][:] = api(x, y)
out_holders[1][:] = api(x, y)
out_holders[2][:] = api(x, y)
out_holders[3][:] = api(x, y)
for out in out_holders:
np.testing.assert_allclose(
out.numpy(), np_api(self.input, self.other)
)
def test_tensor_api_dygraph_param_alias(self):
paddle.disable_static()
for api, np_api in zip(self.api_names, self.np_apis):
x = paddle.to_tensor(self.input)
y = paddle.to_tensor(self.other)
api = getattr(x, api)
out1 = api(y)
out2 = api(y)
for out in [out1, out2]:
np.testing.assert_allclose(
out.numpy(), np_api(self.input, self.other)
)
if __name__ == '__main__':
paddle.enable_static()
unittest.main()