Skip to content

Testing Plugins and Proxies

This page covers testing patterns for custom connection plugins, proxies, and capability interfaces.

Testing a Plugin Directly

Instantiate the plugin with a DeviceTypeDto and check its identity. To test a capability method without a device, replace require_raw_connection with a fake that answers the commands the plugin sends. These tests run against the SDK’s own FRR plugin (neops_worker_sdk/connection/plugins/frr/netmiko_plugin.py):

import copy

import pytest
from neops_workflow_engine_client import DeviceTypeDto, PlatformTypeDto

from neops_worker_sdk.connection.plugins.frr import FRRNetmikoPlugin


@pytest.fixture
def frr_device() -> DeviceTypeDto:
    return DeviceTypeDto(
        id=1,
        hostname="frr-rtr-01",
        ip="10.0.1.1",
        platform=PlatformTypeDto(short_name="frr"),
        username="admin",
        password="admin",
    )


def test_plugin_identity(frr_device: DeviceTypeDto) -> None:
    plugin = FRRNetmikoPlugin(frr_device)
    assert plugin.platform == "frr"
    assert plugin.connection_type == "ssh"
    assert plugin.connection_library == "netmiko"


FRR_OUTPUTS = {
    "/usr/bin/vtysh -c 'show running-config'": "hostname frr-rtr-01\n!",
    "cat /etc/frr/frr.conf": "hostname frr-rtr-01",
}


class FakeNetmiko:
    """Answers the commands the plugin sends, so no SSH happens."""

    def send_command(self, command: str) -> str:
        return FRR_OUTPUTS[command]


def test_plugin_reads_the_config(frr_device: DeviceTypeDto, monkeypatch: pytest.MonkeyPatch) -> None:
    plugin = FRRNetmikoPlugin(frr_device)
    monkeypatch.setattr(plugin, "require_raw_connection", FakeNetmiko)

    config = plugin.get_config()

    assert config.running == "hostname frr-rtr-01\n!"
    assert config.startup == "hostname frr-rtr-01"

For integration tests that actually connect, use a remote lab fixture to provide a real device.

Registry Isolation

Plugin tests must not leak registrations into other tests. Snapshot the registry, clear it for the test, and put the snapshot back afterwards:

from neops_worker_sdk.connection.registry import (  # noqa: E402
    _connection_type_defaults,
    _platform_defaults,
    _plugin_registry,
    clear_registry,
    get_registry_snapshot,
    register_connection_plugin,
)
from neops_worker_sdk.connection.types import BaseConnection, ConnectionPlugin  # noqa: E402


@pytest.fixture
def isolated_registry():
    """Start from an empty registry and put the previous one back afterwards."""
    snapshot = copy.deepcopy(get_registry_snapshot())
    clear_registry()
    try:
        yield
    finally:
        clear_registry()
        _plugin_registry.update(snapshot["plugins"])
        _platform_defaults.update(snapshot["platform_defaults"])
        _connection_type_defaults.update(snapshot["connection_type_defaults"])


@pytest.mark.usefixtures("isolated_registry")
def test_plugin_registers() -> None:
    @register_connection_plugin()
    class TestPlugin(ConnectionPlugin):
        platform = "test"
        connection_type = "ssh"
        connection_library = "mock"

        def _create_connection(self) -> BaseConnection:
            raise NotImplementedError

    snapshot = get_registry_snapshot()
    assert snapshot["plugins"]["test"]["ssh"]["mock"] is TestPlugin

Warning

Clearing without restoring is a leak too: every later test then runs without the SDK’s bundled plugins. Registering without clearing leaves the test plugin visible in all subsequent tests, which causes flaky AmbiguousPluginError failures that are hard to debug.

Testing Custom Capabilities

Verify that a proxy correctly raises NotImplementedForThisPlatform when the resolved plugin does not implement a method:

from neops_worker_sdk.connection.exceptions import NotImplementedForThisPlatform


def test_unimplemented_capability_raises(frr_device):
    # Assume MinimalPlugin implements NetworkDeviceLifecycleCapability but not cancel_reload()
    proxy = FullProxy.get_connection(
        frr_device, "ssh", "netmiko"
    )
    try:
        with pytest.raises(NotImplementedForThisPlatform) as exc_info:
            proxy.cancel_reload()  # not implemented

        assert exc_info.value.context.method_name == "cancel_reload"
        assert exc_info.value.context.platform == "frr"
    finally:
        proxy.close()

Integration: Proxy + Plugin Together

Test the full stack — plugin resolution, connection, and capability calls — against a mock or real device:

@pytest.mark.asyncio
async def test_proxy_resolves_correct_plugin(frr_device):
    with FRRProxy.connect(
        frr_device, "ssh", "netmiko"
    ) as proxy:
        information = proxy.get_information()
        assert information.vendor is not None

For real-device tests, use @fb_test_case_with_lab with a topology that includes the target platform.

Asserting Registration State

get_registry_snapshot() returns the full registry for inspection:

snapshot = get_registry_snapshot()
Key Type Content
plugins dict[platform, dict[type, dict[library, class]]] All registered plugins
platform_defaults dict[platform, class] Platform-level defaults
connection_type_defaults dict[(platform, type), class] Type-level defaults

Use this to verify that your plugin package registers exactly the plugins you expect, and no more.