"""
Tests for normalize refactoring in testcase_op.py:
- _normalize_scalar_field_by_dist (string/scalar fields)
- _normalize_shape_field_by_dist (shape fields)
- _normalize_range_field_by_dist (range/pair fields)
- Distribution computation (_compute_input/output_distribution)
- flat_* properties after normalize
- ELEWISE/REDUCE output resolution before output_distribution
- Edge cases: None, empty, single float, compressed, already nested, length mismatch
"""
import pytest
from unittest.mock import patch
from ttk.core_modules.testcase_manager.testcase_op import TestcaseOp
def _make_testcase(op_name="Add", input_shapes=((8,), (8,)),
input_dtypes=("float16", "float16"),
output_shapes=((8,),),
output_dtypes=("float16",),
const_input_indexes=None,
**kwargs):
case = TestcaseOp()
case.testcase_name = f"test_{op_name or 'None'}"
case.op_name = op_name
case.input_shapes = input_shapes
case.input_dtypes = input_dtypes
case.output_shapes = output_shapes
case.output_dtypes = output_dtypes
if const_input_indexes is not None:
case.const_input_indexes = const_input_indexes
case.input_ori_shapes = kwargs.pop("input_ori_shapes", input_shapes)
case.output_ori_shapes = kwargs.pop("output_ori_shapes", output_shapes)
case.attributes = kwargs.pop("attributes", {})
n_in = len(input_shapes)
n_out = len(output_shapes or ())
case.input_formats = kwargs.pop("input_formats", ("ND",) * n_in)
case.input_ori_formats = kwargs.pop("input_ori_formats", ("ND",) * n_in)
case.output_formats = kwargs.pop("output_formats", ("ND",) * n_out)
case.output_ori_formats = kwargs.pop("output_ori_formats", ("ND",) * n_out)
case.input_data_ranges = kwargs.pop("input_data_ranges", (None,) * n_in)
for k, v in kwargs.items():
setattr(case, k, v)
return case
@pytest.fixture(autouse=True)
def _mock_op_info(monkeypatch):
monkeypatch.delenv("ASCEND_HOME_PATH", raising=False)
monkeypatch.delenv("ASCEND_TOOLKIT_HOME", raising=False)
monkeypatch.delenv("ASCEND_OPP_PATH", raising=False)
def _validate(case):
n_in = len(case.input_shapes) if case.input_shapes else 0
n_out = len(case.output_shapes) if case.output_shapes and not isinstance(case.output_shapes, str) else 0
with patch('ttk.core_modules.operator.op_info_keeper.OpInfoKeeper') as mock:
mock.return_value.info_of.return_value = {
"coreType.value": "AiCore",
"inputs": [{"name": f"i{i}"} for i in range(n_in)],
"outputs": [{"name": f"o{i}"} for i in range(n_out)],
}
case.validate()
class TestDistributionComputation:
def test_input_dist_nested(self):
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)))
_validate(case)
assert case.input_distribution == (2, 0)
def test_input_dist_flat(self):
case = _make_testcase(input_shapes=((8,), (8,)))
_validate(case)
assert case.input_distribution == (0, 0)
def test_input_dist_empty(self):
case = _make_testcase(input_shapes=(), input_dtypes=())
_validate(case)
assert case.input_distribution == ()
def test_output_dist_flat(self):
case = _make_testcase(output_shapes=((8,),))
_validate(case)
assert case.output_distribution == (0,)
def test_output_dist_nested(self):
case = _make_testcase(
output_shapes=(((8,), (8,)),),
output_dtypes=(("float16", "float16"),),
)
_validate(case)
assert case.output_distribution == (2,)
def test_output_dist_string_euclidean(self):
"""ELEWISE resolves before output_distribution is computed."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float16"),
output_shapes="ELEWISE",
output_dtypes=("float16",),
)
_validate(case)
assert isinstance(case.output_shapes, tuple)
assert case.output_distribution == (0,)
def test_combined_dist(self):
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float16"),
output_shapes=(((8,), (8,)),),
output_dtypes=(("float16", "float16"),),
)
_validate(case)
assert case.tensor_list_distribution == (2, 0, 2)
class TestNormalizeFieldByDist:
def test_compressed_broadcast(self):
"""('float16',) broadcasts to match distribution."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=("float16",),
output_shapes=((8,),),
output_dtypes=("float16",),
)
_validate(case)
assert case.input_dtypes == (("float16", "float16"), "float16")
def test_compressed_single_float(self):
"""Single float absolute_precision broadcasts."""
case = _make_testcase()
case.absolute_precision = 1e-5
_validate(case)
assert case.absolute_precision == (1e-05,)
def test_compressed_output_dtypes(self):
"""('float16',) output_dtypes broadcasts with output dist."""
case = _make_testcase(
output_shapes=(((8,), (8,)),),
output_dtypes=("float16",),
)
_validate(case)
assert case.output_dtypes == (("float16", "float16"),)
def test_already_nested_skip(self):
"""Already nested field is not modified."""
dtypes = (("float16", "float16"), "float16")
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=dtypes,
)
_validate(case)
assert case.input_dtypes == dtypes
def test_length_mismatch_invalid(self):
"""len(field) > len(dist) → invalid."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=("float16", "float32", "int32"),
)
_validate(case)
assert case.is_valid is False
def test_empty_field_skip(self):
"""Empty field is skipped, no normalize applied."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=("float16", "float16"),
input_formats=(),
)
_validate(case)
assert case.input_formats == ()
class TestNormalizeShapeField:
def test_compressed_shape_broadcast(self):
"""Single shape broadcasts to all positions."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_ori_shapes=((3, 4),),
)
_validate(case)
assert case.input_ori_shapes == (((3, 4), (3, 4)), (3, 4))
def test_already_nested_shape_skip(self):
"""Already nested shape field is not modified."""
ori = (((3, 4), (5, 4)), (8,))
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_ori_shapes=ori,
)
_validate(case)
assert case.input_ori_shapes == ori
def test_none_element_in_shape(self):
"""Mixed None and shape elements."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_ori_shapes=(None, None),
)
_validate(case)
assert case.is_valid
class TestNormalizeRangeField:
def test_compressed_range_broadcast(self):
"""Single range broadcasts to all tensors in TensorList."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_data_ranges=((0, 1),),
)
_validate(case)
assert case.input_data_ranges == (((0, 1), (0, 1)), (0, 1))
def test_already_nested_range_skip(self):
"""Already nested ranges are not modified."""
ranges = (((None, 1.0), (-1.0, 1.0)), (0.0, 5.0))
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float16"),
input_data_ranges=ranges,
)
_validate(case)
assert case.input_data_ranges == ranges
def test_precision_tolerances_compressed(self):
"""Compressed precision_tolerances broadcast."""
case = _make_testcase(
output_shapes=(((8,), (8,)),),
precision_tolerances=((0.004, 0.004),),
)
_validate(case)
assert case.precision_tolerances == (((0.004, 0.004), (0.004, 0.004)),)
def test_precision_tolerances_pair_not_destructed(self):
"""(rtol, ptol) pairs must not be destructed during normalize."""
case = _make_testcase(
output_shapes=(((8,), (8,)),),
precision_tolerances=(((0.001, 0.002), (0.003, 0.004)),),
)
_validate(case)
assert case.precision_tolerances == (((0.001, 0.002), (0.003, 0.004)),)
def test_non_tuple_field_skip(self):
"""Non-tuple field (e.g. float absolute_precision) is wrapped."""
case = _make_testcase()
case.precision_tolerances = None
_validate(case)
assert case.precision_tolerances is None
def test_bare_scalar_broadcast(self):
"""1e-5 bare scalar → broadcast to all positions."""
case = _make_testcase(
output_shapes=(((8,), (8,)),),
)
case.absolute_precision = 1e-5
_validate(case)
assert case.absolute_precision == ((1e-5, 1e-5),)
class TestFlatPropertiesAfterNormalize:
def test_flat_input_shapes_nested(self):
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)))
_validate(case)
assert case.flat_input_shapes == ((3, 4), (5, 4), (8,))
def test_flat_input_dtypes_compressed(self):
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=("float16",),
)
_validate(case)
assert case.flat_input_dtypes == ("float16", "float16", "float16")
def test_flat_input_dtypes_already_nested(self):
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float32"), "int32"),
)
_validate(case)
assert case.flat_input_dtypes == ("float16", "float32", "int32")
def test_flat_output_dtypes_compressed(self):
case = _make_testcase(
output_shapes=(((8,), (8,)),),
output_dtypes=("float16",),
)
_validate(case)
assert case.flat_output_dtypes == ("float16", "float16")
def test_flat_input_data_ranges_compressed(self):
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_data_ranges=((0, 1),),
)
_validate(case)
assert case.flat_input_data_ranges == ((0, 1), (0, 1), (0, 1))
def test_flat_input_data_ranges_nested(self):
ranges = (((None, 1.0), (-1.0, 1.0)), (0.0, 5.0))
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float16"),
input_data_ranges=ranges,
)
_validate(case)
assert case.flat_input_data_ranges == ((None, 1.0), (-1.0, 1.0), (0.0, 5.0))
def test_flat_precision_tolerances_nested(self):
case = _make_testcase(
output_shapes=(((8,), (8,)),),
precision_tolerances=(((0.001, 0.002), (0.003, 0.004)),),
)
_validate(case)
assert case.flat_precision_tolerances == ((0.001, 0.002), (0.003, 0.004))
def test_flat_absolute_precision_single_float(self):
case = _make_testcase()
case.absolute_precision = 1e-5
_validate(case)
assert case.flat_absolute_precision == (1e-05,)
def test_flat_absolute_precision_tuple(self):
case = _make_testcase(output_shapes=((8,), (9,)))
case.absolute_precision = (1e-5, 1e-6)
_validate(case)
assert case.flat_absolute_precision == (1e-5, 1e-6)
def test_flat_absolute_precision_per_output_no_double_wrap(self):
"""Per-output tuple (e.g. from result CSV) must not be double-nested."""
case = _make_testcase(output_shapes=((8,), (9,), (10,)))
case.absolute_precision = (1e-5, 1e-5, 1e-5)
_validate(case)
assert case.flat_absolute_precision == (1e-5, 1e-5, 1e-5)
for val in case.flat_absolute_precision:
assert isinstance(val, float), f"Expected float, got {type(val)}: {val}"
def test_flat_input_ori_shapes_nested(self):
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_ori_shapes=(((3, 4), (5, 4)), (8,)),
)
_validate(case)
assert case.flat_input_ori_shapes == ((3, 4), (5, 4), (8,))
def test_flat_output_formats_compressed(self):
case = _make_testcase(
output_shapes=(((8,), (8,)),),
output_formats=("ND",),
)
_validate(case)
assert case.flat_output_formats == ("ND", "ND")
def test_no_dist_flat_identity(self):
"""Flat (non-nested) with no dist — properties return as-is."""
case = _make_testcase(input_shapes=((8,), (8,)))
_validate(case)
assert case.flat_input_shapes == ((8,), (8,))
assert case.flat_input_dtypes == ("float16", "float16")
class TestInferenceResolutionOrder:
def test_elewise_resolves_before_output_dist(self):
"""output_shapes='ELEWISE' resolved before output_distribution computed."""
case = _make_testcase(
input_shapes=((8,), (8,)),
input_dtypes=("float16", "float16"),
output_shapes="ELEWISE",
output_dtypes=("float16",),
)
_validate(case)
assert isinstance(case.output_shapes, tuple)
assert len(case.output_shapes) == 1
assert case.output_distribution == (0,)
def test_elewise_output_dtypes_normalized(self):
"""After ELEWISE resolves, output_dtypes gets normalized."""
case = _make_testcase(
input_shapes=((8,), (8,)),
input_dtypes=("float16", "float16"),
output_shapes="ELEWISE",
output_dtypes=("float16",),
)
_validate(case)
assert case.output_dtypes == ("float16",)
def test_reduce_resolves(self):
"""REDUCE resolves correctly."""
case = _make_testcase(
input_shapes=((8, 4),),
input_dtypes=("float16",),
output_shapes="REDUCE",
output_dtypes=("float16",),
attributes={"axis": (1,)},
)
_validate(case)
assert isinstance(case.output_shapes, tuple)
assert case.output_distribution == (0,)
class TestNormalizeEdgeCases:
def test_empty_input_shapes(self):
case = _make_testcase(input_shapes=(), input_dtypes=(), input_ori_shapes=())
_validate(case)
assert case.input_distribution == ()
def test_output_shapes_none(self):
case = _make_testcase()
case.output_shapes = None
_validate(case)
assert case.is_valid is False
def test_multiple_tensorlists(self):
"""Multiple TensorList positions in input."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), ((8,), (9,), (10,))),
input_dtypes=("float16",),
)
_validate(case)
assert case.input_distribution == (2, 3)
assert case.input_dtypes == (("float16", "float16"), ("float16", "float16", "float16"))
assert case.flat_input_dtypes == ("float16", "float16", "float16", "float16", "float16")
def test_short_field_no_pad_invalid(self):
"""input_dtypes has no pad value — fewer values than dist → invalid."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,), (9,)),
input_dtypes=("float16", "float32"),
)
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "CASE_FIELD_AMBIGUOUS"
def test_precision_tolerances_flat_no_output_dist(self):
"""precision_tolerances with no output_distribution stays as-is."""
case = _make_testcase(output_shapes=((8,),))
case.precision_tolerances = ((0.004, 0.004),)
_validate(case)
assert case.precision_tolerances == ((0.004, 0.004),)
def test_all_none_fields(self):
"""Optional fields are None/empty — no crash during normalize."""
case = _make_testcase()
case.input_ori_shapes = None
case.input_data_ranges = None
case.precision_tolerances = None
case.absolute_precision = None
_validate(case)
assert case.is_valid
class TestPerTensorListCompressed:
"""TensorList position has compressed (len-1) value that needs broadcast."""
def test_input_dtype_per_tl_compressed(self):
"""('float32',) at TensorList(2) position → ('float32','float32')."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16",), "float32"),
)
_validate(case)
assert case.input_dtypes == (("float16", "float16"), "float32")
assert case.flat_input_dtypes == ("float16", "float16", "float32")
def test_input_format_per_tl_compressed(self):
"""('ND',) at TensorList(2) position → ('ND','ND')."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_formats=(("ND",), "NZ"),
)
_validate(case)
assert case.input_formats == (("ND", "ND"), "NZ")
def test_output_dtype_per_tl_compressed(self):
"""('float16',) at output TensorList(2) position → ('float16','float16')."""
case = _make_testcase(
output_shapes=(((8,), (8,)),),
output_dtypes=(("float16",),),
)
_validate(case)
assert case.output_dtypes == (("float16", "float16"),)
assert case.flat_output_dtypes == ("float16", "float16")
def test_mixed_compressed_and_expanded(self):
"""One position compressed, another fully expanded."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float32"), "int32"),
)
_validate(case)
assert case.input_dtypes == (("float16", "float32"), "int32")
def test_multiple_tensorlists_both_compressed(self):
"""Two TensorList positions, both with compressed dtype."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), ((8,), (9,), (10,))),
input_dtypes=(("float16",), ("int32",)),
)
_validate(case)
assert case.input_dtypes == (("float16", "float16"), ("int32", "int32", "int32"))
assert case.flat_input_dtypes == ("float16", "float16", "int32", "int32", "int32")
def test_scalar_at_tensorlist_position(self):
"""Scalar value at TensorList position broadcast to match count."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=("float16", "float32"),
)
_validate(case)
assert case.input_dtypes == (("float16", "float16"), "float32")
class TestNonTensorListMultiOutputCompressed:
"""Flat (non-nested) outputs with compressed single dtype/value."""
def test_dtype_broadcast_5_outputs(self):
"""('uint16',) with 5 flat outputs → 5-element tuple."""
case = _make_testcase(
output_shapes=((8,), (8,), (8,), (8,), (8,)),
output_dtypes=("uint16",),
)
_validate(case)
assert case.output_dtypes == ("uint16", "uint16", "uint16", "uint16", "uint16")
assert case.flat_output_dtypes == ("uint16", "uint16", "uint16", "uint16", "uint16")
def test_format_broadcast_3_outputs(self):
"""('ND',) with 3 flat outputs → 3-element tuple."""
case = _make_testcase(
output_shapes=((8,), (16,), (32,)),
output_formats=("ND",),
)
_validate(case)
assert case.output_formats == ("ND", "ND", "ND")
def test_single_output_no_broadcast(self):
"""Single output with single dtype — no broadcast needed."""
case = _make_testcase(
output_shapes=((8, 1716),),
output_dtypes=("int32",),
)
_validate(case)
assert case.output_dtypes == ("int32",)
def test_per_param_compressed_3_outputs_invalid(self):
"""output_dtypes has no pad value — 2 dtypes for 3 outputs → invalid."""
case = _make_testcase(
output_shapes=((8,), (16,), (32,)),
output_dtypes=("float16", "float32"),
)
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "CASE_FIELD_AMBIGUOUS"
def test_input_dtype_broadcast_flat(self):
"""('float32',) with 3 flat inputs → 3-element tuple."""
case = _make_testcase(
input_shapes=((8,), (16,), (32,)),
input_dtypes=("float32",),
)
_validate(case)
assert case.input_dtypes == ("float32", "float32", "float32")
class TestIsFieldAlreadyNested:
"""Tests for already-nested detection via _normalize_field_by_dist."""
def _assert_already_nested_scalar(self, field, dist):
"""Verify scalar field is detected as already-nested (no modification)."""
case = TestcaseOp()
case.is_valid = True
case.input_dtypes = field
case._normalize_field_by_dist("input_dtypes", dist, TestcaseOp._is_scalar_group)
assert case.is_valid is True
assert case.input_dtypes == field
def _assert_not_already_nested_scalar(self, field, dist):
"""Verify scalar field is NOT already-nested (gets normalized or rejected)."""
case = TestcaseOp()
case.is_valid = True
case.input_dtypes = field
case._normalize_field_by_dist("input_dtypes", dist, TestcaseOp._is_scalar_group)
assert case.input_dtypes != field or case.is_valid is False
def _assert_already_nested_range(self, field, dist):
"""Verify range field is detected as already-nested (no modification)."""
case = TestcaseOp()
case.is_valid = True
case.input_data_ranges = field
case._normalize_field_by_dist("input_data_ranges", dist, TestcaseOp._is_range_group)
assert case.is_valid is True
assert case.input_data_ranges == field
def _assert_not_already_nested_range(self, field, dist):
"""Verify range field is NOT already-nested (gets normalized or rejected)."""
case = TestcaseOp()
case.is_valid = True
case.input_data_ranges = field
case._normalize_field_by_dist("input_data_ranges", dist, TestcaseOp._is_range_group)
assert case.input_data_ranges != field or case.is_valid is False
def test_fully_nested_matches_dist(self):
self._assert_already_nested_scalar(
(("float16", "float16"), "float16"), (2, 0))
def test_compressed_not_nested(self):
self._assert_not_already_nested_scalar(("float16",), (2, 0))
def test_len_mismatch(self):
self._assert_not_already_nested_scalar(("float16", "float16"), (3,))
def test_inner_len_mismatch(self):
"""Per-TensorList compressed: inner len != dist count → not already nested."""
self._assert_not_already_nested_scalar(
(("float16",), "float16"), (2, 0))
def test_all_single_tensors(self):
"""Flat (no TensorList) — all-zero dist, matching length → already nested."""
self._assert_already_nested_scalar(("float16", "float16"), (0, 0))
def test_single_tensorlist(self):
self._assert_already_nested_scalar((("a", "b", "c"),), (3,))
def test_range_fully_nested(self):
"""Range field already nested."""
self._assert_already_nested_range(
(((0.0, 1.0), (0.5, 1.0)), (0.0, 5.0)), (2, 0))
def test_range_non_group_tuple_not_nested(self):
"""(0.0, 1.0) at dist=2 position is not a range group → not already nested."""
self._assert_not_already_nested_range(
((0.0, 1.0), (0.0, 5.0)), (2, 0))
class TestFlattenByDistribution:
"""Tests for _flatten_by_distribution static method."""
def test_flat_values_flat_dist(self):
result = TestcaseOp._flatten_by_distribution(("a", "b"), (0, 0))
assert result == ("a", "b")
def test_expand_tuple_list(self):
result = TestcaseOp._flatten_by_distribution(("a", ("b", "c")), (0, 2))
assert result == ("a", "b", "c")
def test_broadcast_single_in_list(self):
"""('b',) at dist=2 position → broadcast to ('b','b')."""
result = TestcaseOp._flatten_by_distribution(("a", ("b",)), (0, 2))
assert result == ("a", "b", "b")
def test_scalar_to_tensorlist(self):
"""Scalar at dist=2 position → broadcast to 2 copies."""
result = TestcaseOp._flatten_by_distribution(("a", "b"), (2, 0))
assert result == ("a", "a", "b")
def test_mixed(self):
result = TestcaseOp._flatten_by_distribution(
(("x", "y"), "z"), (2, 0))
assert result == ("x", "y", "z")
class TestNormalizeIdempotent:
"""Verify that normalize on already-normalized fields is a no-op."""
def test_fully_nested_dtypes_not_double_expanded(self):
dtypes = (("float16", "float16"), "float16")
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=dtypes,
)
_validate(case)
assert case.input_dtypes == dtypes
def test_normalized_then_normalize_again(self):
"""Calling _normalize_compressed_fields twice produces same result."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=("float16",),
)
_validate(case)
first = case.input_dtypes
case._normalize_compressed_fields()
assert case.input_dtypes == first
class TestAppendOutputMetadata:
"""Tests for append_output_metadata: encapsulates metadata append with cache consistency."""
def test_flat_output_no_dist(self):
"""1 output, no TensorList → all flat properties include appended metadata."""
case = _make_testcase(
output_shapes=((8, 1716),),
output_dtypes=("int32",),
)
_validate(case)
assert case.flat_output_dtypes == ("int32",)
case.append_output_metadata("uint64", "ND", (9,))
assert case.output_dtypes == ("int32", "uint64")
assert case.flat_output_dtypes == ("int32", "uint64")
assert case.flat_output_formats[-1] == "ND"
assert case.flat_output_shapes[-1] == (9,)
def test_flat_5_outputs_no_dist(self):
"""5 outputs, no TensorList → flat properties include appended metadata."""
case = _make_testcase(
output_shapes=((8,), (8,), (8,), (8,), (8,)),
output_dtypes=("uint16",),
)
_validate(case)
case.append_output_metadata("uint64", "ND", (9,))
assert case.flat_output_dtypes == ("uint16",) * 5 + ("uint64",)
def test_tensorlist_output(self):
"""TensorList output → flat properties flatten then append metadata."""
case = _make_testcase(
output_shapes=(((8,), (8,)),),
output_dtypes=(("float16",),),
)
_validate(case)
assert case.flat_output_dtypes == ("float16", "float16")
case.append_output_metadata("uint64", "ND", (9,))
assert case.output_dtypes == (("float16", "float16"), "uint64")
assert case.flat_output_dtypes == ("float16", "float16", "uint64")
assert case.flat_output_shapes == ((8,), (8,), (9,))
def test_mixed_tensorlist_and_flat(self):
"""TensorList + flat output → flat correct after append."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
output_shapes=(((8,), (8,)), (4,)),
output_dtypes=("float16",),
)
_validate(case)
case.append_output_metadata("uint64", "ND", (9,))
assert case.flat_output_dtypes == ("float16", "float16", "float16", "uint64")
def test_cache_hit_refreshed(self):
"""flat properties accessed before append, then after — cache refreshed."""
case = _make_testcase(
output_shapes=((8,),),
output_dtypes=("int32",),
)
_validate(case)
_ = case.flat_output_dtypes
case.append_output_metadata("uint64", "ND", (9,))
assert case.flat_output_dtypes == ("int32", "uint64")
class TestNormalizeManualBinaries:
def test_empty_not_modified(self):
case = _make_testcase()
case.manual_input_binaries = ()
_validate(case)
assert case.manual_input_binaries == ()
def test_none_not_modified(self):
case = _make_testcase()
case.manual_input_binaries = None
_validate(case)
assert case.manual_input_binaries is None
def test_single_string_wrapped(self):
case = _make_testcase(input_shapes=((8,),),
input_dtypes=("float16",))
case.manual_input_binaries = 'ax.csv'
_validate(case)
assert case.manual_input_binaries == ('ax.csv',)
def test_flat_tuple_preserved(self):
case = _make_testcase(input_shapes=((8,), (8,), (8,)),
input_dtypes=("float16", "float16", "float16"))
case.manual_input_binaries = ('ax.csv', 'ay.csv', 'az.csv')
_validate(case)
assert case.manual_input_binaries == ('ax.csv', 'ay.csv', 'az.csv')
def test_none_unquoted_preserved(self):
case = _make_testcase(input_shapes=((8,), None, (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = ('ax.csv', None, 'az.csv')
_validate(case)
assert case.manual_input_binaries == ('ax.csv', None, 'az.csv')
def test_none_quoted_converted(self):
case = _make_testcase(input_shapes=((8,), None, (8,)),
input_dtypes=("float16", "float16", "float32"))
case.manual_input_binaries = ('ax.csv', 'None', 'az.csv')
_validate(case)
assert case.manual_input_binaries == ('ax.csv', None, 'az.csv')
def test_nested_tuple_preserved(self):
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('ax.csv', 'ay.csv'), 'az.csv')
_validate(case)
assert case.manual_input_binaries == (('ax.csv', 'ay.csv'), 'az.csv')
def test_nested_with_none_quoted(self):
case = _make_testcase(input_shapes=(((3, 4), None), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('ax.csv', 'None'), 'az.csv')
_validate(case)
assert case.manual_input_binaries == (('ax.csv', None), 'az.csv')
def test_nested_with_none_unquoted(self):
case = _make_testcase(input_shapes=(((3, 4), None), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('ax.csv', None), 'az.csv')
_validate(case)
assert case.manual_input_binaries == (('ax.csv', None), 'az.csv')
def test_list_converted_to_tuple(self):
case = _make_testcase()
case.manual_input_binaries = ['ax.csv', 'ay.csv']
_validate(case)
assert case.manual_input_binaries == ('ax.csv', 'ay.csv')
assert isinstance(case.manual_input_binaries, tuple)
def test_nested_list_converted_to_tuple(self):
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (['ax.csv', 'ay.csv'], 'az.csv')
_validate(case)
assert case.manual_input_binaries == (('ax.csv', 'ay.csv'), 'az.csv')
assert isinstance(case.manual_input_binaries[0], tuple)
def test_nested_rejected_without_tensor_list(self):
case = _make_testcase(input_shapes=((8,), (8,)),
input_dtypes=("float16", "float16"))
case.manual_input_binaries = (('ax.csv', 'ay.csv'), 'az.csv')
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_mixed_nested_flat_rejected(self):
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = ('ax.csv', ('az.csv',))
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_flat_count_exceeds_inputs_rejected(self):
case = _make_testcase(input_shapes=((8,), (8,)),
input_dtypes=("float16", "float16"))
case.manual_input_binaries = ('ax.csv', 'ay.csv', 'az.csv')
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_flat_with_none_input_explicit_none_ok(self):
case = _make_testcase(input_shapes=((8,), None, (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = ('ax.csv', None, 'az.csv')
_validate(case)
assert case.manual_input_binaries == ('ax.csv', None, 'az.csv')
def test_flat_missing_required_input_rejected(self):
case = _make_testcase(input_shapes=((8,), (8,), (8,)),
input_dtypes=("float16", "float16", "float16"))
case.manual_input_binaries = ('ax.csv', 'ay.csv')
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_flat_file_for_none_input_rejected(self):
case = _make_testcase(input_shapes=((8,), None, (8,)),
input_dtypes=("float16", "float16", "float32"))
case.manual_input_binaries = ('ax.csv', 'unexpected.csv', 'az.csv')
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_nested_group_count_exceeds_inputs_rejected(self):
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('ax.csv', 'ay.csv', 'extra.csv'), 'az.csv')
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_nested_with_optional_input_none_ok(self):
case = _make_testcase(input_shapes=(((3, 4), None), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('ax.csv', None), 'az.csv')
_validate(case)
assert case.manual_input_binaries == (('ax.csv', None), 'az.csv')
def test_nested_missing_required_input_rejected(self):
"""TensorList group has 2 non-None inputs but only 1 file."""
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('ax.csv',), 'az.csv')
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_flat_input_reshaped_to_nested(self):
"""Flat binaries for TensorList inputs → reshaped to match input_shapes."""
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = ('ax.csv', 'ay.csv', 'az.csv')
_validate(case)
assert case.manual_input_binaries == (('ax.csv', 'ay.csv'), 'az.csv')
def test_flat_with_none_padded_to_nested(self):
"""Flat binaries with None → reshaped and padded to match input_shapes."""
case = _make_testcase(input_shapes=(((3, 4), None), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = ('ax.csv', None, 'az.csv')
_validate(case)
assert case.manual_input_binaries == (('ax.csv', None), 'az.csv')
def test_no_tensor_list_stays_flat(self):
"""No TensorList → binaries stays flat tuple."""
case = _make_testcase(input_shapes=((8,), (8,)),
input_dtypes=("float16", "float16"))
case.manual_input_binaries = ('ax.csv', 'ay.csv')
_validate(case)
assert case.manual_input_binaries == ('ax.csv', 'ay.csv')
def test_single_input_stays_single(self):
case = _make_testcase(input_shapes=((8,),),
input_dtypes=("float16",))
case.manual_input_binaries = 'ax.csv'
_validate(case)
assert case.manual_input_binaries == ('ax.csv',)
def test_flat_all_none_inputs_all_none_binaries(self):
case = _make_testcase(input_shapes=(None, None),
input_dtypes=("float16", "float16"))
case.manual_input_binaries = (None, None)
_validate(case)
assert case.manual_input_binaries == (None, None)
def test_flat_trailing_none_input(self):
case = _make_testcase(input_shapes=((8,), (8,), None),
input_dtypes=("float16", "float16", "float16"))
case.manual_input_binaries = ('ax.csv', 'ay.csv', None)
_validate(case)
assert case.manual_input_binaries == ('ax.csv', 'ay.csv', None)
def test_tensor_list_group_none_mixed_nested(self):
"""TensorList with one None and one non-None input."""
case = _make_testcase(input_shapes=(((3, 4), None), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('ax.csv', None), 'az.csv')
_validate(case)
assert case.manual_input_binaries == (('ax.csv', None), 'az.csv')
def test_flat_binaries_exceed_non_tensor_list_inputs_rejected(self):
"""3 flat binaries for 2 non-TensorList inputs (None, None not recognized as TensorList)."""
case = _make_testcase(input_shapes=((None, None), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (None, None, 'az.csv')
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_trailing_none_inputs_implicit_none_binaries(self):
"""Binaries shorter because trailing inputs are all None → padded."""
case = _make_testcase(input_shapes=((8,), None, None),
input_dtypes=("float16", "float16", "float16"))
case.manual_input_binaries = ('ax.csv',)
_validate(case)
assert case.manual_input_binaries == ('ax.csv', None, None)
def test_nested_top_level_count_mismatch_rejected(self):
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('ax.csv', 'ay.csv'),)
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_nested_tensor_list_position_is_str_rejected(self):
"""Nested format but TensorList position is str instead of tuple."""
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = ('ax.csv', 'az.csv')
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_nested_non_tensor_list_position_is_tuple_rejected(self):
"""Non-TensorList position given as tuple."""
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('ax.csv', 'ay.csv'), ('az.csv',))
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_invalid_type_rejected(self):
case = _make_testcase(input_shapes=((8,),),
input_dtypes=("float16",))
case.manual_input_binaries = 123
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_file_for_none_in_tensor_list_group_rejected(self):
case = _make_testcase(input_shapes=(((3, 4), None), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('ax.csv', 'unexpected.csv'), 'az.csv')
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
def test_missing_file_for_non_none_in_tensor_list_group_rejected(self):
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('ax.csv', None), 'az.csv')
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "MANUAL_BINARIES_INVALID"
class TestFlatManualInputBinaries:
"""Tests for flat_manual_input_binaries — uses deep_flatten."""
def test_flat_no_dist(self):
case = _make_testcase(input_shapes=((8,), (8,)),
input_dtypes=("float16", "float16"))
case.manual_input_binaries = ('a.csv', 'b.csv')
_validate(case)
assert case.flat_manual_input_binaries == ('a.csv', 'b.csv')
def test_nested_tensorlist(self):
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('a.csv', 'b.csv'), 'c.csv')
_validate(case)
assert case.flat_manual_input_binaries == ('a.csv', 'b.csv', 'c.csv')
def test_nested_with_none(self):
case = _make_testcase(input_shapes=(((3, 4), None), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = (('a.csv', None), 'c.csv')
_validate(case)
assert case.flat_manual_input_binaries == ('a.csv', None, 'c.csv')
def test_none_when_not_set(self):
case = _make_testcase()
_validate(case)
assert case.flat_manual_input_binaries is None
def test_empty_tuple(self):
case = _make_testcase()
case.manual_input_binaries = ()
_validate(case)
assert case.flat_manual_input_binaries == ()
def test_flat_reshaped_to_nested_then_flat(self):
"""Flat binaries → reshape to nested → flat property returns flat."""
case = _make_testcase(input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float32"))
case.manual_input_binaries = ('a.csv', 'b.csv', 'c.csv')
_validate(case)
assert case.manual_input_binaries == (('a.csv', 'b.csv'), 'c.csv')
assert case.flat_manual_input_binaries == ('a.csv', 'b.csv', 'c.csv')
class TestIsScalarGroup:
"""Tests for _is_scalar_group classmethod."""
def test_tuple_is_group(self):
assert TestcaseOp._is_scalar_group(('NCHW', 'NHWC')) is True
def test_list_is_group(self):
assert TestcaseOp._is_scalar_group(['NCHW', 'NHWC']) is True
def test_string_is_not_group(self):
assert TestcaseOp._is_scalar_group('NCHW') is False
def test_int_is_not_group(self):
assert TestcaseOp._is_scalar_group(42) is False
def test_float_is_not_group(self):
assert TestcaseOp._is_scalar_group(1e-5) is False
def test_empty_tuple_is_not_group(self):
assert TestcaseOp._is_scalar_group(()) is True
class TestIsRangeGroup:
"""Tests for _is_range_group classmethod."""
def test_range_list_is_group(self):
assert TestcaseOp._is_range_group(((0.0, 1.0), (0.5, 1.0))) is True
def test_single_range_is_not_group(self):
"""A bare range (0.0, 1.0) is NOT a group — it's a single atomic value."""
assert TestcaseOp._is_range_group((0.0, 1.0)) is False
def test_string_tuple_is_not_group(self):
assert TestcaseOp._is_range_group(('NCHW', 'NHWC')) is False
def test_empty_tuple_is_not_group(self):
assert TestcaseOp._is_range_group(()) is False
def test_single_element_range_list(self):
"""((0.0, 1.0),) is a 1-element range group."""
assert TestcaseOp._is_range_group(((0.0, 1.0),)) is True
def test_int_tuple_is_not_group(self):
"""(1, 2) looks like a range but val[0]=1 is not tuple → not a group."""
assert TestcaseOp._is_range_group((1, 2)) is False
class TestNormalizeFieldAlreadyNested:
"""Already-nested detection via unified _normalize_field_by_dist."""
def test_scalar_fully_nested_skip(self):
"""Already nested scalar field — no modification."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float32"), "int32"),
)
_validate(case)
assert case.input_dtypes == (("float16", "float32"), "int32")
def test_range_fully_nested_skip(self):
"""Already nested range field — no modification."""
ranges = (((None, 1.0), (-1.0, 1.0)), (0.0, 5.0))
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float16"), "float16"),
input_data_ranges=ranges,
)
_validate(case)
assert case.input_data_ranges == ranges
def test_all_flat_dist_zero_already_nested(self):
"""Flat (no TensorList) with matching length — already nested."""
case = _make_testcase(
input_shapes=((8,), (8,)),
input_dtypes=("float16", "float32"),
)
_validate(case)
assert case.input_dtypes == ("float16", "float32")
class TestNormalizeFieldLenEqOne:
"""len(field)==1 branch: unwrap, reject, broadcast."""
def test_scalar_compressed_broadcast(self):
"""('float16',) broadcasts to match distribution."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=("float16",),
)
_validate(case)
assert case.input_dtypes == (("float16", "float16"), "float16")
def test_scalar_single_element_group_unwrap(self):
"""(('NCHW',),) → unwrap to 'NCHW' → broadcast."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_formats=(("ND",),),
)
_validate(case)
assert case.input_formats == (("ND", "ND"), "ND")
def test_scalar_multi_element_group_rejected(self):
"""Bug fix: (('NCHW','NHWC'),) with dist=(2,0) → REJECT.
Old code would broadcast ('NCHW','NHWC') to position 1 (num=0),
producing a tuple where a single string is expected. Now correctly
rejected as CASE_FIELD_AMBIGUOUS.
"""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
)
case.input_dtypes = (("float16", "float32"),)
case._normalize_field_by_dist(
"input_dtypes", (2, 0), TestcaseOp._is_scalar_group)
assert case.is_valid is False
assert case.fail_reason == "CASE_FIELD_AMBIGUOUS"
def test_range_compressed_broadcast(self):
"""Single range ((0,1),) broadcasts to all positions."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_data_ranges=((0, 1),),
)
_validate(case)
assert case.input_data_ranges == (((0, 1), (0, 1)), (0, 1))
def test_range_single_element_group_unwrap(self):
"""(((0.0,1.0),),) → unwrap to (0.0,1.0) → broadcast."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
)
case.input_data_ranges = (((0.0, 1.0),),)
case._normalize_field_by_dist(
"input_data_ranges", (2, 0), TestcaseOp._is_range_group)
assert case.input_data_ranges == (((0.0, 1.0), (0.0, 1.0)), (0.0, 1.0))
def test_range_multi_element_group_rejected(self):
"""(((0.0,1.0),(0.5,1.0)),) with dist=(2,0) → REJECT."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
)
case.input_data_ranges = (((0.0, 1.0), (0.5, 1.0)),)
case._normalize_field_by_dist(
"input_data_ranges", (2, 0), TestcaseOp._is_range_group)
assert case.is_valid is False
assert case.fail_reason == "CASE_FIELD_AMBIGUOUS"
class TestNormalizeFieldLenBetweenOneAndDist:
"""1 < len(field) < len(dist): pad or reject."""
def test_scalar_short_field_no_pad_rejected(self):
"""input_dtypes has no pad — fewer values than dist → invalid."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,), (9,)),
input_dtypes=("float16", "float32"),
)
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "CASE_FIELD_AMBIGUOUS"
def test_range_short_field_with_pad(self):
"""1 < len(field) < len(dist): pad missing positions with pad_value."""
case = _make_testcase(
output_shapes=(((8,), (8,)), (4,), (6,)),
)
case.precision_tolerances = ((0.001, 0.002), (0.003, 0.004))
case._normalize_field_by_dist(
"precision_tolerances", (2, 0, 0), TestcaseOp._is_range_group,
pad_value=None)
assert case.precision_tolerances == (
((0.001, 0.002), (0.001, 0.002)), (0.003, 0.004), None)
class TestNormalizeFieldLenGtDist:
"""len(field) > len(dist) → REJECT."""
def test_scalar_len_gt_dist(self):
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=("float16", "float32", "int32"),
)
_validate(case)
assert case.is_valid is False
assert case.fail_reason == "CASE_FIELD_AMBIGUOUS"
def test_range_len_gt_dist(self):
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
)
case.input_data_ranges = ((0, 1), (-1, 1), (0, 5))
case._normalize_field_by_dist(
"input_data_ranges", (2, 0), TestcaseOp._is_range_group)
assert case.is_valid is False
assert case.fail_reason == "CASE_FIELD_AMBIGUOUS"
class TestNormalizeFieldPerPositionExpand:
"""Per-position expansion after broadcast/pad."""
def test_scalar_group_matches_dist(self):
"""Group at TensorList position with matching len → use as-is."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16", "float32"), "int32"),
)
_validate(case)
assert case.input_dtypes == (("float16", "float32"), "int32")
def test_scalar_group_single_broadcast(self):
"""('float16',) at TensorList(2) → broadcast to ('float16','float16')."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=(("float16",), "float32"),
)
_validate(case)
assert case.input_dtypes == (("float16", "float16"), "float32")
def test_scalar_bare_value_broadcast(self):
"""'float16' at TensorList(2) → broadcast to ('float16','float16')."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_dtypes=("float16", "float32"),
)
_validate(case)
assert case.input_dtypes == (("float16", "float16"), "float32")
def test_range_group_matches_dist(self):
"""Range group at TensorList position with matching len → use as-is."""
case = _make_testcase(
output_shapes=(((8,), (8,)),),
precision_tolerances=(((0.001, 0.002), (0.003, 0.004)),),
)
_validate(case)
assert case.precision_tolerances == (((0.001, 0.002), (0.003, 0.004)),)
def test_range_group_single_broadcast(self):
"""((0.004,0.004),) at TensorList(2) → broadcast."""
case = _make_testcase(
output_shapes=(((8,), (8,)),),
precision_tolerances=(((0.004, 0.004),),),
)
_validate(case)
assert case.precision_tolerances == (((0.004, 0.004), (0.004, 0.004)),)
def test_range_group_mismatch_rejected(self):
"""Range group with wrong len at TensorList position → REJECT."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
)
case.input_data_ranges = (((0, 1), (0, 1), (0, 1)), (0, 5))
case._normalize_field_by_dist(
"input_data_ranges", (2, 0), TestcaseOp._is_range_group)
assert case.is_valid is False
assert case.fail_reason == "CASE_FIELD_AMBIGUOUS"
class TestNormalizeFieldGuardChecks:
"""Entry guard tests."""
def test_not_is_valid_returns_early(self):
"""If is_valid is False, normalize does nothing."""
case = _make_testcase()
case.is_valid = False
case.input_dtypes = ("float16",)
original = case.input_dtypes
case._normalize_field_by_dist(
"input_dtypes", (2, 0), TestcaseOp._is_scalar_group)
assert case.input_dtypes == original
def test_empty_field_returns_early(self):
"""Empty field → skip normalize."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
input_formats=(),
)
_validate(case)
assert case.input_formats == ()
def test_none_field_returns_early(self):
"""None field → skip normalize."""
case = _make_testcase()
case.precision_tolerances = None
_validate(case)
assert case.precision_tolerances is None
class TestNormalizeFieldCallerPrewrap:
"""Caller is responsible for wrapping bare scalar values."""
def test_scalar_bare_string_wrapped_by_caller(self):
"""Bare 'ND' must be wrapped to ('ND',) by caller before calling."""
case = _make_testcase(
input_shapes=(((3, 4), (5, 4)), (8,)),
)
case.input_dtypes = ("float16",)
case._normalize_field_by_dist(
"input_dtypes", (2, 0), TestcaseOp._is_scalar_group)
assert case.input_dtypes == (("float16", "float16"), "float16")
