Antoni Nuñez Romeu
103 lines
3.7 KiB
Python
103 lines
3.7 KiB
Python
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from src.connectors.registry import ConnectorRegistry, _BUILTIN_CONNECTORS
|
|
from src.connectors.mock.mock_connector import MockConnector
|
|
from src.connectors.base import BaseInventoryConnector
|
|
from src.core.exceptions import ConnectorException
|
|
|
|
|
|
@pytest.fixture
|
|
def registry() -> ConnectorRegistry:
|
|
ConnectorRegistry._instance = None
|
|
reg = ConnectorRegistry.get_registry()
|
|
return reg
|
|
|
|
|
|
class TestConnectorRegistrySingleton:
|
|
def test_singleton(self, registry: ConnectorRegistry):
|
|
reg2 = ConnectorRegistry.get_registry()
|
|
assert reg2 is registry
|
|
|
|
def test_new_instance_after_reset(self, registry: ConnectorRegistry):
|
|
ConnectorRegistry._instance = None
|
|
reg2 = ConnectorRegistry.get_registry()
|
|
assert reg2 is not registry
|
|
|
|
|
|
class TestConnectorRegistryBuiltinTypes:
|
|
def test_builtin_connectors_include_mock(self):
|
|
assert "mock" in _BUILTIN_CONNECTORS
|
|
|
|
def test_builtin_connectors_include_farmatic(self):
|
|
assert "farmatic" in _BUILTIN_CONNECTORS
|
|
|
|
def test_builtin_connectors_include_custom_rest(self):
|
|
assert "custom_rest" in _BUILTIN_CONNECTORS
|
|
|
|
|
|
class TestConnectorRegistryGetClass:
|
|
def test_get_mock_connector_class(self, registry: ConnectorRegistry):
|
|
cls = registry.get_connector_class("mock")
|
|
assert cls is MockConnector
|
|
|
|
def test_get_unknown_type_raises(self, registry: ConnectorRegistry):
|
|
with pytest.raises(ConnectorException, match="No connector registered"):
|
|
registry.get_connector_class("nonexistent_type")
|
|
|
|
|
|
class TestConnectorRegistryCreateInstance:
|
|
def test_create_mock_instance(self, registry: ConnectorRegistry):
|
|
instance = registry.create_instance("mock")
|
|
assert isinstance(instance, BaseInventoryConnector)
|
|
assert instance.connector_type == "mock"
|
|
|
|
def test_create_instance_with_config(self, registry: ConnectorRegistry):
|
|
instance = registry.create_instance("mock", {"simulate_latency": False, "error_rate": 0.0})
|
|
assert instance._simulate_latency is False
|
|
|
|
def test_create_unknown_type_raises(self, registry: ConnectorRegistry):
|
|
with pytest.raises(ConnectorException):
|
|
registry.create_instance("nonexistent")
|
|
|
|
|
|
class TestConnectorRegistryInstanceCache:
|
|
def test_get_or_create_returns_cached(self, registry: ConnectorRegistry):
|
|
cid = uuid4()
|
|
inst1 = registry.get_or_create_instance(cid, "mock")
|
|
inst2 = registry.get_or_create_instance(cid, "mock")
|
|
assert inst1 is inst2
|
|
|
|
def test_remove_instance(self, registry: ConnectorRegistry):
|
|
cid = uuid4()
|
|
registry.get_or_create_instance(cid, "mock")
|
|
registry.remove_instance(cid)
|
|
assert cid not in registry._instances
|
|
|
|
def test_remove_nonexistent_instance_no_error(self, registry: ConnectorRegistry):
|
|
registry.remove_instance(uuid4())
|
|
|
|
|
|
class TestConnectorRegistryListTypes:
|
|
def test_list_registered_types(self, registry: ConnectorRegistry):
|
|
types = registry.list_registered_types()
|
|
assert "mock" in types
|
|
assert "farmatic" in types
|
|
assert "custom_rest" in types
|
|
|
|
def test_register_custom_type(self, registry: ConnectorRegistry):
|
|
mock = MockConnector()
|
|
mock.connector_type = "test_custom"
|
|
registry.register("test_custom", mock)
|
|
types = registry.list_registered_types()
|
|
assert "test_custom" in types
|
|
|
|
def test_unregister_type(self, registry: ConnectorRegistry):
|
|
mock = MockConnector()
|
|
mock.connector_type = "temp_type"
|
|
registry.register("temp_type", mock)
|
|
registry.unregister("temp_type")
|
|
types = registry.list_registered_types()
|
|
assert "temp_type" not in types
|