mxaccessgw/clients/python/tests/test_auth_options.py

"""Tests for auth metadata and connection options."""

import pytest

from zb_mom_ww_mxgateway.auth import REDACTED, ApiKey, auth_metadata, redact_secret
from zb_mom_ww_mxgateway import options as options_module
from zb_mom_ww_mxgateway.options import ClientOptions, create_channel


def test_auth_metadata_adds_bearer_api_key() -> None:
    assert auth_metadata("mxgw_test_secret") == (
        ("authorization", "Bearer mxgw_test_secret"),
    )


def test_api_key_repr_is_redacted() -> None:
    api_key = ApiKey("mxgw_test_secret")

    assert "mxgw_test_secret" not in repr(api_key)
    assert REDACTED in repr(api_key)


def test_redact_secret_replaces_known_values() -> None:
    redacted = redact_secret(
        "authorization failed for mxgw_test_secret",
        ["mxgw_test_secret"],
    )

    assert redacted == f"authorization failed for {REDACTED}"


def test_client_options_reject_plaintext_with_ca_file() -> None:
    with pytest.raises(ValueError, match="ca_file"):
        ClientOptions(
            endpoint="localhost:5000",
            plaintext=True,
            ca_file="ca.pem",
        )


def test_client_options_repr_redacts_api_key() -> None:
    options = ClientOptions(endpoint="localhost:5000", api_key="mxgw_test_secret")

    assert "mxgw_test_secret" not in repr(options)
    assert REDACTED in repr(options)


def test_create_channel_uses_plaintext_channel(monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[tuple[str, object]] = []

    def fake_insecure_channel(endpoint: str, *, options: object) -> str:
        calls.append((endpoint, options))
        return "plain-channel"

    monkeypatch.setattr(
        options_module.grpc.aio,
        "insecure_channel",
        fake_insecure_channel,
    )

    channel = create_channel(ClientOptions(endpoint="localhost:5000", plaintext=True))

    assert channel == "plain-channel"
    assert calls == [
        (
            "localhost:5000",
            [
                ("grpc.max_receive_message_length", 16 * 1024 * 1024),
                ("grpc.max_send_message_length", 16 * 1024 * 1024),
            ],
        ),
    ]


def test_create_channel_uses_tls_channel(monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[tuple[str, object, object]] = []

    def fake_credentials(*, root_certificates: object) -> str:
        assert root_certificates is None
        return "creds"

    def fake_secure_channel(endpoint: str, credentials: object, *, options: object) -> str:
        calls.append((endpoint, credentials, options))
        return "tls-channel"

    monkeypatch.setattr(
        options_module.grpc,
        "ssl_channel_credentials",
        fake_credentials,
    )
    monkeypatch.setattr(
        options_module.grpc.aio,
        "secure_channel",
        fake_secure_channel,
    )

    channel = create_channel(
        ClientOptions(
            endpoint="gateway.example:5001",
            server_name_override="gateway.test",
        ),
    )

    assert channel == "tls-channel"
    assert calls == [
            (
                "gateway.example:5001",
                "creds",
                [
                    ("grpc.max_receive_message_length", 16 * 1024 * 1024),
                    ("grpc.max_send_message_length", 16 * 1024 * 1024),
                    ("grpc.ssl_target_name_override", "gateway.test"),
                ],
            ),
        ]