Joseph Doherty
ddad573b75
- Resolve 14 conflicts from popping local stash on top of origin'seed1e88+8d3352fdoc-comment additions (11 mechanical, plus version.rs, DashboardAuthenticatorTests.cs, DashboardGalaxyProjector.cs) - Fix 4 test files that used AGENTS.md as the repo-root sentinel (now use CLAUDE.md, since AGENTS.md was removed in4731ab5) - Redirect 10 doc citations from AGENTS.md to the matching gateway.md sections (Value Model, Status Model, Security, STA Worker Thread Model, gRPC Layer rule, cancellation rule) Verified: solution build clean, x86 worker build clean, 266/266 gateway tests passing, 121/121 worker tests passing. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
116 lines
3.3 KiB
Python
116 lines
3.3 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",
|
|
[
|
|
("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"),
|
|
],
|
|
),
|
|
]
|