[CI] 【Hackathon 9th Sprint No.36】NO.36 功能模块单测补充 #5058
Conversation
|
Thanks for your contribution! |
There was a problem hiding this comment.
Pull Request Overview
This PR adds comprehensive unit tests for the fastdeploy/cache_manager/prefix_cache_manager.py module, achieving 80% code coverage as part of Hackathon 9th Sprint No.36. The tests cover various aspects of the PrefixCacheManager including block allocation/recycling, cache matching, multimodal input handling, swap operations, and process lifecycle management.
Key Changes
- Added 931 lines of test code with 27 test methods covering core functionality
- Implemented extensive mock infrastructure to enable testing without actual dependencies (paddle, paddleformers)
- Achieved 80% coverage of the prefix_cache_manager module with comprehensive test scenarios
| class _PollingProcess(_DummyProcess): | ||
| def __init__(self, *args, poll_value=None, **kwargs): | ||
| super().__init__(*args, **kwargs) | ||
| self._poll_value = poll_value | ||
|
|
||
| def poll(self): | ||
| return self._poll_value |
There was a problem hiding this comment.
Missing docstring for _PollingProcess class. Consider adding a brief description explaining it extends _DummyProcess to return a configurable poll value.
| class _DummyThread: | ||
| def __init__(self, target=None, **kwargs): | ||
| self.target = target | ||
| self.started = False | ||
|
|
||
| def start(self): | ||
| self.started = True |
There was a problem hiding this comment.
Missing docstring for _DummyThread class. Consider adding a brief description of its purpose as a test stub for threading.Thread.
| class _ImmediateFuture: | ||
| def __init__(self, fn=None, *args): | ||
| self._result = fn(*args) if fn is not None else None | ||
|
|
||
| def result(self): | ||
| return self._result | ||
|
|
||
| def done(self): | ||
| return True |
There was a problem hiding this comment.
Missing docstring for _ImmediateFuture class. Consider adding a brief description explaining it mimics a Future object that completes immediately.
| class _StubLogger: | ||
| def __init__(self): | ||
| self.logger = self | ||
|
|
||
| def setLevel(self, *_): | ||
| pass |
There was a problem hiding this comment.
Missing docstring for _StubLogger class. Consider adding a brief description explaining it provides a minimal logger stub for testing purposes.
| return self.metrics[name] | ||
|
|
||
|
|
||
| class _DummyIPCSignal: |
There was a problem hiding this comment.
Missing docstring for _DummyIPCSignal class. Consider adding a brief description of its purpose as a test stub for IPC signals.
| class _DummyIPCSignal: | |
| class _DummyIPCSignal: | |
| """Test stub for IPC signals used in unit tests.""" |
| ), | ||
| patch( | ||
| "fastdeploy.cache_manager.prefix_cache_manager.subprocess.Popen", | ||
| lambda *args, **kwargs: _DummyProcess(*args, **kwargs), |
There was a problem hiding this comment.
This 'lambda' is just a simple wrapper around a callable object. Use that object directly.
| ), | ||
| patch( | ||
| "fastdeploy.cache_manager.prefix_cache_manager.subprocess.Popen", | ||
| lambda *args, **kwargs: _DummyProcess(*args, **kwargs), |
There was a problem hiding this comment.
This 'lambda' is just a simple wrapper around a callable object. Use that object directly.
| ), | ||
| patch( | ||
| "fastdeploy.cache_manager.prefix_cache_manager.subprocess.Popen", | ||
| lambda *args, **kwargs: _DummyProcess(*args, **kwargs), |
There was a problem hiding this comment.
This 'lambda' is just a simple wrapper around a callable object. Use that object directly.
| lambda *args, **kwargs: _DummyProcess(*args, **kwargs), | |
| _DummyProcess, |
| manager.free_cpu_executor_pool = types.SimpleNamespace(submit=lambda fn, *args: _ImmediateFuture(fn, *args)) | ||
| manager.free_gpu_executor_pool = types.SimpleNamespace(submit=lambda fn, *args: _ImmediateFuture(fn, *args)) |
There was a problem hiding this comment.
This 'lambda' is just a simple wrapper around a callable object. Use that object directly.
| manager.free_cpu_executor_pool = types.SimpleNamespace(submit=lambda fn, *args: _ImmediateFuture(fn, *args)) | |
| manager.free_gpu_executor_pool = types.SimpleNamespace(submit=lambda fn, *args: _ImmediateFuture(fn, *args)) | |
| manager.free_cpu_executor_pool = types.SimpleNamespace(submit=_ImmediateFuture) | |
| manager.free_gpu_executor_pool = types.SimpleNamespace(submit=_ImmediateFuture) |
| manager.cache_config.enable_hierarchical_cache = True | ||
| manager.cache_task_queue = _DummyEngineCacheQueue() | ||
| manager.free_cpu_executor_pool = types.SimpleNamespace(submit=lambda fn, *args: _ImmediateFuture(fn, *args)) | ||
| manager.free_gpu_executor_pool = types.SimpleNamespace(submit=lambda fn, *args: _ImmediateFuture(fn, *args)) |
There was a problem hiding this comment.
This 'lambda' is just a simple wrapper around a callable object. Use that object directly.
| manager.free_gpu_executor_pool = types.SimpleNamespace(submit=lambda fn, *args: _ImmediateFuture(fn, *args)) | |
| manager.free_gpu_executor_pool = types.SimpleNamespace(submit=_ImmediateFuture) |
3 participants
Motivation
NO.36 功能模块 fastdeploy/cache_manager/prefix_cache_manager.py 单测补充
Modifications
new tests/cache_manager/prefix_cache_manager.py
Usage or Command
tests/cache_manager/prefix_cache_manager.py:Accuracy Tests
tests/cache_manager/prefix_cache_manager.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.