# SPDX-License-Identifier: Apache-2.0
"""
Unit tests for the L2 adapter factory registry and
PluginL2AdapterConfig.
"""

# Standard
import os
import tempfile
import types

# Third Party
import pytest

# First Party
from lmcache.v1.distributed.l2_adapters.config import (
    L2AdapterConfigBase,
    get_type_name_for_config,
)
from lmcache.v1.distributed.l2_adapters.factory import (
    create_l2_adapter_from_registry,
    register_l2_adapter_factory,
)
from lmcache.v1.distributed.l2_adapters.mock_l2_adapter import MockL2AdapterConfig
from lmcache.v1.distributed.l2_adapters.plugin_l2_adapter import PluginL2AdapterConfig
from lmcache.v1.distributed.l2_adapters.raw_block_l2_adapter import (
    RawBlockL2AdapterConfig,
)
from lmcache.v1.platform import create_event_notifier


def _has_raw_block_ext() -> bool:
    try:
        # Third Party
        import lmcache_rust_raw_block_io  # noqa: F401

        return True
    except Exception:
        return False


# =========================================================
# Helpers
# =========================================================


class _FakeL2Adapter:
    """Minimal stub that passes issubclass check."""

    def __init__(self, params=None, **kwargs):
        self.params = params
        self.kwargs = kwargs


class _NotAnAdapter:
    """Class that does NOT subclass L2AdapterInterface."""

    pass


def _make_fake_module(
    adapter_cls: type,
    class_name: str = "FakeL2Adapter",
) -> types.ModuleType:
    """Create a fake Python module containing *adapter_cls*."""
    mod = types.ModuleType("fake_l2_module")
    setattr(mod, class_name, adapter_cls)
    return mod


# =========================================================
# Tests: factory registry basics
# =========================================================


class TestFactoryRegistry:
    """Tests for register/create via factory registry."""

    def test_mock_factory_is_registered(self):
        """Mock adapter factory should be auto-registered
        on import."""
        config = MockL2AdapterConfig(
            max_size_gb=0.001,
            mock_bandwidth_gb=10.0,
        )
        name = get_type_name_for_config(config)
        assert name == "mock"

    def test_raw_block_factory_is_registered(self):
        config = RawBlockL2AdapterConfig(
            device_path="/tmp/raw-block-dev",
            slot_bytes=64 * 1024,
            use_odirect=False,
            meta_enable_periodic=False,
        )
        name = get_type_name_for_config(config)
        assert name == "raw_block"

    def test_create_mock_via_registry(self):
        """create_l2_adapter_from_registry creates a
        MockL2Adapter."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.mock_l2_adapter import (
            MockL2Adapter,
        )

        config = MockL2AdapterConfig(
            max_size_gb=0.001,
            mock_bandwidth_gb=10.0,
        )
        adapter = create_l2_adapter_from_registry(config)
        assert isinstance(adapter, MockL2Adapter)
        adapter.close()

    @pytest.mark.skipif(
        not _has_raw_block_ext(),
        reason="lmcache_rust_raw_block_io extension not installed",
    )
    def test_create_raw_block_via_registry(self):
        # First Party
        from lmcache.v1.distributed.l2_adapters.raw_block_l2_adapter import (
            RawBlockL2Adapter,
        )

        with tempfile.TemporaryDirectory() as td:
            dev_path = os.path.join(td, "dev.bin")
            with open(dev_path, "wb") as f:
                f.truncate(8 * 1024 * 1024)

            config = RawBlockL2AdapterConfig(
                device_path=dev_path,
                slot_bytes=64 * 1024,
                use_odirect=False,
                meta_total_bytes=1 * 1024 * 1024,
                meta_enable_periodic=False,
            )
            adapter = create_l2_adapter_from_registry(config)
            assert isinstance(adapter, RawBlockL2Adapter)
            adapter.close()

    def test_duplicate_factory_raises(self):
        """Registering the same factory name twice should
        raise ValueError."""
        with pytest.raises(ValueError, match="already registered"):
            register_l2_adapter_factory("mock", lambda c, **kw: None)

    def test_unknown_config_raises(self):
        """Config class not in the registry should fail."""

        class _UnknownConfig:
            pass

        with pytest.raises(ValueError, match="Unregistered"):
            get_type_name_for_config(_UnknownConfig())


# =========================================================
# Tests: PluginL2AdapterConfig parsing
# =========================================================


class TestPluginL2AdapterConfig:
    """Tests for PluginL2AdapterConfig.from_dict."""

    def test_valid_config(self):
        d = {
            "type": "plugin",
            "module_path": "my.module",
            "class_name": "MyAdapter",
            "adapter_params": {"host": "localhost"},
        }
        cfg = PluginL2AdapterConfig.from_dict(d)
        assert cfg.module_path == "my.module"
        assert cfg.class_name == "MyAdapter"
        assert cfg.adapter_params == {"host": "localhost"}

    def test_default_adapter_params(self):
        d = {
            "type": "plugin",
            "module_path": "my.module",
            "class_name": "MyAdapter",
        }
        cfg = PluginL2AdapterConfig.from_dict(d)
        assert cfg.adapter_params == {}

    def test_missing_module_path_raises(self):
        d = {"type": "plugin", "class_name": "X"}
        with pytest.raises(ValueError, match="module_path"):
            PluginL2AdapterConfig.from_dict(d)

    def test_missing_class_name_raises(self):
        d = {"type": "plugin", "module_path": "x"}
        with pytest.raises(ValueError, match="class_name"):
            PluginL2AdapterConfig.from_dict(d)

    def test_invalid_adapter_params_raises(self):
        d = {
            "type": "plugin",
            "module_path": "x",
            "class_name": "X",
            "adapter_params": "not_a_dict",
        }
        with pytest.raises(ValueError, match="adapter_params"):
            PluginL2AdapterConfig.from_dict(d)


# =========================================================
# Tests: plugin adapter factory dynamic loading
# =========================================================


class TestPluginAdapterFactory:
    """Tests for the plugin adapter factory using monkeypatch
    to mock importlib.import_module."""

    def test_load_external_adapter(self, monkeypatch):
        """Successfully load an external adapter class."""
        # First Party
        from lmcache.v1.distributed.l2_adapters import plugin_l2_adapter as plugin_mod

        # Make _FakeL2Adapter pass issubclass check
        monkeypatch.setattr(
            plugin_mod,
            "_L2AI",
            _FakeL2Adapter,
        )

        fake_mod = _make_fake_module(_FakeL2Adapter)
        monkeypatch.setattr(
            "importlib.import_module",
            lambda path: fake_mod,
        )

        config = PluginL2AdapterConfig(
            module_path="fake_l2_module",
            class_name="FakeL2Adapter",
            adapter_params={"host": "localhost"},
        )
        adapter = create_l2_adapter_from_registry(config)
        assert isinstance(adapter, _FakeL2Adapter)
        assert adapter.params["host"] == "localhost"

    def test_import_error_raises(self, monkeypatch):
        """ImportError propagates when module not found."""
        monkeypatch.setattr(
            "importlib.import_module",
            lambda p: (_ for _ in ()).throw(ImportError("no such module")),
        )
        config = PluginL2AdapterConfig(
            module_path="nonexistent",
            class_name="X",
        )
        with pytest.raises(ImportError, match="nonexistent"):
            create_l2_adapter_from_registry(config)

    def test_missing_class_raises(self, monkeypatch):
        """AttributeError when class not in module."""
        fake_mod = types.ModuleType("empty_mod")
        monkeypatch.setattr(
            "importlib.import_module",
            lambda p: fake_mod,
        )
        config = PluginL2AdapterConfig(
            module_path="empty_mod",
            class_name="NoSuchClass",
        )
        with pytest.raises(AttributeError, match="NoSuchClass"):
            create_l2_adapter_from_registry(config)

    def test_not_subclass_raises(self, monkeypatch):
        """TypeError when class is not an L2AdapterInterface
        subclass."""
        fake_mod = _make_fake_module(_NotAnAdapter, "BadAdapter")
        monkeypatch.setattr(
            "importlib.import_module",
            lambda p: fake_mod,
        )
        config = PluginL2AdapterConfig(
            module_path="fake_mod",
            class_name="BadAdapter",
        )
        with pytest.raises(TypeError, match="not a subclass"):
            create_l2_adapter_from_registry(config)


# =========================================================
# Tests: plugin registration and initialization
# =========================================================


class _FakeL2AdapterWithDesc:
    """Stub that records l1_memory_desc."""

    def __init__(self, config, **kwargs):
        self.config = config
        self.l1_memory_desc = kwargs.get("l1_memory_desc")


class _FakeConfig(L2AdapterConfigBase):
    """Minimal config subclass for discovery tests."""

    def __init__(self, **kwargs):
        self.kwargs = kwargs

    @classmethod
    def from_dict(cls, d: dict) -> "_FakeConfig":
        return cls(**d)

    @classmethod
    def help(cls) -> str:
        return "fake"


class TestPluginRegistration:
    """Verify plugin self-registration on import."""

    def test_plugin_type_registered(self):
        """'plugin' config type should be registered."""
        cfg = PluginL2AdapterConfig(module_path="x", class_name="X")
        name = get_type_name_for_config(cfg)
        assert name == "plugin"

    def test_plugin_factory_registered(self):
        """'plugin' factory should be callable via
        create_l2_adapter_from_registry (smoke test
        with ImportError)."""
        cfg = PluginL2AdapterConfig(
            module_path="nonexistent.module",
            class_name="X",
        )
        with pytest.raises(ImportError):
            create_l2_adapter_from_registry(cfg)


class TestPluginInitialization:
    """Verify config-class discovery and l1_memory_desc
    forwarding during plugin initialization."""

    def _patch_base(self, monkeypatch, adapter_cls):
        """Make *adapter_cls* pass issubclass check."""
        # First Party
        from lmcache.v1.distributed.l2_adapters import plugin_l2_adapter as plugin_mod

        monkeypatch.setattr(
            plugin_mod,
            "_L2AI",
            adapter_cls,
        )

    # -- l1_memory_desc forwarding --

    def test_l1_memory_desc_forwarded(self, monkeypatch):
        """l1_memory_desc should reach the adapter
        constructor when provided."""
        self._patch_base(monkeypatch, _FakeL2AdapterWithDesc)
        fake_mod = _make_fake_module(
            _FakeL2AdapterWithDesc,
            "Adapter",
        )
        monkeypatch.setattr(
            "importlib.import_module",
            lambda _: fake_mod,
        )
        cfg = PluginL2AdapterConfig(
            module_path="m",
            class_name="Adapter",
            adapter_params={"k": "v"},
        )
        sentinel = object()
        adapter = create_l2_adapter_from_registry(cfg, l1_memory_desc=sentinel)
        assert adapter.l1_memory_desc is sentinel

    def test_l1_memory_desc_omitted(self, monkeypatch):
        """l1_memory_desc should be None when not
        provided."""
        self._patch_base(monkeypatch, _FakeL2AdapterWithDesc)
        fake_mod = _make_fake_module(
            _FakeL2AdapterWithDesc,
            "Adapter",
        )
        monkeypatch.setattr(
            "importlib.import_module",
            lambda _: fake_mod,
        )
        cfg = PluginL2AdapterConfig(
            module_path="m",
            class_name="Adapter",
        )
        adapter = create_l2_adapter_from_registry(cfg)
        assert adapter.l1_memory_desc is None

    # -- config class discovery --

    def test_init_with_explicit_config_class(self, monkeypatch):
        """Explicit config_class_name should be used."""
        self._patch_base(monkeypatch, _FakeL2AdapterWithDesc)
        fake_mod = _make_fake_module(
            _FakeL2AdapterWithDesc,
            "Adapter",
        )
        fake_mod.MyCfg = _FakeConfig  # type: ignore[attr-defined]
        monkeypatch.setattr(
            "importlib.import_module",
            lambda _: fake_mod,
        )
        cfg = PluginL2AdapterConfig(
            module_path="m",
            class_name="Adapter",
            config_class_name="MyCfg",
            adapter_params={"x": 1},
        )
        adapter = create_l2_adapter_from_registry(cfg)
        assert isinstance(adapter.config, _FakeConfig)
        assert adapter.config.kwargs == {"x": 1}

    def test_init_with_convention_config_class(self, monkeypatch):
        """Config class discovered via ClassName+'Config'
        convention."""
        self._patch_base(monkeypatch, _FakeL2AdapterWithDesc)
        fake_mod = _make_fake_module(
            _FakeL2AdapterWithDesc,
            "Adapter",
        )
        # Convention: "Adapter" + "Config"
        fake_mod.AdapterConfig = _FakeConfig  # type: ignore[attr-defined]
        monkeypatch.setattr(
            "importlib.import_module",
            lambda _: fake_mod,
        )
        cfg = PluginL2AdapterConfig(
            module_path="m",
            class_name="Adapter",
            adapter_params={"y": 2},
        )
        adapter = create_l2_adapter_from_registry(cfg)
        assert isinstance(adapter.config, _FakeConfig)

    def test_init_fallback_raw_dict(self, monkeypatch):
        """When no config class is found, adapter receives
        a raw dict."""
        self._patch_base(monkeypatch, _FakeL2AdapterWithDesc)
        fake_mod = _make_fake_module(
            _FakeL2AdapterWithDesc,
            "Adapter",
        )
        monkeypatch.setattr(
            "importlib.import_module",
            lambda _: fake_mod,
        )
        cfg = PluginL2AdapterConfig(
            module_path="m",
            class_name="Adapter",
            adapter_params={"z": 3},
        )
        adapter = create_l2_adapter_from_registry(cfg)
        assert isinstance(adapter.config, dict)
        assert adapter.config == {"z": 3}


# =========================================================
# Tests: lazy loading (add_pending_module,
# ensure_adapter_loaded, load_all_adapters,
# get_all_registered_names)
# =========================================================


class TestLazyLoading:
    """Tests for the deferred module import mechanism."""

    @pytest.fixture(autouse=True)
    def _isolate_registry(self, monkeypatch):
        """Snapshot and restore the factory registry and
        pending-module list around each test."""
        # First Party
        from lmcache.v1.distributed.l2_adapters import factory as fmod

        orig_reg = fmod._L2_ADAPTER_FACTORY_REGISTRY.copy()
        orig_pending = list(fmod._PENDING_MODULES)
        # Start each test with a clean slate
        fmod._PENDING_MODULES.clear()
        yield
        fmod._L2_ADAPTER_FACTORY_REGISTRY.clear()
        fmod._L2_ADAPTER_FACTORY_REGISTRY.update(orig_reg)
        fmod._PENDING_MODULES.clear()
        fmod._PENDING_MODULES.extend(orig_pending)

    # -- add_pending_module --

    def test_add_pending_module_appends(self):
        """Module path is appended to the pending list."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.factory import (
            _PENDING_MODULES,
            add_pending_module,
        )

        add_pending_module("fake.mod.a")
        assert "fake.mod.a" in _PENDING_MODULES

    def test_add_pending_module_dedup(self):
        """Duplicate paths are not added twice."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.factory import (
            _PENDING_MODULES,
            add_pending_module,
        )

        add_pending_module("fake.mod.b")
        add_pending_module("fake.mod.b")
        assert _PENDING_MODULES.count("fake.mod.b") == 1

    # -- ensure_adapter_loaded --

    def test_ensure_already_registered(self):
        """No import happens when name is already in the
        registry."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.factory import (
            _L2_ADAPTER_FACTORY_REGISTRY,
            _PENDING_MODULES,
            ensure_adapter_loaded,
        )

        _L2_ADAPTER_FACTORY_REGISTRY["pre"] = lambda c, d: None
        _PENDING_MODULES.append("should.not.import")
        ensure_adapter_loaded("pre")
        # Pending list untouched
        assert "should.not.import" in _PENDING_MODULES

    def test_ensure_imports_until_found(self, monkeypatch):
        """Modules are imported one-by-one until the
        requested name appears in the registry."""
        # First Party
        from lmcache.v1.distributed.l2_adapters import factory as fmod

        imported: list[str] = []

        def _fake_import(path):
            imported.append(path)
            if path == "mod_b":
                fmod._L2_ADAPTER_FACTORY_REGISTRY["lazy_b"] = lambda c, d: None

        monkeypatch.setattr("importlib.import_module", _fake_import)
        fmod._PENDING_MODULES.extend(["mod_a", "mod_b", "mod_c"])
        fmod.ensure_adapter_loaded("lazy_b")

        assert imported == ["mod_a", "mod_b"]
        # mod_c should remain pending
        assert "mod_c" in fmod._PENDING_MODULES

    def test_ensure_raises_last_import_error(self, monkeypatch):
        """When all pending modules fail, the last
        ImportError is raised."""
        # First Party
        from lmcache.v1.distributed.l2_adapters import factory as fmod

        def _fail_import(path):
            raise ImportError(path)

        monkeypatch.setattr("importlib.import_module", _fail_import)
        fmod._PENDING_MODULES.extend(["bad_a", "bad_b"])

        with pytest.raises(ImportError, match="bad_b"):
            fmod.ensure_adapter_loaded("missing")

    def test_ensure_no_error_when_not_found_silently(self, monkeypatch):
        """When pending modules import OK but name is
        still missing, no error is raised (no last_err).
        """
        # First Party
        from lmcache.v1.distributed.l2_adapters import factory as fmod

        monkeypatch.setattr("importlib.import_module", lambda p: None)
        fmod._PENDING_MODULES.extend(["ok_a", "ok_b"])

        # Should not raise
        fmod.ensure_adapter_loaded("nonexistent")

    # -- load_all_adapters --

    def test_load_all_adapters_imports_everything(self, monkeypatch):
        """All pending modules are imported."""
        # First Party
        from lmcache.v1.distributed.l2_adapters import factory as fmod

        imported: list[str] = []
        monkeypatch.setattr(
            "importlib.import_module",
            lambda p: imported.append(p),
        )
        fmod._PENDING_MODULES.extend(["m1", "m2", "m3"])
        fmod.load_all_adapters()

        assert imported == ["m1", "m2", "m3"]
        assert len(fmod._PENDING_MODULES) == 0

    def test_load_all_adapters_skips_failures(self, monkeypatch):
        """Modules that fail to import are silently
        skipped."""
        # First Party
        from lmcache.v1.distributed.l2_adapters import factory as fmod

        imported: list[str] = []

        def _selective_import(path):
            if path == "bad":
                raise ImportError(path)
            imported.append(path)

        monkeypatch.setattr("importlib.import_module", _selective_import)
        fmod._PENDING_MODULES.extend(["good1", "bad", "good2"])
        fmod.load_all_adapters()

        assert imported == ["good1", "good2"]
        assert len(fmod._PENDING_MODULES) == 0

    # -- get_all_registered_names --

    def test_get_all_registered_names_sorted(self, monkeypatch):
        """Returns sorted list after loading all pending
        modules."""
        # First Party
        from lmcache.v1.distributed.l2_adapters import factory as fmod

        def _register_import(path):
            fmod._L2_ADAPTER_FACTORY_REGISTRY[path] = lambda c, d: None

        monkeypatch.setattr("importlib.import_module", _register_import)
        fmod._PENDING_MODULES.extend(["z_mod", "a_mod"])

        names = fmod.get_all_registered_names()
        # Should contain at least the two we just added
        assert "z_mod" in names
        assert "a_mod" in names
        # Must be sorted
        assert names == sorted(names)

    def test_lazy_module_not_in_sys_modules(self, monkeypatch):
        """Pending modules are NOT in sys.modules until
        ensure_adapter_loaded triggers the import."""
        # Standard
        import sys

        # First Party
        from lmcache.v1.distributed.l2_adapters import factory as fmod

        sentinel = "test.lazy.sentinel.module"
        fmod._PENDING_MODULES.append(sentinel)
        assert sentinel not in sys.modules

        def _fake_import(path):
            mod = types.ModuleType(path)
            sys.modules[path] = mod
            fmod._L2_ADAPTER_FACTORY_REGISTRY["sentinel"] = lambda c, d: None

        monkeypatch.setattr("importlib.import_module", _fake_import)
        fmod.ensure_adapter_loaded("sentinel")
        assert sentinel in sys.modules
        # Cleanup
        sys.modules.pop(sentinel, None)


# =========================================================
# Tests: create_l2_adapter public API
# =========================================================


class TestCreateL2Adapter:
    """Tests for the public create_l2_adapter function."""

    def test_create_mock_adapter(self):
        """create_l2_adapter dispatches to MockL2Adapter."""
        # First Party
        from lmcache.v1.distributed.l2_adapters import (
            create_l2_adapter,
        )
        from lmcache.v1.distributed.l2_adapters.mock_l2_adapter import (
            MockL2Adapter,
        )

        config = MockL2AdapterConfig(
            max_size_gb=0.001,
            mock_bandwidth_gb=10.0,
        )
        adapter = create_l2_adapter(config)
        assert isinstance(adapter, MockL2Adapter)
        adapter.close()


# =========================================================
# Tests: NativePluginL2AdapterConfig factory
# =========================================================


class _FakeNativeConnector:
    """Minimal mock that satisfies the native connector
    interface check."""

    def __init__(self, **kwargs):
        self.kwargs = kwargs

    def event_fd(self):
        return -1

    def submit_batch_get(self, keys, memoryviews):
        return 0

    def submit_batch_set(self, keys, memoryviews):
        return 0

    def submit_batch_exists(self, keys):
        return 0

    def drain_completions(self):
        return []

    def close(self):
        pass


class _MissingMethodConnector:
    """Connector missing the 'close' method."""

    def event_fd(self):
        return -1

    def submit_batch_get(self, keys, memoryviews):
        return 0

    def submit_batch_set(self, keys, memoryviews):
        return 0

    def submit_batch_exists(self, keys):
        return 0

    def drain_completions(self):
        return []


class TestNativePluginFactory:
    """Tests for the native_plugin adapter factory."""

    def test_load_native_connector(self, monkeypatch):
        """Successfully load a native connector."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.native_connector_l2_adapter import (
            NativeConnectorL2Adapter,
        )
        from lmcache.v1.distributed.l2_adapters.native_plugin_l2_adapter import (
            NativePluginL2AdapterConfig,
        )

        fake_mod = types.ModuleType("fake_native_mod")
        fake_mod.FakeClient = _FakeNativeConnector  # type: ignore[attr-defined]
        monkeypatch.setattr(
            "importlib.import_module",
            lambda p: fake_mod,
        )

        cfg = NativePluginL2AdapterConfig(
            module_path="fake_native_mod",
            class_name="FakeClient",
            adapter_params={"host": "localhost"},
        )
        adapter = create_l2_adapter_from_registry(cfg)
        assert isinstance(adapter, NativeConnectorL2Adapter)
        adapter.close()

    def test_import_error_raises(self, monkeypatch):
        """ImportError propagates when module not found."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.native_plugin_l2_adapter import (
            NativePluginL2AdapterConfig,
        )

        monkeypatch.setattr(
            "importlib.import_module",
            lambda p: (_ for _ in ()).throw(ImportError("no such module")),
        )
        cfg = NativePluginL2AdapterConfig(
            module_path="nonexistent",
            class_name="X",
        )
        with pytest.raises(ImportError, match="nonexistent"):
            create_l2_adapter_from_registry(cfg)

    def test_missing_class_raises(self, monkeypatch):
        """AttributeError when class not in module."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.native_plugin_l2_adapter import (
            NativePluginL2AdapterConfig,
        )

        fake_mod = types.ModuleType("empty_mod")
        monkeypatch.setattr(
            "importlib.import_module",
            lambda p: fake_mod,
        )
        cfg = NativePluginL2AdapterConfig(
            module_path="empty_mod",
            class_name="NoSuchClass",
        )
        with pytest.raises(AttributeError, match="NoSuchClass"):
            create_l2_adapter_from_registry(cfg)

    def test_missing_method_raises(self, monkeypatch):
        """TypeError when connector is missing a required
        method."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.native_plugin_l2_adapter import (
            NativePluginL2AdapterConfig,
        )

        fake_mod = types.ModuleType("incomplete_mod")
        fake_mod.Bad = _MissingMethodConnector  # type: ignore[attr-defined]
        monkeypatch.setattr(
            "importlib.import_module",
            lambda p: fake_mod,
        )
        cfg = NativePluginL2AdapterConfig(
            module_path="incomplete_mod",
            class_name="Bad",
        )
        with pytest.raises(TypeError, match="close"):
            create_l2_adapter_from_registry(cfg)

    def test_adapter_params_forwarded(self, monkeypatch):
        """adapter_params are forwarded as kwargs to the
        connector class."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.native_connector_l2_adapter import (
            NativeConnectorL2Adapter,
        )
        from lmcache.v1.distributed.l2_adapters.native_plugin_l2_adapter import (
            NativePluginL2AdapterConfig,
        )

        fake_mod = types.ModuleType("param_mod")
        fake_mod.Client = _FakeNativeConnector  # type: ignore[attr-defined]
        monkeypatch.setattr(
            "importlib.import_module",
            lambda p: fake_mod,
        )

        cfg = NativePluginL2AdapterConfig(
            module_path="param_mod",
            class_name="Client",
            adapter_params={"host": "myhost", "port": 42},
        )
        adapter = create_l2_adapter_from_registry(cfg)
        assert isinstance(adapter, NativeConnectorL2Adapter)
        adapter.close()

    def test_native_plugin_type_registered(self):
        """'native_plugin' config type should be
        registered."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.native_plugin_l2_adapter import (
            NativePluginL2AdapterConfig,
        )

        cfg = NativePluginL2AdapterConfig(
            module_path="x",
            class_name="X",
        )
        name = get_type_name_for_config(cfg)
        assert name == "native_plugin"


# =========================================================
# Tests: FSNativeL2AdapterConfig factory
# =========================================================


class _FakeLMCacheFSClient:
    """Mock for lmcache.lmcache_fs.LMCacheFSClient."""

    def __init__(
        self,
        base_path,
        num_workers,
        relative_tmp_dir="",
        use_odirect=False,
        read_ahead_size=0,
    ):
        self.base_path = base_path
        self.num_workers = num_workers
        self.relative_tmp_dir = relative_tmp_dir
        self.use_odirect = use_odirect
        self.read_ahead_size = read_ahead_size
        self._efd = create_event_notifier()
        self._closed = False

    def event_fd(self):
        return self._efd.fileno()

    def submit_batch_get(self, keys, memoryviews):
        return 0

    def submit_batch_set(self, keys, memoryviews):
        return 0

    def submit_batch_exists(self, keys):
        return 0

    def drain_completions(self):
        return []

    def close(self):
        if not self._closed:
            self._closed = True
            self._efd.close()


class TestFSNativeAdapterFactory:
    """Tests for the fs_native adapter factory."""

    def test_factory_creates_adapter(self, monkeypatch):
        """Factory creates a NativeConnectorL2Adapter
        wrapping the FS client."""
        # First Party
        # Re-register so registry points to patched fn
        from lmcache.v1.distributed.l2_adapters.factory import (
            _L2_ADAPTER_FACTORY_REGISTRY,
        )
        from lmcache.v1.distributed.l2_adapters.fs_native_l2_adapter import (
            FSNativeL2AdapterConfig,
        )
        from lmcache.v1.distributed.l2_adapters.native_connector_l2_adapter import (
            NativeConnectorL2Adapter,
        )

        old = _L2_ADAPTER_FACTORY_REGISTRY["fs_native"]
        _L2_ADAPTER_FACTORY_REGISTRY["fs_native"] = _patched_fs_factory

        try:
            cfg = FSNativeL2AdapterConfig(
                base_path="/tmp/lmcache_test_factory",
                num_workers=2,
            )
            adapter = create_l2_adapter_from_registry(cfg)
            assert isinstance(adapter, NativeConnectorL2Adapter)
            adapter.close()
        finally:
            _L2_ADAPTER_FACTORY_REGISTRY["fs_native"] = old

    def test_factory_forwards_all_params(self, monkeypatch):
        """All config params are forwarded to the
        FS client constructor."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.fs_native_l2_adapter import (
            FSNativeL2AdapterConfig,
        )

        captured: dict = {}

        class _CaptureFSClient(_FakeLMCacheFSClient):
            def __init__(self, *args, **kwargs):
                super().__init__(*args, **kwargs)
                captured["args"] = args
                captured["kwargs"] = kwargs

        # First Party
        from lmcache.v1.distributed.l2_adapters.factory import (
            _L2_ADAPTER_FACTORY_REGISTRY,
        )

        def _capture_factory(config, l1_memory_desc=None):
            assert isinstance(config, FSNativeL2AdapterConfig)
            client = _CaptureFSClient(
                config.base_path,
                config.num_workers,
                config.relative_tmp_dir,
                config.use_odirect,
                config.read_ahead_size or 0,
            )
            # First Party
            from lmcache.v1.distributed.l2_adapters.native_connector_l2_adapter import (
                NativeConnectorL2Adapter,
            )

            return NativeConnectorL2Adapter(client)

        old = _L2_ADAPTER_FACTORY_REGISTRY["fs_native"]
        _L2_ADAPTER_FACTORY_REGISTRY["fs_native"] = _capture_factory

        try:
            cfg = FSNativeL2AdapterConfig(
                base_path="/data/kv",
                num_workers=8,
                relative_tmp_dir=".tmp",
                use_odirect=True,
                read_ahead_size=4096,
            )
            adapter = create_l2_adapter_from_registry(cfg)
            assert captured["args"] == (
                "/data/kv",
                8,
                ".tmp",
                True,
                4096,
            )
            adapter.close()
        finally:
            _L2_ADAPTER_FACTORY_REGISTRY["fs_native"] = old

    def test_factory_import_error_raises(self, monkeypatch):
        """RuntimeError raised when C++ FS extension is
        not available."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.fs_native_l2_adapter import (
            FSNativeL2AdapterConfig,
        )

        # Patch the actual factory to simulate missing
        # C++ extension
        def _broken_import(name, *args, **kwargs):
            if name == "lmcache.lmcache_fs":
                raise ImportError("no C++ FS extension")
            return _original_import(name, *args, **kwargs)

        # Standard
        import builtins

        _original_import = builtins.__import__
        monkeypatch.setattr(builtins, "__import__", _broken_import)

        cfg = FSNativeL2AdapterConfig(
            base_path="/tmp/test",
        )
        with pytest.raises(RuntimeError, match="C\\+\\+ FS"):
            create_l2_adapter_from_registry(cfg)

    def test_fs_native_type_registered(self):
        """'fs_native' config type should be
        registered."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.fs_native_l2_adapter import (
            FSNativeL2AdapterConfig,
        )

        cfg = FSNativeL2AdapterConfig(
            base_path="/tmp/test",
        )
        name = get_type_name_for_config(cfg)
        assert name == "fs_native"

    def test_fs_native_factory_registered(self):
        """'fs_native' factory should be callable via
        create_l2_adapter_from_registry (smoke test
        with RuntimeError from missing C++ ext)."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.fs_native_l2_adapter import (
            FSNativeL2AdapterConfig,
        )

        cfg = FSNativeL2AdapterConfig(
            base_path="/tmp/test_smoke",
        )
        # Without the C++ extension built, expect
        # RuntimeError
        try:
            adapter = create_l2_adapter_from_registry(cfg)
            adapter.close()
        except RuntimeError:
            pass  # Expected when C++ ext not built

    def test_read_ahead_size_none_becomes_zero(self, monkeypatch):
        """read_ahead_size=None should be passed as 0
        to the native client."""
        # First Party
        from lmcache.v1.distributed.l2_adapters.fs_native_l2_adapter import (
            FSNativeL2AdapterConfig,
        )

        captured: dict = {}

        class _CaptureFSClient2(_FakeLMCacheFSClient):
            def __init__(self, *args, **kwargs):
                super().__init__(*args, **kwargs)
                captured["read_ahead"] = args[4]

        # First Party
        from lmcache.v1.distributed.l2_adapters.factory import (
            _L2_ADAPTER_FACTORY_REGISTRY,
        )

        def _capture_factory2(config, l1_memory_desc=None):
            assert isinstance(config, FSNativeL2AdapterConfig)
            client = _CaptureFSClient2(
                config.base_path,
                config.num_workers,
                config.relative_tmp_dir,
                config.use_odirect,
                config.read_ahead_size or 0,
            )
            # First Party
            from lmcache.v1.distributed.l2_adapters.native_connector_l2_adapter import (
                NativeConnectorL2Adapter,
            )

            return NativeConnectorL2Adapter(client)

        old = _L2_ADAPTER_FACTORY_REGISTRY["fs_native"]
        _L2_ADAPTER_FACTORY_REGISTRY["fs_native"] = _capture_factory2

        try:
            cfg = FSNativeL2AdapterConfig(
                base_path="/tmp/test_ra",
            )
            assert cfg.read_ahead_size is None
            adapter = create_l2_adapter_from_registry(cfg)
            assert captured["read_ahead"] == 0
            adapter.close()
        finally:
            _L2_ADAPTER_FACTORY_REGISTRY["fs_native"] = old


def _patched_fs_factory(config, l1_memory_desc=None):
    """Factory that uses _FakeLMCacheFSClient instead
    of the real C++ extension."""
    # First Party
    from lmcache.v1.distributed.l2_adapters.fs_native_l2_adapter import (
        FSNativeL2AdapterConfig,
    )
    from lmcache.v1.distributed.l2_adapters.native_connector_l2_adapter import (
        NativeConnectorL2Adapter,
    )

    assert isinstance(config, FSNativeL2AdapterConfig)
    client = _FakeLMCacheFSClient(
        config.base_path,
        config.num_workers,
        config.relative_tmp_dir,
        config.use_odirect,
        config.read_ahead_size or 0,
    )
    return NativeConnectorL2Adapter(client)