Repository: LMCache/LMCache
File: tests/v1/storage_backend/test_storage_plugin.py (Python)

Code before:
        ),  # (num_layers, 2, chunk_size, num_heads, head_dim)
    )


def create_test_memory_obj(shape=(2, 16, 8, 128), dtype=torch.bfloat16) -> MemoryObj:
    """Create a test MemoryObj using AdHocMemoryAllocator."""
    allocator = AdHocMemoryAllocator(device="cpu")
    memory_obj = allocator.allocate(shape, dtype, fmt=MemoryFormat.KV_T2D)
    return memory_obj


def get_mock_backend(storage_manager_or_backends):
    """
    Get the mock storage plugin instance from storage manager or backends dict.
    Handles both StorageManager and OrderedDict[str, StorageBackendInterface].
    """
    if hasattr(storage_manager_or_backends, "storage_backends"):
        backends = storage_manager_or_backends.storage_backends
    else:
        backends = storage_manager_or_backends

    return backends.get("mock_backend")


class TestCreateDynamicBackends:
    """Test cases for create_dynamic_backends functionality."""

    @pytest.fixture
    def event_manager(self):
        """Create an EventManager for testing."""
        return EventManager()

    @pytest.fixture
    def async_loop(self):
        """Create an async event loop for testing."""
        loop = asyncio.new_event_loop()
        yield loop
        loop.close()

    @pytest.fixture
    def storage_manager_with_mock_backend(self, event_manager):
        """Fixture to create a StorageManager with the mock storage plugin loaded."""
        config = create_test_config(
            extra_config=MOCK_BACKEND_EXTRA_CONFIG,
            storage_plugins=MOCK_BACKEND_STORAGE_PLUGINS,
        )
        metadata = create_test_metadata()

        storage_manager = StorageManager(
            config=config,
            metadata=metadata,

Review comment (suggestion):
```suggestion
    """Test cases for storage_plugin_launcher functionality."""
```

Code after:
        ),  # (num_layers, 2, chunk_size, num_heads, head_dim)
    )


def create_test_memory_obj(shape=(2, 16, 8, 128), dtype=torch.bfloat16) -> MemoryObj:
    """Create a test MemoryObj using AdHocMemoryAllocator."""
    allocator = AdHocMemoryAllocator(device="cpu")
    memory_obj = allocator.allocate(shape, dtype, fmt=MemoryFormat.KV_T2D)
    return memory_obj


def get_mock_backend(storage_manager_or_backends):
    """
    Get the mock storage plugin instance from storage manager or backends dict.
    Handles both StorageManager and OrderedDict[str, StorageBackendInterface].
    """
    if hasattr(storage_manager_or_backends, "storage_backends"):
        backends = storage_manager_or_backends.storage_backends
    else:
        backends = storage_manager_or_backends

    return backends.get("mock_backend")


class TestCreateDynamicBackends:
    """Test cases for storage_plugin_launcher functionality."""

    @pytest.fixture
    def event_manager(self):
        """Create an EventManager for testing."""
        return EventManager()

    @pytest.fixture
    def async_loop(self):
        """Create an async event loop for testing."""
        loop = asyncio.new_event_loop()
        yield loop
        loop.close()

    @pytest.fixture
    def storage_manager_with_mock_backend(self, event_manager):
        """Fixture to create a StorageManager with the mock storage plugin loaded."""
        config = create_test_config(
            extra_config=MOCK_BACKEND_EXTRA_CONFIG,
            storage_plugins=MOCK_BACKEND_STORAGE_PLUGINS,
        )
        metadata = create_test_metadata()

        storage_manager = StorageManager(
            config=config,
            metadata=metadata,