diff --git a/python/fast_mlsirm/testlet.py b/python/fast_mlsirm/testlet.py index a4b4281cf..58ce0316a 100644 --- a/python/fast_mlsirm/testlet.py +++ b/python/fast_mlsirm/testlet.py @@ -14,6 +14,19 @@ MAX_TESTLET_RESPONSE_CELLS = 20_000_000 _SUPPORTED_Q_GAMMA = (7, 11, 15, 21, 31, 41) +_NUMPY_INTEGER_TYPES = tuple( + np.dtype(name).type + for name in ("int8", "int16", "int32", "int64", "uint8", "uint16", "uint32", "uint64") +) +_NUMPY_FLOAT_TYPES = tuple( + np.dtype(name).type for name in ("float16", "float32", "float64", "longdouble") +) + + +def _is_exact_type(value_type: type, trusted_types: tuple[type, ...]) -> bool: + """Return whether ``value_type`` is one trusted type without invoking callbacks.""" + + return any(value_type is trusted_type for trusted_type in trusted_types) @dataclass @@ -115,33 +128,72 @@ def fit_testlet( raise ValueError("testlet_id entries must be between 0 and n_items - 1") tid = raw_tid.astype(np.int64, copy=False) n_testlets = int(tid.max()) + 1 - if ( - isinstance(max_iter, (bool, np.bool_)) - or not isinstance(max_iter, (int, np.integer)) - or not 1 <= int(max_iter) <= MAX_MAX_ITER - ): + + if type(model) is not str: + raise ValueError("model must be a built-in string") + if model not in ("rasch", "2pl"): + raise ValueError("model must be either 'rasch' or '2pl'") + model_value = model + + max_iter_type = type(max_iter) + if max_iter_type is int: + max_iter_value = max_iter + elif _is_exact_type(max_iter_type, _NUMPY_INTEGER_TYPES): + max_iter_value = int(max_iter) + else: + raise ValueError(f"max_iter must be an integer between 1 and {MAX_MAX_ITER}") + if not 1 <= max_iter_value <= MAX_MAX_ITER: raise ValueError(f"max_iter must be an integer between 1 and {MAX_MAX_ITER}") - if isinstance(tol, (bool, np.bool_)) or not isinstance( - tol, (int, float, np.integer, np.floating) + + tol_type = type(tol) + if not ( + tol_type is int + or tol_type is float + or _is_exact_type(tol_type, _NUMPY_INTEGER_TYPES) + or _is_exact_type(tol_type, _NUMPY_FLOAT_TYPES) ): raise ValueError("tol must be a finite non-negative number") - tol_value = float(tol) + try: + tol_value = float(tol) + except OverflowError as exc: + raise ValueError("tol must be a finite non-negative number") from exc if not np.isfinite(tol_value) or tol_value < 0.0: raise ValueError("tol must be a finite non-negative number") - if ( - isinstance(q_gamma, (bool, np.bool_)) - or not isinstance(q_gamma, (int, np.integer)) - or int(q_gamma) not in _SUPPORTED_Q_GAMMA - ): + + q_gamma_type = type(q_gamma) + if q_gamma_type is int: + q_gamma_value = q_gamma + elif _is_exact_type(q_gamma_type, _NUMPY_INTEGER_TYPES): + q_gamma_value = int(q_gamma) + else: + raise ValueError(f"q_gamma must be one of {_SUPPORTED_Q_GAMMA}") + if q_gamma_value not in _SUPPORTED_Q_GAMMA: raise ValueError(f"q_gamma must be one of {_SUPPORTED_Q_GAMMA}") - if isinstance(init_sigma2, (bool, np.bool_)) or not isinstance( - init_sigma2, (int, float, np.integer, np.floating) + + init_sigma2_type = type(init_sigma2) + if not ( + init_sigma2_type is int + or init_sigma2_type is float + or _is_exact_type(init_sigma2_type, _NUMPY_INTEGER_TYPES) + or _is_exact_type(init_sigma2_type, _NUMPY_FLOAT_TYPES) ): raise ValueError("init_sigma2 must be a finite non-negative number") - init_sigma2_value = float(init_sigma2) + try: + init_sigma2_value = float(init_sigma2) + except OverflowError as exc: + raise ValueError("init_sigma2 must be a finite non-negative number") from exc if not np.isfinite(init_sigma2_value) or init_sigma2_value < 0.0: raise ValueError("init_sigma2 must be a finite non-negative number") + estimate_sigma_type = type(estimate_sigma) + if estimate_sigma_type is not bool and estimate_sigma_type is not np.bool_: + raise ValueError("estimate_sigma must be a Boolean") + estimate_sigma_value = bool(estimate_sigma) + require_convergence_type = type(require_convergence) + if require_convergence_type is not bool and require_convergence_type is not np.bool_: + raise ValueError("require_convergence must be a Boolean") + require_convergence_value = bool(require_convergence) + from .fitstats import _core_module core = _core_module() @@ -157,11 +209,11 @@ def fit_testlet( int(n_persons), int(n_items), int(n_testlets), - str(model), - int(max_iter), + model_value, + max_iter_value, tol_value, - int(q_gamma), - bool(estimate_sigma), + q_gamma_value, + estimate_sigma_value, init_sigma2_value, ) fit = TestletFit( @@ -181,11 +233,11 @@ def fit_testlet( if not fit.converged: message = ( "testlet calibration did not converge: " - f"reason={fit.termination_reason}, iterations={fit.n_iter}/{max_iter}, " + f"reason={fit.termination_reason}, iterations={fit.n_iter}/{max_iter_value}, " "final_loglik_change=" f"{fit.final_loglik_change:.12g}, tolerance={tol_value:.12g}" ) - if require_convergence: + if require_convergence_value: raise RuntimeError(message) warnings.warn(message, RuntimeWarning, stacklevel=2) return fit diff --git a/tests/test_testlet_control_callbacks.py b/tests/test_testlet_control_callbacks.py new file mode 100644 index 000000000..24a038230 --- /dev/null +++ b/tests/test_testlet_control_callbacks.py @@ -0,0 +1,276 @@ +"""Fail-closed callback-boundary tests for testlet public controls.""" + +from __future__ import annotations + +from unittest.mock import patch + +import numpy as np +import pytest + +from fast_mlsirm.testlet import fit_testlet + + +class _HostileInt(int): + """Integer subclass that records forbidden conversion callbacks.""" + + calls = 0 + + def __int__(self): + type(self).calls += 1 + raise AssertionError("hostile __int__ executed") + + def __repr__(self): + type(self).calls += 1 + raise AssertionError("hostile __repr__ executed") + + +class _HostileNpInt(np.int64): + """NumPy integer subclass that records forbidden conversion callbacks.""" + + calls = 0 + + def __int__(self): + type(self).calls += 1 + raise AssertionError("hostile numpy __int__ executed") + + def __repr__(self): + type(self).calls += 1 + raise AssertionError("hostile numpy __repr__ executed") + + +class _HostileFloat(float): + """Float subclass that records forbidden conversion callbacks.""" + + calls = 0 + + def __float__(self): + type(self).calls += 1 + raise AssertionError("hostile __float__ executed") + + def __repr__(self): + type(self).calls += 1 + raise AssertionError("hostile __repr__ executed") + + +class _HostileNpFloat(np.float64): + """NumPy floating subclass that records forbidden conversion callbacks.""" + + calls = 0 + + def __float__(self): + type(self).calls += 1 + raise AssertionError("hostile numpy __float__ executed") + + def __repr__(self): + type(self).calls += 1 + raise AssertionError("hostile __repr__ executed") + + +class _HostileStr(str): + """String subclass that records forbidden conversion callbacks.""" + + calls = 0 + + def __str__(self): + type(self).calls += 1 + raise AssertionError("hostile __str__ executed") + + def __repr__(self): + type(self).calls += 1 + raise AssertionError("hostile __repr__ executed") + + +class _HostileBool: + """Boolean-like object that records forbidden truth-value callbacks.""" + + calls = 0 + + def __bool__(self): + type(self).calls += 1 + raise AssertionError("hostile __bool__ executed") + + def __repr__(self): + type(self).calls += 1 + raise AssertionError("hostile __repr__ executed") + + +def _binary() -> np.ndarray: + """Return a deterministic valid response matrix.""" + + return np.zeros((2, 2), dtype=np.float64) + + +def _tid() -> np.ndarray: + """Return a deterministic valid two-item testlet assignment.""" + + return np.array([0, 0], dtype=np.int64) + + +def _assert_rejected_without_callback(keyword: str, value: object, cls: type) -> None: + """Require rejection before caller conversion and before native discovery.""" + + cls.calls = 0 + controls: dict[str, object] = {"q_gamma": 7} + controls[keyword] = value + with patch( + "fast_mlsirm.fitstats._core_module", + side_effect=AssertionError("native core discovery must not run"), + ): + with pytest.raises(ValueError): + fit_testlet(_binary(), _tid(), **controls) + assert cls.calls == 0 + + +@pytest.mark.parametrize( + ("keyword", "constructor", "args"), + [ + ("max_iter", _HostileInt, (7,)), + ("max_iter", _HostileNpInt, (7,)), + ("q_gamma", _HostileInt, (7,)), + ("q_gamma", _HostileNpInt, (7,)), + ("tol", _HostileFloat, (1e-6,)), + ("tol", _HostileNpFloat, (1e-6,)), + ("init_sigma2", _HostileFloat, (0.5,)), + ("init_sigma2", _HostileNpFloat, (0.5,)), + ("model", _HostileStr, ("rasch",)), + ("estimate_sigma", _HostileBool, ()), + ("require_convergence", _HostileBool, ()), + ], +) +def test_fit_testlet_rejects_control_subclasses_before_callbacks( + keyword: str, + constructor: type, + args: tuple[object, ...], +) -> None: + """Rejected public controls must not execute caller conversion callbacks.""" + + value = constructor(*args) + _assert_rejected_without_callback(keyword, value, constructor) + + +@pytest.mark.parametrize( + ("keyword", "message"), + [ + ("tol", "tol must be a finite non-negative number"), + ("init_sigma2", "init_sigma2 must be a finite non-negative number"), + ], +) +def test_fit_testlet_normalizes_oversized_integer_float_controls( + keyword: str, + message: str, +) -> None: + """Trusted integers that cannot become floats fail with package-owned errors.""" + + with patch( + "fast_mlsirm.fitstats._core_module", + side_effect=AssertionError("native core discovery must not run"), + ): + with pytest.raises(ValueError, match=message): + fit_testlet(_binary(), _tid(), q_gamma=7, **{keyword: 10**10_000}) + + +def test_fit_testlet_rejects_hostile_scalar_metaclass_hash_before_callback() -> None: + """NumPy scalar admission must not hash a caller-controlled metaclass.""" + + calls: list[str] = [] + + class HostileMeta(type): + def __hash__(cls) -> int: + calls.append("type-__hash__") + raise AssertionError("type hash callback executed") + + class HostileNumpyInt(np.int64, metaclass=HostileMeta): + pass + + with patch( + "fast_mlsirm.fitstats._core_module", + side_effect=AssertionError("native core discovery must not run"), + ): + with pytest.raises(ValueError, match="max_iter must be an integer"): + fit_testlet(_binary(), _tid(), max_iter=HostileNumpyInt(3), q_gamma=7) + + assert calls == [] + + +def test_fit_testlet_rejects_hostile_scalar_metaclass_equality_before_callback() -> None: + """Built-in scalar admission must not compare caller types for equality.""" + + calls: list[str] = [] + + class HostileMeta(type): + __hash__ = type.__hash__ + + def __eq__(cls, other: object) -> bool: + calls.append("type-__eq__") + raise AssertionError("type equality callback executed") + + class HostileNumpyFloat(np.float64, metaclass=HostileMeta): + pass + + with patch( + "fast_mlsirm.fitstats._core_module", + side_effect=AssertionError("native core discovery must not run"), + ): + with pytest.raises(ValueError, match="tol must be a finite non-negative number"): + fit_testlet(_binary(), _tid(), tol=HostileNumpyFloat(1e-6), q_gamma=7) + + assert calls == [] + + +def test_fit_testlet_preserves_genuine_numpy_scalars() -> None: + """Exact NumPy scalar classes remain valid public controls.""" + + class _Core: + @staticmethod + def fit_testlet(*args): + assert args[6] == "rasch" + assert type(args[7]) is int + assert args[7] == 3 + assert type(args[8]) is float + assert args[8] == pytest.approx(1e-6) + assert type(args[9]) is int + assert args[9] == 7 + assert type(args[10]) is bool + assert args[10] is False + assert type(args[11]) is float + assert args[11] == pytest.approx(0.25) + return { + "model": "rasch", + "a": [1.0, 1.0], + "b": [0.0, 0.0], + "beta": [0.0, 0.0], + "sigma2": [0.0], + "theta": [0.0, 0.0], + "loglik_trace": [0.0], + "n_iter": 1, + "converged": True, + "n_parameters": 2, + "termination_reason": "tolerance", + "final_loglik_change": 0.0, + } + + with patch("fast_mlsirm.fitstats._core_module", return_value=_Core()): + fit = fit_testlet( + _binary(), + _tid(), + model="rasch", + max_iter=np.int64(3), + tol=np.float64(1e-6), + q_gamma=np.int64(7), + estimate_sigma=np.bool_(False), + init_sigma2=np.float64(0.25), + require_convergence=np.bool_(False), + ) + assert fit.converged + + +@pytest.mark.parametrize("model", ["", "invalid_model", "Rasch", "2PL"]) +def test_fit_testlet_rejects_unknown_builtin_model_before_native_core(model: str) -> None: + """Only the Rust-supported testlet model identifiers may cross the boundary.""" + + with patch( + "fast_mlsirm.fitstats._core_module", + side_effect=AssertionError("native core discovery must not run"), + ): + with pytest.raises(ValueError, match="model must be either 'rasch' or '2pl'"): + fit_testlet(_binary(), _tid(), model=model, q_gamma=7)