generated from coulomb/repo-seed
ARTIFACT-STORE-WP-0007 D7.4: STS temporary credential support (session token + refreshable file refs)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -2,7 +2,9 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from collections.abc import AsyncIterator
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
@@ -194,3 +196,55 @@ async def test_health_uses_head_bucket(backend: S3Backend) -> None:
|
||||
status = await backend.health()
|
||||
assert status.healthy is True
|
||||
assert status.backend_id == "s3"
|
||||
|
||||
|
||||
def test_client_passes_session_token(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
captured: dict[str, object] = {}
|
||||
|
||||
class FakeSession:
|
||||
def __init__(self, **kwargs: object) -> None:
|
||||
captured.update(kwargs)
|
||||
|
||||
def client(self, *args: object, **kwargs: object) -> object:
|
||||
return object()
|
||||
|
||||
monkeypatch.setitem(sys.modules, "aioboto3", SimpleNamespace(Session=FakeSession))
|
||||
backend_with_token = S3Backend(
|
||||
S3BackendConfig(
|
||||
endpoint_url="http://minio.local:9000",
|
||||
region="us-east-1",
|
||||
bucket="bucket",
|
||||
access_key_id="AKIA-temporary",
|
||||
secret_access_key="temp-secret",
|
||||
session_token="temp-session-token",
|
||||
)
|
||||
)
|
||||
backend_with_token._client()
|
||||
assert captured["aws_session_token"] == "temp-session-token"
|
||||
|
||||
|
||||
def test_credentials_provider_re_resolves_per_client(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
seen_tokens: list[object] = []
|
||||
|
||||
class FakeSession:
|
||||
def __init__(self, **kwargs: object) -> None:
|
||||
seen_tokens.append(kwargs.get("aws_session_token"))
|
||||
|
||||
def client(self, *args: object, **kwargs: object) -> object:
|
||||
return object()
|
||||
|
||||
monkeypatch.setitem(sys.modules, "aioboto3", SimpleNamespace(Session=FakeSession))
|
||||
rotation = iter(["token-1", "token-2"])
|
||||
backend_rotating = S3Backend(
|
||||
S3BackendConfig(
|
||||
endpoint_url="http://minio.local:9000",
|
||||
region="us-east-1",
|
||||
bucket="bucket",
|
||||
),
|
||||
credentials_provider=lambda: ("key", "secret", next(rotation)),
|
||||
)
|
||||
backend_rotating._client()
|
||||
backend_rotating._client()
|
||||
assert seen_tokens == ["token-1", "token-2"]
|
||||
|
||||
Reference in New Issue
Block a user