import os import tempfile import textwrap import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import patch from dns_sync.adapters import Endpoint from dns_sync.registry import ( configured_adapter_endpoints, enabled_plugins, load_adapter, ) class PluginRegistryTests(unittest.TestCase): def test_default_enabled_plugin_is_npm(self): with patch.dict(os.environ, {}, clear=True): self.assertEqual( enabled_plugins(), frozenset({"npm"}), ) def test_enabled_plugins_are_explicit(self): with patch.dict( os.environ, { "ENABLED_PLUGINS": ( "npm,verizon" ) }, clear=True, ): self.assertEqual( enabled_plugins(), frozenset( { "npm", "verizon", } ), ) def test_duplicate_enabled_plugin_is_rejected(self): with patch.dict( os.environ, { "ENABLED_PLUGINS": "npm,npm", }, clear=True, ): with self.assertRaisesRegex( RuntimeError, "duplicate plugin", ): enabled_plugins() def test_disabled_adapter_is_not_loaded(self): config = SimpleNamespace() endpoint = Endpoint( "verizon", "192.168.1.1:443", ) with patch.dict( os.environ, { "ENABLED_PLUGINS": "npm", }, clear=True, ): with self.assertRaisesRegex( RuntimeError, "is not enabled", ): load_adapter( "verizon", config, endpoint, ) def test_external_adapter_is_loaded(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) plugin = root / "example" plugin.mkdir() (plugin / "__init__.py").write_text( textwrap.dedent( """ class Plugin: name = "example" def __init__( self, config, endpoint, ): self.config = config self.endpoint = endpoint """ ) ) config = SimpleNamespace() endpoint = Endpoint( "example", "example.test:443", ) with patch.dict( os.environ, { "ENABLED_PLUGINS": ( "example" ), "DNS_SYNC_PLUGIN_PATH": ( directory ), }, clear=True, ): adapter = load_adapter( "example", config, endpoint, ) self.assertEqual( adapter.name, "example", ) self.assertEqual( adapter.endpoint, endpoint, ) def test_external_adapter_can_import_plugin_submodule(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) plugin = root / "example_package" plugin.mkdir() (plugin / "helper.py").write_text( "VALUE = 'loaded-from-submodule'\n" ) (plugin / "__init__.py").write_text( textwrap.dedent( """ from .helper import VALUE class Plugin: name = "example_package" def __init__( self, config, endpoint, ): self.config = config self.endpoint = endpoint self.value = VALUE """ ) ) config = SimpleNamespace() endpoint = Endpoint( "example_package", "example.test:443", ) with patch.dict( os.environ, { "ENABLED_PLUGINS": ( "example_package" ), "DNS_SYNC_PLUGIN_PATH": directory, }, clear=True, ): adapter = load_adapter( "example_package", config, endpoint, ) self.assertEqual( adapter.value, "loaded-from-submodule", ) def test_missing_enabled_adapter_fails_closed(self): with tempfile.TemporaryDirectory() as directory: config = SimpleNamespace() endpoint = Endpoint( "missing", "example.test:443", ) with patch.dict( os.environ, { "ENABLED_PLUGINS": "missing", "DNS_SYNC_PLUGIN_PATH": ( directory ), }, clear=True, ): with self.assertRaisesRegex( RuntimeError, "Unknown adapter plugin", ): load_adapter( "missing", config, endpoint, ) def test_builtin_adapter_supplies_configured_endpoints(self): endpoint = Endpoint( "npm", "192.168.1.3:81", ) config = SimpleNamespace() with patch.dict( os.environ, { "ENABLED_PLUGINS": "npm", "NPM_ENDPOINTS": "192.168.1.3", }, clear=True, ): endpoints = configured_adapter_endpoints( "npm", config, ) self.assertEqual( endpoints, (endpoint,), ) def test_external_adapter_cache_respects_plugin_root(self): with tempfile.TemporaryDirectory() as first_directory, \ tempfile.TemporaryDirectory() as second_directory: first_root = Path(first_directory) second_root = Path(second_directory) first_plugin = first_root / "example_cache" second_plugin = second_root / "example_cache" first_plugin.mkdir() second_plugin.mkdir() first_plugin.joinpath("__init__.py").write_text( textwrap.dedent( """ from dns_sync.adapters import Endpoint def configured_endpoints(config): return ( Endpoint( "example_cache", "first.example:443", ), ) class Plugin: name = "example_cache" def __init__( self, config, endpoint, ): self.endpoint = endpoint """ ) ) second_plugin.joinpath("__init__.py").write_text( textwrap.dedent( """ from dns_sync.adapters import Endpoint def configured_endpoints(config): return ( Endpoint( "example_cache", "second.example:443", ), ) class Plugin: name = "example_cache" def __init__( self, config, endpoint, ): self.endpoint = endpoint """ ) ) with patch.dict( os.environ, { "ENABLED_PLUGINS": "example_cache", "DNS_SYNC_PLUGIN_PATH": first_directory, }, clear=True, ): first = configured_adapter_endpoints( "example_cache", SimpleNamespace(), ) with patch.dict( os.environ, { "ENABLED_PLUGINS": "example_cache", "DNS_SYNC_PLUGIN_PATH": second_directory, }, clear=True, ): second = configured_adapter_endpoints( "example_cache", SimpleNamespace(), ) self.assertEqual( tuple(endpoint.identity for endpoint in first), ("example_cache://first.example:443",), ) self.assertEqual( tuple(endpoint.identity for endpoint in second), ("example_cache://second.example:443",), ) def test_external_adapter_supplies_configured_endpoints(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) plugin = root / "example" plugin.mkdir() (plugin / "__init__.py").write_text( textwrap.dedent( """ import os from dns_sync.adapters import Endpoint def configured_endpoints(config): address = os.environ["EXAMPLE_ENDPOINT"] return ( Endpoint("example", address), ) class Plugin: name = "example" def __init__( self, config, endpoint, ): self.config = config self.endpoint = endpoint """ ) ) with patch.dict( os.environ, { "ENABLED_PLUGINS": "example", "DNS_SYNC_PLUGIN_PATH": directory, "EXAMPLE_ENDPOINT": "example.test:9443", }, clear=True, ): endpoints = configured_adapter_endpoints( "example", SimpleNamespace(), ) self.assertEqual( tuple( endpoint.identity for endpoint in endpoints ), ("example://example.test:9443",), ) def test_adapter_missing_endpoint_hook_fails_closed(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) plugin = root / "example_missing_hook" plugin.mkdir() (plugin / "__init__.py").write_text( textwrap.dedent( """ class Plugin: name = "example_missing_hook" def __init__( self, config, endpoint, ): self.config = config self.endpoint = endpoint """ ) ) with patch.dict( os.environ, { "ENABLED_PLUGINS": "example_missing_hook", "DNS_SYNC_PLUGIN_PATH": directory, }, clear=True, ): with self.assertRaisesRegex( RuntimeError, "does not export configured_endpoints", ): configured_adapter_endpoints( "example_missing_hook", SimpleNamespace(), ) def test_adapter_endpoint_hook_cannot_cross_plugin_boundary(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) plugin = root / "example_wrong_endpoint" plugin.mkdir() (plugin / "__init__.py").write_text( textwrap.dedent( """ from dns_sync.adapters import Endpoint def configured_endpoints(config): return ( Endpoint( "other", "example.test:443", ), ) class Plugin: name = "example_wrong_endpoint" def __init__( self, config, endpoint, ): self.config = config self.endpoint = endpoint """ ) ) with patch.dict( os.environ, { "ENABLED_PLUGINS": "example_wrong_endpoint", "DNS_SYNC_PLUGIN_PATH": directory, }, clear=True, ): with self.assertRaisesRegex( RuntimeError, "returned endpoint for plugin", ): configured_adapter_endpoints( "example_wrong_endpoint", SimpleNamespace(), ) if __name__ == "__main__": unittest.main()