david/dns-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move NPM configuration and client behind plugin boundary

Browse Source
This commit is contained in:
Lyncoln
2026-09-20 22:50:37 -04:00
parent 332f994c26
commit c9a3898055
8 changed files with 289 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_npm_adapter.py
+120 -112
View File
@@ -16,73 +16,103 @@ from dns_sync.adapters import (
from dns_sync.certificates.material import CertificateMaterial
from dns_sync.config import AppConfig
from dns_sync.object_normalizer import normalize_discovered_object
from dns_sync.registry import load_adapter
from dns_sync.registry import (
configured_adapter_endpoints,
load_adapter,
)
class NpmAdapterTests(unittest.TestCase):
def config(self):
with patch.dict(
@staticmethod
def npm_env(
endpoints="192.168.1.3,192.168.1.4",
):
return patch.dict(
os.environ,
{
"NPM_ENDPOINTS": "192.168.1.3,192.168.1.4",
"NPM_ENDPOINTS": endpoints,
"DNS_TARGET_IP": "192.168.1.101",
},
clear=True,
):
)
def config(self):
with self.npm_env():
return AppConfig.from_env()
def adapter(self, client):
config = self.config()
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
return_value=client,
):
return load_adapter(
with self.npm_env():
config = AppConfig.from_env()
endpoints = configured_adapter_endpoints(
"npm",
config,
config.npm_endpoints[0],
)
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
return_value=client,
):
return load_adapter(
"npm",
config,
endpoints[0],
)
def target_adapter_with_sources(
self,
target_client,
source_clients,
):
config = self.config()
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
return_value=target_client,
):
adapter = load_adapter(
with self.npm_env():
config = AppConfig.from_env()
endpoints = configured_adapter_endpoints(
"npm",
config,
config.npm_endpoints[1],
)
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
return_value=target_client,
):
adapter = load_adapter(
"npm",
config,
endpoints[1],
)
client_by_identity = {
endpoint_id: client
for endpoint_id, client in source_clients.items()
}
def client_factory(config, endpoint):
def client_factory(endpoint):
return client_by_identity[endpoint.identity]
return adapter, client_factory
def test_registry_creates_one_adapter_per_endpoint(self):
config = self.config()
first_client = Mock()
second_client = Mock()
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
side_effect=[first_client, second_client],
) as client_class:
adapters = [
load_adapter("npm", config, endpoint)
for endpoint in config.npm_endpoints
]
with self.npm_env():
config = AppConfig.from_env()
endpoints = configured_adapter_endpoints(
"npm",
config,
)
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
side_effect=[first_client, second_client],
) as client_class:
adapters = [
load_adapter(
"npm",
config,
endpoint,
)
for endpoint in endpoints
]
self.assertEqual(
tuple(adapter.endpoint.identity for adapter in adapters),
@@ -517,25 +547,14 @@ class NpmAdapterTests(unittest.TestCase):
source_client.certificate_material.assert_not_called()
def test_certificate_material_source_rejects_conflicting_material(self):
config = self.config()
# Add a third endpoint for this ambiguity test.
endpoint_type = type(config.npm_endpoints[0])
third_endpoint = endpoint_type(
"npm",
"192.168.1.5:81",
)
config = __import__(
"dataclasses"
).replace(
config,
npm_endpoints=(
config.npm_endpoints[0],
config.npm_endpoints[1],
third_endpoint,
),
)
with self.npm_env(
"192.168.1.3,192.168.1.4,192.168.1.5"
):
config = AppConfig.from_env()
endpoints = configured_adapter_endpoints(
"npm",
config,
)
target_client = Mock()
source_a = Mock()
@@ -568,22 +587,25 @@ class NpmAdapterTests(unittest.TestCase):
)
)
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
return_value=target_client,
with self.npm_env(
"192.168.1.3,192.168.1.4,192.168.1.5"
):
adapter = load_adapter(
"npm",
config,
config.npm_endpoints[1],
)
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
return_value=target_client,
):
adapter = load_adapter(
"npm",
config,
endpoints[1],
)
clients = {
"npm://192.168.1.3:81": source_a,
"npm://192.168.1.5:81": source_b,
}
def client_factory(config, endpoint):
def client_factory(endpoint):
return clients[endpoint.identity]
desired = NormalizedObject(
@@ -607,24 +629,14 @@ class NpmAdapterTests(unittest.TestCase):
adapter._certificate_material_source(desired)
def test_certificate_material_source_accepts_identical_replicas(self):
config = self.config()
endpoint_type = type(config.npm_endpoints[0])
third_endpoint = endpoint_type(
"npm",
"192.168.1.5:81",
)
config = __import__(
"dataclasses"
).replace(
config,
npm_endpoints=(
config.npm_endpoints[0],
config.npm_endpoints[1],
third_endpoint,
),
)
with self.npm_env(
"192.168.1.3,192.168.1.4,192.168.1.5"
):
config = AppConfig.from_env()
endpoints = configured_adapter_endpoints(
"npm",
config,
)
target_client = Mock()
source_a = Mock()
@@ -660,22 +672,25 @@ class NpmAdapterTests(unittest.TestCase):
source_a.certificate_material.return_value = common_material
source_b.certificate_material.return_value = common_material
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
return_value=target_client,
with self.npm_env(
"192.168.1.3,192.168.1.4,192.168.1.5"
):
adapter = load_adapter(
"npm",
config,
config.npm_endpoints[1],
)
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
return_value=target_client,
):
adapter = load_adapter(
"npm",
config,
endpoints[1],
)
clients = {
"npm://192.168.1.3:81": source_a,
"npm://192.168.1.5:81": source_b,
}
def client_factory(config, endpoint):
def client_factory(endpoint):
return clients[endpoint.identity]
desired = NormalizedObject(
@@ -699,24 +714,14 @@ class NpmAdapterTests(unittest.TestCase):
self.assertEqual(resolved, common_material)
def test_certificate_material_source_tolerates_unavailable_unrelated_source(self):
config = self.config()
endpoint_type = type(config.npm_endpoints[0])
third_endpoint = endpoint_type(
"npm",
"192.168.1.5:81",
)
config = __import__(
"dataclasses"
).replace(
config,
npm_endpoints=(
config.npm_endpoints[0],
config.npm_endpoints[1],
third_endpoint,
),
)
with self.npm_env(
"192.168.1.3,192.168.1.4,192.168.1.5"
):
config = AppConfig.from_env()
endpoints = configured_adapter_endpoints(
"npm",
config,
)
target_client = Mock()
unavailable = Mock()
@@ -743,22 +748,25 @@ class NpmAdapterTests(unittest.TestCase):
)
available.certificate_material.return_value = material
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
return_value=target_client,
with self.npm_env(
"192.168.1.3,192.168.1.4,192.168.1.5"
):
adapter = load_adapter(
"npm",
config,
config.npm_endpoints[1],
)
with patch(
"dns_sync.plugins.adapters.npm.NpmClient",
return_value=target_client,
):
adapter = load_adapter(
"npm",
config,
endpoints[1],
)
clients = {
"npm://192.168.1.3:81": unavailable,
"npm://192.168.1.5:81": available,
}
def client_factory(config, endpoint):
def client_factory(endpoint):
return clients[endpoint.identity]
desired = NormalizedObject(