import os import unittest from unittest.mock import Mock, patch from dns_sync.adapters import ( CERTIFICATES, PROXY_HOSTS, Capability, CapabilityOperation, NormalizedObject, ObjectReference, ) 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 class NpmAdapterTests(unittest.TestCase): def config(self): with patch.dict( os.environ, { "NPM_ENDPOINTS": "192.168.1.3,192.168.1.4", "DNS_TARGET_IP": "192.168.1.101", }, clear=True, ): 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( "npm", config, config.npm_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( "npm", config, config.npm_endpoints[1], ) client_by_identity = { endpoint_id: client for endpoint_id, client in source_clients.items() } def client_factory(config, 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 ] self.assertEqual( tuple(adapter.endpoint.identity for adapter in adapters), ( "npm://192.168.1.3:81", "npm://192.168.1.4:81", ), ) self.assertEqual(client_class.call_count, 2) self.assertIs(adapters[0].client, first_client) self.assertIs(adapters[1].client, second_client) def test_capabilities_cover_discovery_and_verify_only(self): adapter = self.adapter(Mock()) self.assertEqual( adapter.capabilities, frozenset( { Capability( PROXY_HOSTS, CapabilityOperation.DISCOVER, ), Capability( PROXY_HOSTS, CapabilityOperation.VERIFY, ), Capability( CERTIFICATES, CapabilityOperation.DISCOVER, ), Capability( CERTIFICATES, CapabilityOperation.VERIFY, ), } ), ) self.assertNotIn( Capability( PROXY_HOSTS, CapabilityOperation.APPLY, ), adapter.capabilities, ) self.assertNotIn( Capability( CERTIFICATES, CapabilityOperation.APPLY, ), adapter.capabilities, ) def test_discovers_proxy_hosts_with_endpoint_local_key(self): client = Mock() client.proxy_hosts.return_value = [ { "id": 12, "domain_names": ["app.example.com"], "forward_scheme": "http", "forward_host": "192.168.1.20", "forward_port": 8080, "certificate_id": 67, "enabled": 1, "unexpected_secret": "must-not-be-copied", } ] adapter = self.adapter(client) discovered = adapter.discover(PROXY_HOSTS) self.assertEqual(len(discovered), 1) self.assertEqual(discovered[0].object_type, PROXY_HOSTS) self.assertEqual(discovered[0].local_key, "12") self.assertEqual( discovered[0].payload["domain_names"], ["app.example.com"], ) self.assertNotIn( "unexpected_secret", discovered[0].payload, ) self.assertEqual( discovered[0].references, ( ObjectReference( relation="certificate", object_type=CERTIFICATES, local_key="67", ), ), ) def test_proxy_host_without_certificate_has_no_reference(self): client = Mock() client.proxy_hosts.return_value = [ { "id": 13, "domain_names": ["plain.example.com"], "forward_scheme": "http", "forward_host": "192.168.1.20", "forward_port": 8080, "certificate_id": 0, "enabled": 1, } ] adapter = self.adapter(client) discovered = adapter.discover(PROXY_HOSTS) self.assertEqual(len(discovered), 1) self.assertEqual(discovered[0].references, ()) def test_certificate_discovery_excludes_secret_material(self): client = Mock() client.certificates.return_value = [ { "id": 67, "domain_names": ["app.example.com"], "provider": "other", "nice_name": "cluster copy", "valid_from": "2026-09-01T00:00:00Z", "expires_on": "2026-12-01T00:00:00Z", "private_key": "must-not-be-copied", "certificate": "must-not-be-copied", "meta": { "cluster_managed": True, "acme_renew": True, "dns_challenge": True, "dns_provider_credentials": "must-not-be-copied", }, } ] adapter = self.adapter(client) discovered = adapter.discover(CERTIFICATES) self.assertEqual(len(discovered), 1) certificate = discovered[0] self.assertEqual(certificate.local_key, "67") self.assertTrue(certificate.payload["cluster_managed"]) self.assertTrue(certificate.payload["acme_renew"]) self.assertNotIn("private_key", certificate.payload) self.assertNotIn("certificate", certificate.payload) self.assertNotIn( "dns_provider_credentials", certificate.payload, ) def test_verifies_proxy_host_exact_normalized_state(self): client = Mock() client.proxy_host.return_value = { "id": 12, "domain_names": ["APP.EXAMPLE.COM."], "forward_scheme": "HTTP", "forward_host": "backend.internal", "forward_port": 8080, "certificate_id": 67, "ssl_forced": True, "caching_enabled": False, "block_exploits": True, "allow_websocket_upgrade": True, "http2_support": True, "hsts_enabled": True, "hsts_subdomains": False, "enabled": True, } adapter = self.adapter(client) expected = normalize_discovered_object( adapter._proxy_host_from_row( client.proxy_host.return_value ) ) self.assertTrue(adapter.verify(expected)) client.proxy_host.assert_called_once_with("12") def test_proxy_host_verify_detects_reference_change(self): client = Mock() client.proxy_host.return_value = { "id": 12, "domain_names": ["app.example.com"], "forward_scheme": "http", "forward_host": "backend.internal", "forward_port": 8080, "certificate_id": 68, "enabled": True, } adapter = self.adapter(client) expected = NormalizedObject( object_type=PROXY_HOSTS, logical_key="app.example.com", local_key="12", payload={ "domain_names": ["app.example.com"], "forward_scheme": "http", "forward_host": "backend.internal", "forward_port": 8080, "ssl_forced": False, "caching_enabled": False, "block_exploits": False, "allow_websocket_upgrade": False, "http2_support": False, "hsts_enabled": False, "hsts_subdomains": False, "enabled": True, }, references=( ObjectReference( relation="certificate", object_type=CERTIFICATES, local_key="67", ), ), ) self.assertFalse(adapter.verify(expected)) def test_certificate_verify_uses_portable_certificate_state(self): client = Mock() client.certificate.return_value = { "id": 84, "domain_names": ["APP.EXAMPLE.COM."], # Receiving NPM owns this as a custom/imported cert. "provider": "other", "nice_name": "local imported copy", "valid_from": "2026-09-01T00:00:00Z", "expires_on": "2026-12-01T00:00:00Z", "meta": { "cluster_managed": False, "acme_renew": False, "dns_challenge": False, "certificate": "must-not-enter-normalized-state", "certificate_key": "must-not-enter-normalized-state", }, } adapter = self.adapter(client) expected = NormalizedObject( object_type=CERTIFICATES, logical_key="app.example.com", local_key="84", payload={ "domain_names": ["app.example.com"], # Source may have been Let's Encrypt. "provider": "letsencrypt", "nice_name": "source certificate", "valid_from": "2026-09-01T00:00:00+00:00", "expires_on": "2026-12-01T00:00:00+00:00", "cluster_managed": True, "acme_renew": True, "dns_challenge": True, }, ) self.assertTrue(adapter.verify(expected)) client.certificate.assert_called_once_with("84") def test_certificate_verify_detects_expiry_change(self): client = Mock() client.certificate.return_value = { "id": 84, "domain_names": ["app.example.com"], "provider": "other", "valid_from": "2026-09-01T00:00:00Z", "expires_on": "2027-01-01T00:00:00Z", "meta": {}, } adapter = self.adapter(client) expected = NormalizedObject( object_type=CERTIFICATES, logical_key="app.example.com", local_key="84", payload={ "domain_names": ["app.example.com"], "provider": "letsencrypt", "nice_name": "", "valid_from": "2026-09-01T00:00:00+00:00", "expires_on": "2026-12-01T00:00:00+00:00", "cluster_managed": False, "acme_renew": False, "dns_challenge": False, }, ) self.assertFalse(adapter.verify(expected)) def test_verify_requires_endpoint_local_key(self): adapter = self.adapter(Mock()) object = NormalizedObject( object_type=CERTIFICATES, logical_key="app.example.com", payload={ "domain_names": ["app.example.com"], }, ) self.assertFalse(adapter.verify(object)) def test_resolves_certificate_material_from_other_npm_endpoint(self): target_client = Mock() source_client = Mock() source_client.certificates.return_value = [ { "id": 10, "domain_names": ["APP.EXAMPLE.COM."], "provider": "letsencrypt", "nice_name": "source", "valid_from": "2026-09-01T00:00:00Z", "expires_on": "2026-12-01T00:00:00Z", "meta": { "cluster_managed": True, "acme_renew": True, }, } ] material = CertificateMaterial( certificate=b"CERT", certificate_key=b"KEY", intermediate_certificate=b"CHAIN", ) source_client.certificate_material.return_value = material adapter, client_factory = ( self.target_adapter_with_sources( target_client, { "npm://192.168.1.3:81": source_client, }, ) ) desired = NormalizedObject( object_type=CERTIFICATES, logical_key="app.example.com", payload={ "domain_names": ["app.example.com"], "provider": "letsencrypt", "nice_name": "source", "valid_from": "2026-09-01T00:00:00+00:00", "expires_on": "2026-12-01T00:00:00+00:00", "cluster_managed": True, "acme_renew": True, "dns_challenge": False, }, ) with patch( "dns_sync.plugins.adapters.npm.NpmClient", side_effect=client_factory, ): resolved = adapter._certificate_material_source( desired ) self.assertEqual(resolved, material) source_client.certificate_material.assert_called_once_with( "10" ) target_client.certificates.assert_not_called() target_client.certificate_material.assert_not_called() def test_certificate_material_source_ignores_wrong_expiry(self): target_client = Mock() source_client = Mock() source_client.certificates.return_value = [ { "id": 10, "domain_names": ["app.example.com"], "provider": "letsencrypt", "valid_from": "2026-09-01T00:00:00Z", "expires_on": "2027-01-01T00:00:00Z", "meta": {}, } ] adapter, client_factory = ( self.target_adapter_with_sources( target_client, { "npm://192.168.1.3:81": source_client, }, ) ) desired = NormalizedObject( object_type=CERTIFICATES, logical_key="app.example.com", payload={ "domain_names": ["app.example.com"], "valid_from": "2026-09-01T00:00:00+00:00", "expires_on": "2026-12-01T00:00:00+00:00", }, ) with patch( "dns_sync.plugins.adapters.npm.NpmClient", side_effect=client_factory, ): with self.assertRaisesRegex( RuntimeError, "No NPM source certificate material found", ): adapter._certificate_material_source(desired) 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, ), ) target_client = Mock() source_a = Mock() source_b = Mock() row_a = { "id": 10, "domain_names": ["app.example.com"], "provider": "letsencrypt", "valid_from": "2026-09-01T00:00:00Z", "expires_on": "2026-12-01T00:00:00Z", "meta": {}, } row_b = dict(row_a) row_b["id"] = 44 source_a.certificates.return_value = [row_a] source_b.certificates.return_value = [row_b] source_a.certificate_material.return_value = ( CertificateMaterial( certificate=b"CERT-A", certificate_key=b"KEY-A", ) ) source_b.certificate_material.return_value = ( CertificateMaterial( certificate=b"CERT-B", certificate_key=b"KEY-B", ) ) with patch( "dns_sync.plugins.adapters.npm.NpmClient", return_value=target_client, ): adapter = load_adapter( "npm", config, config.npm_endpoints[1], ) clients = { "npm://192.168.1.3:81": source_a, "npm://192.168.1.5:81": source_b, } def client_factory(config, endpoint): return clients[endpoint.identity] desired = NormalizedObject( object_type=CERTIFICATES, logical_key="app.example.com", payload={ "domain_names": ["app.example.com"], "valid_from": "2026-09-01T00:00:00+00:00", "expires_on": "2026-12-01T00:00:00+00:00", }, ) with patch( "dns_sync.plugins.adapters.npm.NpmClient", side_effect=client_factory, ): with self.assertRaisesRegex( RuntimeError, "Conflicting NPM certificate material", ): 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, ), ) target_client = Mock() source_a = Mock() source_b = Mock() common_material = CertificateMaterial( certificate=b"CERT", certificate_key=b"KEY", intermediate_certificate=b"CHAIN", ) source_a.certificates.return_value = [ { "id": 10, "domain_names": ["app.example.com"], "provider": "letsencrypt", "valid_from": "2026-09-01T00:00:00Z", "expires_on": "2026-12-01T00:00:00Z", "meta": {}, } ] source_b.certificates.return_value = [ { "id": 44, "domain_names": ["app.example.com"], "provider": "other", "valid_from": "2026-09-01T00:00:00Z", "expires_on": "2026-12-01T00:00:00Z", "meta": {}, } ] 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, ): adapter = load_adapter( "npm", config, config.npm_endpoints[1], ) clients = { "npm://192.168.1.3:81": source_a, "npm://192.168.1.5:81": source_b, } def client_factory(config, endpoint): return clients[endpoint.identity] desired = NormalizedObject( object_type=CERTIFICATES, logical_key="app.example.com", payload={ "domain_names": ["app.example.com"], "valid_from": "2026-09-01T00:00:00+00:00", "expires_on": "2026-12-01T00:00:00+00:00", }, ) with patch( "dns_sync.plugins.adapters.npm.NpmClient", side_effect=client_factory, ): resolved = adapter._certificate_material_source( desired ) 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, ), ) target_client = Mock() unavailable = Mock() available = Mock() unavailable.certificates.side_effect = RuntimeError( "offline" ) available.certificates.return_value = [ { "id": 44, "domain_names": ["app.example.com"], "provider": "other", "valid_from": "2026-09-01T00:00:00Z", "expires_on": "2026-12-01T00:00:00Z", "meta": {}, } ] material = CertificateMaterial( certificate=b"CERT", certificate_key=b"KEY", ) available.certificate_material.return_value = material with patch( "dns_sync.plugins.adapters.npm.NpmClient", return_value=target_client, ): adapter = load_adapter( "npm", config, config.npm_endpoints[1], ) clients = { "npm://192.168.1.3:81": unavailable, "npm://192.168.1.5:81": available, } def client_factory(config, endpoint): return clients[endpoint.identity] desired = NormalizedObject( object_type=CERTIFICATES, logical_key="app.example.com", payload={ "domain_names": ["app.example.com"], "valid_from": "2026-09-01T00:00:00+00:00", "expires_on": "2026-12-01T00:00:00+00:00", }, ) with patch( "dns_sync.plugins.adapters.npm.NpmClient", side_effect=client_factory, ): resolved = adapter._certificate_material_source( desired ) self.assertEqual(resolved, material) def test_unsupported_operations_fail_explicitly(self): adapter = self.adapter(Mock()) with self.assertRaisesRegex(RuntimeError, "cannot discover"): adapter.discover("unsupported_objects") with self.assertRaisesRegex(RuntimeError, "does not support apply"): adapter.apply(Mock()) unsupported = NormalizedObject( object_type="unsupported_objects", logical_key="unsupported", local_key="1", payload={}, ) with self.assertRaisesRegex(RuntimeError, "cannot verify"): adapter.verify(unsupported) if __name__ == "__main__": unittest.main()