import os import unittest from io import BytesIO from zipfile import ZipFile from unittest.mock import Mock, patch from dns_sync.adapters import Endpoint from dns_sync.config import AppConfig from dns_sync.certificates.material import CertificateMaterial from dns_sync.npm_client import NpmClient class NpmClientEndpointTests(unittest.TestCase): def config(self): return AppConfig.from_env() def test_default_client_uses_first_configured_endpoint(self): with patch.dict( os.environ, { "NPM_ENDPOINTS": "192.168.1.3,192.168.1.4:8081", "DNS_TARGET_IP": "192.168.1.101", "NPM_IDENTITY": "test@example.com", "NPM_SECRET": "not-a-real-secret", }, clear=True, ): config = self.config() with patch( "dns_sync.npm_client.requests.Session.request" ) as request: client = NpmClient(config) request.assert_not_called() self.assertEqual(client.api_url, "http://192.168.1.3:81/api") def test_client_can_select_specific_configured_endpoint(self): with patch.dict( os.environ, { "NPM_ENDPOINTS": "192.168.1.3,192.168.1.4:8081", "DNS_TARGET_IP": "192.168.1.101", "NPM_IDENTITY": "test@example.com", "NPM_SECRET": "not-a-real-secret", }, clear=True, ): config = self.config() client = NpmClient(config, config.npm_endpoints[1]) self.assertEqual( client.api_url, "http://192.168.1.4:8081/api", ) self.assertEqual( client.endpoint.identity, "npm://192.168.1.4:8081", ) def test_client_rejects_non_npm_endpoint(self): with patch.dict( os.environ, { "PROXY_IP": "192.168.1.101", "NPM_IDENTITY": "test@example.com", "NPM_SECRET": "not-a-real-secret", }, clear=True, ): config = self.config() endpoint = Endpoint("verizon", "192.168.1.1") with self.assertRaisesRegex(RuntimeError, "npm endpoint"): NpmClient(config, endpoint) class NpmClientMutationTests(unittest.TestCase): def client(self): with patch.dict( os.environ, { "NPM_ENDPOINTS": "192.168.1.3", "DNS_TARGET_IP": "192.168.1.101", "NPM_IDENTITY": "test@example.com", "NPM_SECRET": "not-a-real-secret", }, clear=True, ): config = AppConfig.from_env() return NpmClient( config, config.npm_endpoints[0], ) @staticmethod def response(data): response = Mock() response.json.return_value = data return response @staticmethod def certificate_zip(): buffer = BytesIO() with ZipFile(buffer, "w") as archive: archive.writestr( "cert4.pem", b"LEAF-CERTIFICATE", ) archive.writestr( "fullchain4.pem", b"FULLCHAIN", ) archive.writestr( "chain4.pem", b"INTERMEDIATE-CHAIN", ) archive.writestr( "privkey4.pem", b"PRIVATE-KEY", ) return buffer.getvalue() def test_gets_one_proxy_host_by_endpoint_local_id(self): client = self.client() client.get_json = Mock( return_value={ "id": 12, "domain_names": ["app.example.com"], } ) result = client.proxy_host("12") client.get_json.assert_called_once_with( "/nginx/proxy-hosts/12" ) self.assertEqual(result["id"], 12) def test_creates_proxy_host_with_json_payload(self): client = self.client() payload = { "domain_names": ["app.example.com"], "forward_scheme": "http", "forward_host": "192.168.1.20", "forward_port": 8080, "certificate_id": 67, } client.request = Mock( return_value=self.response({"id": 12}) ) result = client.create_proxy_host(payload) client.request.assert_called_once_with( "POST", "/nginx/proxy-hosts", expected_status=201, json=payload, ) self.assertEqual(result["id"], 12) def test_updates_proxy_host_with_json_payload(self): client = self.client() payload = { "forward_host": "192.168.1.30", } client.request = Mock( return_value=self.response({"id": 12}) ) result = client.update_proxy_host( "12", payload, ) client.request.assert_called_once_with( "PUT", "/nginx/proxy-hosts/12", expected_status=200, json=payload, ) self.assertEqual(result["id"], 12) def test_gets_one_certificate_by_endpoint_local_id(self): client = self.client() client.get_json = Mock( return_value={ "id": 67, "domain_names": ["app.example.com"], } ) result = client.certificate("67") client.get_json.assert_called_once_with( "/nginx/certificates/67" ) self.assertEqual(result["id"], 67) def test_reads_custom_certificate_material_from_metadata(self): client = self.client() client.certificate = Mock( return_value={ "id": 67, "provider": "other", "meta": { "certificate": "CUSTOM-CERTIFICATE", "certificate_key": "CUSTOM-PRIVATE-KEY", "intermediate_certificate": "CUSTOM-CHAIN", }, } ) material = client.certificate_material("67") self.assertEqual( material, CertificateMaterial( certificate=b"CUSTOM-CERTIFICATE", certificate_key=b"CUSTOM-PRIVATE-KEY", intermediate_certificate=b"CUSTOM-CHAIN", ), ) client.certificate.assert_called_once_with("67") def test_reads_letsencrypt_material_from_download_zip(self): client = self.client() client.certificate = Mock( return_value={ "id": 10, "provider": "letsencrypt", } ) response = Mock() response.content = self.certificate_zip() client.request = Mock(return_value=response) material = client.certificate_material("10") self.assertEqual( material.certificate, b"LEAF-CERTIFICATE", ) self.assertEqual( material.certificate_key, b"PRIVATE-KEY", ) self.assertEqual( material.intermediate_certificate, b"INTERMEDIATE-CHAIN", ) client.request.assert_called_once_with( "GET", "/nginx/certificates/10/download", expected_status=200, ) def test_certificate_material_repr_redacts_secrets(self): material = CertificateMaterial( certificate=b"VISIBLE-CERT-SECRET", certificate_key=b"VISIBLE-KEY-SECRET", intermediate_certificate=b"VISIBLE-CHAIN-SECRET", ) rendered = repr(material) self.assertNotIn( "VISIBLE-CERT-SECRET", rendered, ) self.assertNotIn( "VISIBLE-KEY-SECRET", rendered, ) self.assertNotIn( "VISIBLE-CHAIN-SECRET", rendered, ) self.assertIn("", rendered) def test_certificate_material_rejects_missing_secret_parts(self): client = self.client() client.certificate = Mock( return_value={ "id": 67, "provider": "other", "meta": { "certificate": "CERT", "certificate_key": "", }, } ) with self.assertRaisesRegex( RuntimeError, "certificate_key", ): client.certificate_material("67") def test_certificate_material_rejects_unknown_provider(self): client = self.client() client.certificate = Mock( return_value={ "id": 99, "provider": "mystery-provider", } ) with self.assertRaisesRegex( RuntimeError, "Unsupported NPM certificate provider", ): client.certificate_material("99") def test_creates_certificate_metadata_with_json_payload(self): client = self.client() payload = { "provider": "other", "nice_name": "cluster copy", "domain_names": ["app.example.com"], "meta": { "cluster_managed": True, }, } client.request = Mock( return_value=self.response({"id": 84}) ) result = client.create_certificate(payload) client.request.assert_called_once_with( "POST", "/nginx/certificates", expected_status=201, json=payload, ) self.assertEqual(result["id"], 84) def test_uploads_certificate_material_as_multipart(self): client = self.client() response = self.response( { "certificate": "SECRET-CERTIFICATE", "certificate_key": "SECRET-PRIVATE-KEY", } ) client.request = Mock(return_value=response) result = client.upload_certificate( "84", certificate=b"CERTIFICATE", certificate_key=b"PRIVATE-KEY", intermediate_certificate=b"INTERMEDIATE", ) client.request.assert_called_once_with( "POST", "/nginx/certificates/84/upload", expected_status=200, files={ "certificate": ( "certificate.pem", b"CERTIFICATE", ), "certificate_key": ( "certificate_key.pem", b"PRIVATE-KEY", ), "intermediate_certificate": ( "intermediate_certificate.pem", b"INTERMEDIATE", ), }, ) self.assertIsNone(result) response.json.assert_not_called() def test_certificate_upload_requires_both_required_parts(self): client = self.client() with self.assertRaisesRegex( RuntimeError, "requires certificate data", ): client.upload_certificate( "84", certificate=b"", certificate_key=b"PRIVATE-KEY", ) with self.assertRaisesRegex( RuntimeError, "requires certificate key data", ): client.upload_certificate( "84", certificate=b"CERTIFICATE", certificate_key=b"", ) def test_rejects_invalid_endpoint_local_ids(self): client = self.client() for local_id in ("", "abc", True, -1): with self.subTest(local_id=local_id): with self.assertRaisesRegex( RuntimeError, "id is invalid", ): client.certificate(local_id) if __name__ == "__main__": unittest.main()