Joseph Doherty
104 lines
2.9 KiB
Python
104 lines
2.9 KiB
Python
"""Tests for auth metadata and connection options."""
|
|
|
|
import pytest
|
|
|
|
from mxgateway.auth import REDACTED, ApiKey, auth_metadata, redact_secret
|
|
from mxgateway import options as options_module
|
|
from 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", [])]
|
|
|
|
|
|
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.ssl_target_name_override", "gateway.test")],
|
|
),
|
|
]
|