dns-sync/tests/test_npm_client.py

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.certificates.material import CertificateMaterial
from dns_sync.plugins.adapters.npm.client import NpmClient


class NpmClientEndpointTests(unittest.TestCase):
    def test_client_uses_explicit_endpoint(self):
        with patch.dict(
            os.environ,
            {
                "NPM_IDENTITY": "test@example.com",
                "NPM_SECRET": "not-a-real-secret",
            },
            clear=True,
        ):
            endpoint = Endpoint(
                "npm",
                "192.168.1.3:81",
            )
            with patch(
                "dns_sync.plugins.adapters.npm.client.requests.Session.request"
            ) as request:
                client = NpmClient(endpoint)

        request.assert_not_called()
        self.assertEqual(
            client.api_url,
            "http://192.168.1.3:81/api",
        )

    def test_client_can_select_specific_endpoint(self):
        with patch.dict(
            os.environ,
            {
                "NPM_IDENTITY": "test@example.com",
                "NPM_SECRET": "not-a-real-secret",
            },
            clear=True,
        ):
            endpoint = Endpoint(
                "npm",
                "192.168.1.4:8081",
            )
            client = NpmClient(endpoint)

        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,
            {
                "NPM_IDENTITY": "test@example.com",
                "NPM_SECRET": "not-a-real-secret",
            },
            clear=True,
        ):
            endpoint = Endpoint(
                "verizon",
                "192.168.1.1:443",
            )

            with self.assertRaisesRegex(
                RuntimeError,
                "npm endpoint",
            ):
                NpmClient(endpoint)

    def test_client_rejects_invalid_api_scheme(self):
        with patch.dict(
            os.environ,
            {
                "NPM_IDENTITY": "test@example.com",
                "NPM_SECRET": "not-a-real-secret",
                "PROXY_API_SCHEME": "ftp",
            },
            clear=True,
        ):
            with self.assertRaisesRegex(
                RuntimeError,
                "PROXY_API_SCHEME",
            ):
                NpmClient(
                    Endpoint(
                        "npm",
                        "192.168.1.3:81",
                    )
                )

    def test_client_owns_tls_verification_setting(self):
        with patch.dict(
            os.environ,
            {
                "NPM_IDENTITY": "test@example.com",
                "NPM_SECRET": "not-a-real-secret",
                "PROXY_API_SCHEME": "https",
                "PROXY_VERIFY_TLS": "false",
            },
            clear=True,
        ):
            client = NpmClient(
                Endpoint(
                    "npm",
                    "192.168.1.3:81",
                )
            )

        self.assertEqual(
            client.api_url,
            "https://192.168.1.3:81/api",
        )
        self.assertFalse(client.verify_tls)


class NpmClientMutationTests(unittest.TestCase):
    def client(self):
        with patch.dict(
            os.environ,
            {
                "NPM_IDENTITY": "test@example.com",
                "NPM_SECRET": "not-a-real-secret",
            },
            clear=True,
        ):
            return NpmClient(
                Endpoint(
                    "npm",
                    "192.168.1.3:81",
                )
            )

    @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_deletes_proxy_host_by_endpoint_local_id(self):
        client = self.client()
        client.request = Mock()

        result = client.delete_proxy_host("12")

        self.assertIsNone(result)
        client.request.assert_called_once_with(
            "DELETE",
            "/nginx/proxy-hosts/12",
            expected_status=200,
        )

    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("<redacted>", 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_deletes_certificate_by_endpoint_local_id(self):
        client = self.client()
        client.request = Mock()

        result = client.delete_certificate("84")

        self.assertIsNone(result)
        client.request.assert_called_once_with(
            "DELETE",
            "/nginx/certificates/84",
            expected_status=200,
        )

    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()