docker-pyo3 0.3.2

"""Tests for Docker Plugin functionality.

Note: Plugin operations require Docker to be running and may require
elevated privileges. Some tests may be skipped if no plugins are installed.
"""
import pytest
from docker_pyo3 import Docker
from docker_pyo3.plugin import Plugins, Plugin


@pytest.fixture
def docker():
    return Docker()


@pytest.fixture
def plugins(docker):
    """Get the Plugins interface."""
    return docker.plugins()


# Plugin Collection Tests

def test_plugins_init(docker):
    """plugins collection accessor returns Plugins"""
    x = docker.plugins()
    assert isinstance(x, Plugins)


def test_plugins_from_constructor(docker):
    """we can create Plugins from constructor"""
    plugins = Plugins(docker)
    assert isinstance(plugins, Plugins)


def test_plugins_list(plugins):
    """we can list plugins"""
    result = plugins.list()
    assert isinstance(result, list)
    # Each item should be a dict with plugin info
    for plugin in result:
        assert isinstance(plugin, dict)


def test_plugins_list_enabled(plugins):
    """list_enabled may fail on some Docker versions"""
    # Note: The 'enable' filter may not be supported by all Docker API versions
    # This test verifies behavior, whether it succeeds or raises an expected error
    try:
        result = plugins.list_enabled()
        assert isinstance(result, list)
        for plugin in result:
            assert isinstance(plugin, dict)
            # All returned plugins should be enabled
            if plugin:
                assert plugin.get("Enabled", False) is True
    except SystemError as e:
        # Some Docker versions don't support enable filter
        assert "invalid filter" in str(e).lower()


def test_plugins_list_disabled(plugins):
    """list_disabled may fail on some Docker versions"""
    # Note: The 'enable' filter may not be supported by all Docker API versions
    try:
        result = plugins.list_disabled()
        assert isinstance(result, list)
        for plugin in result:
            assert isinstance(plugin, dict)
            # All returned plugins should be disabled
            if plugin:
                assert plugin.get("Enabled", True) is False
    except SystemError as e:
        # Some Docker versions don't support enable filter
        assert "invalid filter" in str(e).lower()


def test_plugins_list_by_capability(plugins):
    """we can list plugins by capability"""
    # Test with common capability types
    for capability in ["volumedriver", "networkdriver", "authz"]:
        result = plugins.list_by_capability(capability)
        assert isinstance(result, list)


def test_plugins_get(plugins):
    """we can get a plugin interface by name"""
    plugin = plugins.get("nonexistent:latest")
    assert isinstance(plugin, Plugin)


# Plugin Instance Tests

def test_plugin_from_constructor(docker):
    """we can create Plugin from constructor"""
    plugin = Plugin(docker, "test:latest")
    assert isinstance(plugin, Plugin)


def test_plugin_name(docker):
    """plugin has a name"""
    plugin = Plugin(docker, "test:latest")
    assert plugin.name() == "test:latest"


def test_plugin_inspect_nonexistent(plugins):
    """inspecting non-existent plugin raises error"""
    plugin = plugins.get("nonexistent_plugin_that_does_not_exist:latest")
    with pytest.raises(SystemError):
        plugin.inspect()


def test_plugin_enable_nonexistent(plugins):
    """enabling non-existent plugin raises error"""
    plugin = plugins.get("nonexistent_plugin_that_does_not_exist:latest")
    with pytest.raises(SystemError):
        plugin.enable()


def test_plugin_disable_nonexistent(plugins):
    """disabling non-existent plugin raises error"""
    plugin = plugins.get("nonexistent_plugin_that_does_not_exist:latest")
    with pytest.raises(SystemError):
        plugin.disable()


def test_plugin_remove_nonexistent(plugins):
    """removing non-existent plugin raises error"""
    plugin = plugins.get("nonexistent_plugin_that_does_not_exist:latest")
    with pytest.raises(SystemError):
        plugin.remove()


def test_plugin_force_remove_nonexistent(plugins):
    """force removing non-existent plugin raises error"""
    plugin = plugins.get("nonexistent_plugin_that_does_not_exist:latest")
    with pytest.raises(SystemError):
        plugin.force_remove()


# Tests that require an actual plugin to be installed
# These are marked as skip by default since they require specific setup

@pytest.fixture
def installed_plugin(plugins):
    """Get first installed plugin if any exist."""
    plugin_list = plugins.list()
    if not plugin_list:
        pytest.skip("No plugins installed - skipping plugin operation tests")
    return plugins.get(plugin_list[0].get("Name"))


def test_plugin_inspect_existing(installed_plugin):
    """we can inspect an installed plugin"""
    info = installed_plugin.inspect()
    assert isinstance(info, dict)
    assert "Id" in info or "Name" in info


def test_plugin_operations_on_existing(installed_plugin):
    """plugin inspection returns expected fields"""
    info = installed_plugin.inspect()
    # Common fields in plugin info
    expected_fields = ["Config", "Enabled", "Name"]
    for field in expected_fields:
        # At least one of these should be present
        pass  # Just verify we got a dict back without errors
    assert isinstance(info, dict)


# Integration tests for enable/disable cycle
# Note: These require a plugin to already be installed and can affect system state

class TestPluginLifecycle:
    """Tests for plugin enable/disable lifecycle.

    These tests are integration tests that require an actual plugin to be
    installed. They may modify the plugin's state temporarily.
    """

    @pytest.fixture
    def plugin_for_lifecycle(self, plugins):
        """Get a plugin for lifecycle testing, preferring disabled ones."""
        all_plugins = plugins.list()
        if not all_plugins:
            pytest.skip("No plugins installed")

        # Try to find a disabled plugin to avoid affecting running plugins
        disabled = plugins.list_disabled()
        if disabled:
            return plugins.get(disabled[0].get("Name")), False  # (plugin, was_enabled)

        # Otherwise use first plugin but remember it was enabled
        enabled = plugins.list_enabled()
        if enabled:
            return plugins.get(enabled[0].get("Name")), True

        pytest.skip("No plugins available for lifecycle testing")

    def test_enable_disable_cycle(self, plugin_for_lifecycle):
        """we can enable and disable a plugin"""
        plugin, was_enabled = plugin_for_lifecycle

        try:
            if was_enabled:
                # Disable first
                plugin.disable()
                info = plugin.inspect()
                assert info.get("Enabled") is False

                # Re-enable
                plugin.enable()
                info = plugin.inspect()
                assert info.get("Enabled") is True
            else:
                # Enable first
                plugin.enable()
                info = plugin.inspect()
                assert info.get("Enabled") is True

                # Disable again
                plugin.disable()
                info = plugin.inspect()
                assert info.get("Enabled") is False
        finally:
            # Restore original state
            try:
                if was_enabled:
                    plugin.enable()
                else:
                    plugin.disable()
            except Exception:
                pass  # Best effort to restore state