#46355 [Test][KV Offloading] Add unit tests for OffloadingSpecFactory and SecondaryTierFactory

原始 PR 作者 Alex-ai-future 合并时间 2026-06-22 22:45 文件变更 4 提交数 3 评论 4 代码增减 +424 / -0

执行摘要

为 KV offloading 工厂类添加 21 个单元测试

PR body 指出 OffloadingSpecFactory 和 SecondaryTierFactory 作为 KV offloading 的插件入口，此前完全无单元测试，任何模块路径变更或注册逻辑调整都可能引发运行时崩溃且无法被 CI 捕获。本 PR 通过 21 个测试锁定工厂行为，确保重构和外部集成（如 llm-d-kv-cache）的安全性。

适合精读测试编写模式，尤其是如何测试动态注册工厂类（restore_registry fixture 避免测试间泄漏，_make_vllm_config 构造真实配置）。对于维护 KV offloading 模块的工程师，应关注这些测试在 CI 中的运行状态。

讨论亮点

评审共 2 条评论，均得到解决：

orozery 指出 test_create_tier_passes_config 无实际断言，是 no-op，作者在后续提交中移除了该测试。
orozery 建议在 pytest.raises 中使用 match 参数进行模式匹配（如 "Unknown secondary tier type.*Supported types:"），作者采纳并修改。

实现拆解

测试基础设施：在 tests/v1/kv_offload/ 和 tests/v1/kv_offload/tiering/ 添加 __init__.py 包文件，使 pytest 能正确发现测试模块。
OffloadingSpecFactory 测试（tests/v1/kv_offload/test_factory.py）：使用 restore_registry fixture 自动保存/恢复内部注册表；提供 _make_vllm_config 和 _make_kv_cache_config 辅助函数构造真实配置；编写预注册完整性哨兵测试（验证所有注册路径可导入且返回正确子类）、端到端 spec 构建测试、动态导入测试（spec_module_path 路径）、错误路径测试（未注册 spec、缺失配置、重复注册）。
SecondaryTierFactory 测试（tests/v1/kv_offload/tiering/test_factory.py）：类似地使用 restore_registry 和 _make_mock_args 辅助；编写预注册完整性哨兵测试、单/多 tier 创建测试、自定义注册测试（模拟外部项目注册新 tier 类型）、错误路径测试（缺少 type、未知 type、重复注册）。
Review 反馈处理：根据 orozery 评审，删除无效的 test_create_tier_passes_config 测试，将错误消息断言改用 pattern matching，并重命名测试文件从 test_tier_factory.py 到 test_factory.py。

文件	模块	状态	重要度
`tests/v1/kv_offload/test_factory.py`	卸载工厂	added	7.48
`tests/v1/kv_offload/tiering/test_factory.py`	次层工厂	added	7.29
`tests/v1/kv_offload/__init__.py`	测试包	added	3.11
`tests/v1/kv_offload/tiering/__init__.py`	测试包	added	3.11

关键符号

restore_registry _make_vllm_config _make_kv_cache_config _make_mock_args test_pre_registered_specs_can_be_imported test_cpu_spec_registered test_tiering_spec_registered test_get_spec_cls_returns_registered_class test_get_spec_cls_default_to_cpu test_pre_registered_tiers_can_be_imported test_example_tier_registered test_create_tier_from_registry test_create_multiple_tiers test_register_new_tier_type test_missing_tier_type_raises

关键源码片段

tests/v1/kv_offload/tiering/test_factory.py test-coverage

覆盖 SecondaryTierFactory 的预注册完整性、多 tier 创建和错误路径

"""
为 SecondaryTierFactory 编写的单元测试。
覆盖预注册完整性、多 tier 创建、自定义注册及错误路径。
"""

from unittest.mock import MagicMock
import pytest

from vllm.v1.kv_offload.tiering.base import SecondaryTierManager
from vllm.v1.kv_offload.tiering.example.manager import ExampleSecondaryTierManager
from vllm.v1.kv_offload.tiering.factory import SecondaryTierFactory

# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture(autouse=True)
def restore_registry():
    """保存并恢复 SecondaryTierFactory._registry，防止测试间泄漏"""
    original = dict(SecondaryTierFactory._registry)
    yield
    SecondaryTierFactory._registry = original


def _make_mock_args():
    """为 create_secondary_tier 构造 mock 参数 (primary_kv_view, offloading_spec)"""
    return MagicMock(), MagicMock()


# ---------------------------------------------------------------------------
# Pre-registration integrity (CI sentinel)
# ---------------------------------------------------------------------------


def test_pre_registered_tiers_can_be_imported():
    """CI 哨兵：示例/FS/OBJ 路径必须可导入并返回 SecondaryTierManager 子类"""
    for tier_type in SecondaryTierFactory._registry:
        cls = SecondaryTierFactory._registry[tier_type]()
        assert issubclass(cls, SecondaryTierManager)


def test_example_tier_registered():
    """验证示例 tier 已注册并指向正确类"""
    cls = SecondaryTierFactory._registry["example"]()
    assert cls is ExampleSecondaryTierManager


# ---------------------------------------------------------------------------
# Normal path — create_secondary_tier
# ---------------------------------------------------------------------------


def test_create_tier_from_registry():
    """已注册 tier_type 创建实例，且正确设置 tier_type"""
    primary_kv_view, offloading_spec = _make_mock_args()
    tier_config = {"type": "example"}
    tier = SecondaryTierFactory.create_secondary_tier(
        tier_config, primary_kv_view, offloading_spec
    )
    assert isinstance(tier, SecondaryTierManager)
    assert tier.tier_type == "example"

评论区精华

测试 `test_create_tier_passes_config` 被判定为 no-op 测试

orozery 指出该测试没有实际断言，无法验证 config 是否传递到 secondary tier。作者确认后删除了该测试。

结论：删除无效测试，保持测试套件纯洁 · 已解决

错误消息断言使用 pattern matching style

orozery 建议在 `pytest.raises` 中使用 `match` 参数进行模式匹配（如 `"Unknown secondary tier type.*Supported types:"`），以提高断言的鲁棒性。

结论：作者采纳建议，修改了断言写法 · 已解决

风险与影响

本 PR 仅涉及测试代码新增，无产品代码变更，引入风险极低。但测试未覆盖并发注册场景和序列化/反序列化路径，这些可作为后续补充方向。

对用户无直接影响。对开发团队，CI 现在会检查工厂预注册路径是否可导入，防止因文件移动导致的运行时崩溃；外部项目（llm-d-kv-cache）依赖的 spec_module_path 动态加载路径得到验证，降低集成风险；开发者在修改工厂注册逻辑时可立即获得测试反馈。

测试覆盖缺口已补无产品代码变更

关联 Issue

未识别关联 Issue

当前没有检测到明确关联的 Issue 链接，后续同步到相关引用后会出现在这里。

完整报告

执行摘要

一句话：为KV offloading工厂类添加21个单元测试
推荐动作：适合精读测试编写模式，尤其是如何测试动态注册工厂类（restore_registry fixture 避免测试间泄漏，_make_vllm_config 构造真实配置）。对于维护 KV offloading 模块的工程师，应关注这些测试在 CI 中的运行状态。

功能与动机

实现拆解

测试基础设施：在 tests/v1/kv_offload/ 和 tests/v1/kv_offload/tiering/ 添加 __init__.py 包文件，使 pytest 能正确发现测试模块。
OffloadingSpecFactory 测试（tests/v1/kv_offload/test_factory.py）：使用 restore_registry fixture 自动保存/恢复内部注册表；提供 _make_vllm_config 和 _make_kv_cache_config 辅助函数构造真实配置；编写预注册完整性哨兵测试（验证所有注册路径可导入且返回正确子类）、端到端 spec 构建测试、动态导入测试（spec_module_path 路径）、错误路径测试（未注册 spec、缺失配置、重复注册）。
SecondaryTierFactory 测试（tests/v1/kv_offload/tiering/test_factory.py）：类似地使用 restore_registry 和 _make_mock_args 辅助；编写预注册完整性哨兵测试、单/多 tier 创建测试、自定义注册测试（模拟外部项目注册新 tier 类型）、错误路径测试（缺少 type、未知 type、重复注册）。
Review 反馈处理：根据 orozery 评审，删除无效的 test_create_tier_passes_config 测试，将错误消息断言改用 pattern matching，并重命名测试文件从 test_tier_factory.py 到 test_factory.py。

关键文件：

tests/v1/kv_offload/test_factory.py（模块卸载工厂；类别 test；类型 test-coverage；符号 restore_registry, _make_vllm_config, _make_kv_cache_config, test_pre_registered_specs_can_be_imported）: 主要测试文件，覆盖 OffloadingSpecFactory 的 21 个测试场景
tests/v1/kv_offload/tiering/test_factory.py（模块次层工厂；类别 test；类型 test-coverage；符号 restore_registry, _make_mock_args, test_pre_registered_tiers_can_be_imported, test_example_tier_registered）: 覆盖 SecondaryTierFactory 的预注册完整性、多 tier 创建和错误路径
tests/v1/kv_offload/__init__.py（模块测试包；类别 test；类型 test-coverage）: 使测试包可被 Python 识别
tests/v1/kv_offload/tiering/__init__.py（模块测试包；类别 test；类型 test-coverage）: 使 tiering 子包可被 Python 识别

关键符号：restore_registry, _make_vllm_config, _make_kv_cache_config, _make_mock_args, test_pre_registered_specs_can_be_imported, test_cpu_spec_registered, test_tiering_spec_registered, test_get_spec_cls_returns_registered_class, test_get_spec_cls_default_to_cpu, test_pre_registered_tiers_can_be_imported, test_example_tier_registered, test_create_tier_from_registry, test_create_multiple_tiers, test_register_new_tier_type, test_missing_tier_type_raises

关键源码片段

`tests/v1/kv_offload/tiering/test_factory.py`

覆盖 SecondaryTierFactory 的预注册完整性、多 tier 创建和错误路径

"""
为 SecondaryTierFactory 编写的单元测试。
覆盖预注册完整性、多 tier 创建、自定义注册及错误路径。
"""

from unittest.mock import MagicMock
import pytest

from vllm.v1.kv_offload.tiering.base import SecondaryTierManager
from vllm.v1.kv_offload.tiering.example.manager import ExampleSecondaryTierManager
from vllm.v1.kv_offload.tiering.factory import SecondaryTierFactory

# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture(autouse=True)
def restore_registry():
    """保存并恢复 SecondaryTierFactory._registry，防止测试间泄漏"""
    original = dict(SecondaryTierFactory._registry)
    yield
    SecondaryTierFactory._registry = original


def _make_mock_args():
    """为 create_secondary_tier 构造 mock 参数 (primary_kv_view, offloading_spec)"""
    return MagicMock(), MagicMock()


# ---------------------------------------------------------------------------
# Pre-registration integrity (CI sentinel)
# ---------------------------------------------------------------------------


def test_pre_registered_tiers_can_be_imported():
    """CI 哨兵：示例/FS/OBJ 路径必须可导入并返回 SecondaryTierManager 子类"""
    for tier_type in SecondaryTierFactory._registry:
        cls = SecondaryTierFactory._registry[tier_type]()
        assert issubclass(cls, SecondaryTierManager)


def test_example_tier_registered():
    """验证示例 tier 已注册并指向正确类"""
    cls = SecondaryTierFactory._registry["example"]()
    assert cls is ExampleSecondaryTierManager


# ---------------------------------------------------------------------------
# Normal path — create_secondary_tier
# ---------------------------------------------------------------------------


def test_create_tier_from_registry():
    """已注册 tier_type 创建实例，且正确设置 tier_type"""
    primary_kv_view, offloading_spec = _make_mock_args()
    tier_config = {"type": "example"}
    tier = SecondaryTierFactory.create_secondary_tier(
        tier_config, primary_kv_view, offloading_spec
    )
    assert isinstance(tier, SecondaryTierManager)
    assert tier.tier_type == "example"

评论区精华

评审共 2 条评论，均得到解决：

orozery 指出 test_create_tier_passes_config 无实际断言，是 no-op，作者在后续提交中移除了该测试。
orozery 建议在 pytest.raises 中使用 match 参数进行模式匹配（如 "Unknown secondary tier type.*Supported types:"），作者采纳并修改。
测试 test_create_tier_passes_config 被判定为 no-op (testing): 删除无效测试，保持测试套件纯洁
错误消息断言使用 pattern matching (style): 作者采纳建议，修改了断言写法

风险与影响

风险：本 PR 仅涉及测试代码新增，无产品代码变更，引入风险极低。但测试未覆盖并发注册场景和序列化/反序列化路径，这些可作为后续补充方向。
影响：对用户无直接影响。对开发团队，CI 现在会检查工厂预注册路径是否可导入，防止因文件移动导致的运行时崩溃；外部项目（llm-d-kv-cache）依赖的 spec_module_path 动态加载路径得到验证，降低集成风险；开发者在修改工厂注册逻辑时可立即获得测试反馈。
风险标记：测试覆盖缺口已补, 无产品代码变更

关联脉络

PR #43468 [feature][kv_offload] Self-describing KV events for OffloadingConnector: 同一个功能线，引入了 OffloadingConnector 和自描述 KV 事件，是工厂类的上游依赖。本 PR 的测试覆盖了这些工厂类的行为。
PR #46216 [CPUOffloadingManager] Maintain evictable list in LRUCachePolicy: 改进了 LRU 淘汰策略，属于 KV offload 子系统的优化。本 PR 的测试间接保障了工厂类的稳定性。

#46355 [Test][KV Offloading] Add unit tests for OffloadingSpecFactory and SecondaryTierFactory

执行摘要

为 KV offloading 工厂类添加 21 个单元测试

实现拆解

评论区精华

风险与影响

关联 Issue

未识别关联 Issue

完整报告

参与讨论