[CI] 【Hackathon 9th Sprint No.19】NO.19 功能模块单测补充 #5063
Conversation
|
Thanks for your contribution! |
There was a problem hiding this comment.
Pull Request Overview
This PR adds comprehensive unit test coverage for the fastdeploy/model_executor/layers/pooler.py module as part of Hackathon 9th Sprint No.19. The implementation uses an extensive mocking infrastructure to test the pooler functionality without requiring actual Paddle dependencies, achieving 89% code coverage.
Key changes:
- Added a complete test suite with 693 lines covering pooler functionality
- Implemented mock versions of Paddle tensor operations and neural network layers
- Created test cases for pooler helpers, activations, heads, pooling methods, and dispatching logic
|
|
||
| class _Logger: | ||
| def __init__(self): | ||
| self.messages: List[tuple[str, str]] = [] |
There was a problem hiding this comment.
The _Logger class uses Python 3.9+ syntax tuple[str, str] in the type annotation which may cause compatibility issues. Consider using List[Tuple[str, str]] from the typing module for broader Python version compatibility, especially since the rest of the file uses from typing import imports.
| class _FakeTensor: | ||
| __array_priority__ = 1000 | ||
|
|
||
| def __init__(self, array: Any, dtype: Optional[str] = None, place: Optional[_FakePlace] = None) -> None: | ||
| if isinstance(array, _FakeTensor): | ||
| array = array.array | ||
| if dtype is not None: | ||
| self.array = np.array(array, dtype=dtype) | ||
| else: | ||
| self.array = np.array(array) | ||
| self.place = place or _FakePlace(False) | ||
|
|
||
| def __repr__(self) -> str: # pragma: no cover - debug helper | ||
| return f"_FakeTensor({self.array!r})" | ||
|
|
||
| def __len__(self) -> int: | ||
| return len(self.array) | ||
|
|
||
| @property | ||
| def dtype(self): # pragma: no cover - compatibility helper | ||
| return self.array.dtype | ||
|
|
||
| @property | ||
| def shape(self): | ||
| return self.array.shape | ||
|
|
||
| def numpy(self): | ||
| return self.array | ||
|
|
||
| def tolist(self): | ||
| return self.array.tolist() | ||
|
|
||
| def item(self): | ||
| return self.array.item() | ||
|
|
||
| def astype(self, dtype: str) -> "_FakeTensor": | ||
| return _FakeTensor(self.array.astype(dtype), place=self.place) | ||
|
|
||
| def unsqueeze(self, axis: int) -> "_FakeTensor": | ||
| return _FakeTensor(np.expand_dims(self.array, axis=axis), place=self.place) | ||
|
|
||
| def split(self, lengths: List[int]): | ||
| outputs = [] | ||
| start = 0 | ||
| for length in lengths: | ||
| outputs.append(_FakeTensor(self.array[start : start + length], place=self.place)) | ||
| start += length | ||
| return outputs | ||
|
|
||
| def cuda(self) -> "_FakeTensor": | ||
| return _FakeTensor(self.array.copy(), place=_FakePlace(True)) | ||
|
|
||
| def __getitem__(self, item): | ||
| if isinstance(item, _FakeTensor): | ||
| item = item.array | ||
| result = self.array.__getitem__(item) | ||
| if isinstance(result, np.ndarray): | ||
| return _FakeTensor(result, place=self.place) | ||
| return result | ||
|
|
||
| def __setitem__(self, key, value): | ||
| if isinstance(value, _FakeTensor): | ||
| value = value.array | ||
| self.array.__setitem__(key, value) | ||
|
|
||
| def __iter__(self): | ||
| if self.array.ndim == 1: | ||
| for value in self.array: | ||
| yield value.item() if hasattr(value, "item") else value | ||
| else: | ||
| for row in self.array: | ||
| yield _FakeTensor(row, place=self.place) | ||
|
|
||
| def _binary_op(self, other: Any, op): | ||
| other_array = other.array if isinstance(other, _FakeTensor) else other | ||
| return _FakeTensor(op(self.array, other_array), place=self.place) | ||
|
|
||
| def __add__(self, other): | ||
| return self._binary_op(other, np.add) | ||
|
|
||
| def __radd__(self, other): | ||
| return self._binary_op(other, np.add) | ||
|
|
||
| def __sub__(self, other): | ||
| return self._binary_op(other, np.subtract) | ||
|
|
||
| def __rsub__(self, other): | ||
| return _FakeTensor(other, place=self.place)._binary_op(self, np.subtract) | ||
|
|
||
| def __truediv__(self, other): | ||
| return self._binary_op(other, np.divide) | ||
|
|
||
| def __mul__(self, other): # pragma: no cover - completeness helper | ||
| return self._binary_op(other, np.multiply) | ||
|
|
||
| def __eq__(self, other): | ||
| other_array = other.array if isinstance(other, _FakeTensor) else other | ||
| return self.array == other_array | ||
|
|
There was a problem hiding this comment.
The _FakeTensor class lacks documentation explaining its purpose as a numpy-backed mock for paddle.Tensor. Given its complexity (100+ lines), a class docstring explaining its role in the testing framework would significantly improve code maintainability.
| def _make_cursor(prompt_lens: List[int], device: str = "cpu", num_tokens: Optional[List[int]] = None): | ||
| paddle = sys.modules["paddle"] | ||
| metadata = sys.modules["fastdeploy.model_executor.layers.pool.metadata"] | ||
| if num_tokens is None: | ||
| num_tokens = prompt_lens | ||
| starts = [0] | ||
| for length in num_tokens[:-1]: | ||
| starts.append(starts[-1] + length) | ||
| first_indices = paddle.to_tensor(starts) | ||
| last_indices = paddle.to_tensor([s + l - 1 for s, l in zip(starts, num_tokens)]) | ||
| if device == "gpu": | ||
| first_indices = first_indices.cuda() | ||
| last_indices = last_indices.cuda() | ||
| return metadata.PoolingCursor( | ||
| index=list(range(len(prompt_lens))), | ||
| first_token_indices_gpu=first_indices, | ||
| last_token_indices_gpu=last_indices, | ||
| prompt_lens_cpu=paddle.to_tensor(prompt_lens), | ||
| num_scheduled_tokens_cpu=paddle.to_tensor(num_tokens), | ||
| ) | ||
|
|
||
|
|
||
| def _make_metadata( | ||
| prompt_lens: List[int], | ||
| pooling_params: List[Any], | ||
| token_ids: Optional[np.ndarray] = None, | ||
| device: str = "cpu", | ||
| num_tokens: Optional[List[int]] = None, | ||
| ): | ||
| paddle = sys.modules["paddle"] | ||
| metadata_mod = sys.modules["fastdeploy.model_executor.layers.pool.metadata"] | ||
| prompt_tensor = paddle.to_tensor(prompt_lens) | ||
| token_tensor = None if token_ids is None else paddle.to_tensor(token_ids) | ||
| cursor = _make_cursor(prompt_lens, device=device, num_tokens=num_tokens) | ||
| return metadata_mod.PoolingMetadata( | ||
| prompt_lens=prompt_tensor, | ||
| prompt_token_ids=token_tensor, | ||
| pooling_params=pooling_params, | ||
| pooling_cursor=cursor, | ||
| ) |
There was a problem hiding this comment.
The helper functions _make_cursor() and _make_metadata() lack docstrings. Given they are used extensively in the tests, adding documentation explaining their parameters and return values would improve test maintainability.
| @@ -0,0 +1,693 @@ | |||
| import argparse | |||
There was a problem hiding this comment.
The argparse import at the beginning of the file is only used at the very end (line 687) for a command-line feature. Consider moving the import closer to where it's used or adding a comment explaining why it's needed at the module level.
|
|
||
| def __truediv__(self, other): | ||
| return self._binary_op(other, np.divide) | ||
|
|
There was a problem hiding this comment.
The __mul__ method has a pragma comment # pragma: no cover - completeness helper suggesting it's not intended to be tested, yet it's implemented. If this method is needed for completeness but not used in tests, consider adding a comment explaining why it's needed. Otherwise, if it's truly not needed, consider removing it to reduce maintenance burden.
| # Implemented for completeness: ensures _FakeTensor supports all standard binary operations | |
| # even if not directly tested. This avoids runtime errors in test code using multiplication. |
| def _build_paddle_module() -> types.ModuleType: | ||
| paddle = types.ModuleType("paddle") | ||
| paddle.Tensor = _FakeTensor | ||
|
|
||
| def to_tensor(data, dtype=None): | ||
| return _FakeTensor(data, dtype=dtype) | ||
|
|
||
| def zeros(shape, dtype="float32"): | ||
| return _FakeTensor(np.zeros(shape, dtype=dtype)) | ||
|
|
||
| def cumsum(tensor, axis=0, out=None): | ||
| result = _FakeTensor(np.cumsum(tensor.array, axis=axis), place=tensor.place) | ||
| if out is not None: | ||
| out[:] = result | ||
| return out | ||
| return result | ||
|
|
||
| def stack(tensors): | ||
| arrays = [tensor.array if isinstance(tensor, _FakeTensor) else tensor for tensor in tensors] | ||
| return _FakeTensor(np.stack(arrays)) | ||
|
|
||
| def all(tensor): | ||
| if isinstance(tensor, _FakeTensor): | ||
| value = tensor.array | ||
| else: | ||
| value = np.array(tensor) | ||
| return _FakeTensor(np.array(value.all(), dtype=bool)) | ||
|
|
||
| paddle.to_tensor = to_tensor | ||
| paddle.zeros = zeros | ||
| paddle.cumsum = cumsum | ||
| paddle.stack = stack | ||
| paddle.all = all | ||
|
|
||
| nn_module = types.ModuleType("paddle.nn") | ||
|
|
||
| class Layer(_FakeLayer): | ||
| pass | ||
|
|
||
| nn_module.Layer = Layer | ||
| nn_module.Identity = _Identity | ||
| nn_module.Sigmoid = _Sigmoid | ||
| nn_module.Softmax = _Softmax | ||
|
|
||
| functional = types.ModuleType("paddle.nn.functional") | ||
|
|
||
| def _ensure_array(x): | ||
| return x.array if isinstance(x, _FakeTensor) else np.array(x) | ||
|
|
||
| def sigmoid(x): | ||
| arr = _ensure_array(x) | ||
| return _FakeTensor(1 / (1 + np.exp(-arr))) | ||
|
|
||
| def softmax(x, axis=-1): | ||
| arr = _ensure_array(x) | ||
| shifted = arr - arr.max(axis=axis, keepdims=True) | ||
| exp = np.exp(shifted) | ||
| return _FakeTensor(exp / exp.sum(axis=axis, keepdims=True)) | ||
|
|
||
| def normalize(x, p=2, axis=-1): | ||
| arr = _ensure_array(x).astype(np.float32) | ||
| norm = np.linalg.norm(arr, ord=p, axis=axis, keepdims=True) | ||
| norm[norm == 0] = 1 | ||
| return _FakeTensor(arr / norm) | ||
|
|
||
| functional.sigmoid = sigmoid | ||
| functional.softmax = softmax | ||
| functional.normalize = normalize | ||
|
|
||
| paddle.nn = nn_module | ||
| paddle.nn.functional = functional | ||
| paddle._FakeTensor = _FakeTensor | ||
| paddle._FakePlace = _FakePlace | ||
| paddle.F = functional | ||
| return paddle |
There was a problem hiding this comment.
The _build_paddle_module() function and other _build_*_module() functions lack documentation. These functions create significant mock infrastructure and should include docstrings explaining what modules they're mocking and why, to help future maintainers understand the test architecture.
| def _import_pooler_module(): | ||
| saved: Dict[str, Optional[types.ModuleType]] = {} | ||
|
|
||
| def _install(name: str, module: types.ModuleType): | ||
| saved[name] = sys.modules.get(name) | ||
| sys.modules[name] = module | ||
|
|
||
| for name, builder in [ | ||
| ("paddle", _build_paddle_module), | ||
| ("fastdeploy.config", _build_config_module), | ||
| ("fastdeploy.engine.pooling_params", _build_pooling_params_module), | ||
| ("fastdeploy.model_executor.layers.pool.metadata", _build_metadata_module), | ||
| ("fastdeploy.output.pooler", _build_output_module), | ||
| ("fastdeploy.utils", _build_utils_module), | ||
| ("fastdeploy.engine.tasks", _build_tasks_module), | ||
| ]: | ||
| module = builder() | ||
| _install(name, module) | ||
| if name == "paddle": | ||
| _install("paddle.nn", module.nn) | ||
| _install("paddle.nn.functional", module.nn.functional) | ||
|
|
||
| fastdeploy_pkg = types.ModuleType("fastdeploy") | ||
| fastdeploy_pkg.__path__ = [] | ||
| _install("fastdeploy", fastdeploy_pkg) | ||
| for pkg_name in ("fastdeploy.model_executor", "fastdeploy.model_executor.layers"): | ||
| pkg = types.ModuleType(pkg_name) | ||
| pkg.__path__ = [] | ||
| _install(pkg_name, pkg) | ||
|
|
||
| spec = importlib.util.spec_from_file_location( | ||
| "fastdeploy.model_executor.layers.pooler", | ||
| PROJECT_ROOT / "fastdeploy" / "model_executor" / "layers" / "pooler.py", | ||
| ) | ||
| assert spec and spec.loader is not None | ||
| module = importlib.util.module_from_spec(spec) | ||
| sys.modules[spec.name] = module | ||
| spec.loader.exec_module(module) | ||
|
|
||
| def _cleanup(): | ||
| for name, original in saved.items(): | ||
| if original is None: | ||
| sys.modules.pop(name, None) | ||
| else: | ||
| sys.modules[name] = original | ||
| for name in [ | ||
| "fastdeploy.model_executor.layers.pooler", | ||
| "fastdeploy.model_executor.layers", | ||
| "fastdeploy.model_executor", | ||
| "fastdeploy", | ||
| ]: | ||
| sys.modules.pop(name, None) | ||
|
|
||
| return module, _cleanup |
There was a problem hiding this comment.
[nitpick] The _import_pooler_module() function is quite complex (54 lines) and handles multiple responsibilities: module creation, installation, cleanup, and actual import. Consider breaking this into smaller, more focused functions (e.g., _create_mock_modules(), _install_modules(), _import_target_module()) to improve readability and maintainability.
| def __repr__(self) -> str: # pragma: no cover - debug helper | ||
| return f"_FakeTensor({self.array!r})" |
There was a problem hiding this comment.
The __repr__ method includes a pragma comment # pragma: no cover - debug helper but this is a test file where coverage is being measured. Consider removing the pragma comment since all code in the test infrastructure should ideally be covered, or if it's truly just a debug helper that won't be called in tests, add a comment explaining why it exists.
| def dtype(self): # pragma: no cover - compatibility helper | ||
| return self.array.dtype |
There was a problem hiding this comment.
The dtype property has a pragma comment # pragma: no cover - compatibility helper but is actually used in test code (e.g., line 534 checks result.dtype). This pragma comment appears incorrect and should be removed since the property is actually covered by tests.
| self.assertTrue(np.allclose(outputs[0].array.sum(), 1.0)) | ||
| self.assertTrue(np.allclose(outputs[1].array, logits[1].array)) |
There was a problem hiding this comment.
Similar to line 573, the np.allclose calls on lines 581-582 should explicitly specify tolerance parameters for more predictable test behavior and better test maintainability.
| self.assertTrue(np.allclose(outputs[0].array.sum(), 1.0)) | |
| self.assertTrue(np.allclose(outputs[1].array, logits[1].array)) | |
| self.assertTrue(np.allclose(outputs[0].array.sum(), 1.0, rtol=1e-5, atol=1e-8)) | |
| self.assertTrue(np.allclose(outputs[1].array, logits[1].array, rtol=1e-5, atol=1e-8)) |
1 participant
Motivation
NO.19 功能模块 fastdeploy/model_executor/layers/pooler.py 单测补充
Modifications
add tests/model_executor/test_pooler.py
Usage or Command
tests/model_executor/test_pooler.py:Accuracy Tests
tests/model_executor/test_pooler.py:Checklist
[FDConfig],[APIServer],[Engine],[Scheduler],[PD Disaggregation],[Executor],[Graph Optimization],[Speculative Decoding],[RL],[Models],[Quantization],[Loader],[OP],[KVCache],[DataProcessor],[BugFix],[Docs],[CI],[Optimization],[Feature],[Benchmark],[Others],[XPU],[HPU],[GCU],[DCU],[Iluvatar],[Metax]]pre-commitbefore commit.releasebranch, make sure the PR has been submitted to thedevelopbranch, then cherry-pick it to thereleasebranch with the[Cherry-Pick]PR tag.