furyhawk/pydantic-ai-harness
1
0
Fork 0
mirror of https://github.com/pydantic/pydantic-ai-harness.git synced 2026-07-21 10:55:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Graduate LocalStack from experimental; use auth token, not the expired account bypass

Browse Source 
The experimental tier is retired, so move the capability to a top-level
pydantic_ai_harness/localstack submodule: drop warn_experimental, swap the
README's experimental warning for the soft stability note, add the docs/localstack.md
page (nav.json + docs-parity meta), and repoint the top-level README link.

Address whummer's PR review: LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT was a
grace-period bypass that expired 2026-04-06, and the single image now needs an
auth token to start. Stop defaulting the dead flag in the container launch, and
switch the CI job to the LOCALSTACK_AUTH_TOKEN secret (bound to the
localstack-integration environment) with LOCALSTACK_REQUIRE_AUTH_TOKEN=1 so it
runs the live container tests instead of skipping them.
This commit is contained in:
David SF
2026-07-14 13:02:24 -05:00
parent fb801af948
commit cb25642407
18 changed files with 312 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/main.yml
+8 -2
View File
@@ -156,11 +156,17 @@ jobs:
localstack-integration:
runs-on: ubuntu-latest
name: LocalStack integration tests
# The auth token secret lives in this environment; binding the job to it also
# satisfies zizmor's secrets-outside-env audit (same pattern as the release job).
environment: localstack-integration
env:
LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT: '1'
# Since LocalStack 2026.03.0 the single image requires an auth token to
# start, so the live tests need it. LOCALSTACK_REQUIRE_AUTH_TOKEN makes a
# missing token fail the job instead of silently skipping the container tests.
LOCALSTACK_AUTH_TOKEN: ${{ secrets.LOCALSTACK_AUTH_TOKEN }}
LOCALSTACK_REQUIRE_AUTH_TOKEN: '1'
LOCALSTACK_AWS_CLI: aws
LOCALSTACK_IMAGE: localstack/localstack
LOCALSTACK_SKIP_MANAGED_CONTAINERS: '1'
steps:
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
with:
README.md
+1 -1
View File
@@ -148,7 +148,7 @@ We studied leading coding agents, agent frameworks, and Claw-style assistants to
| | **Tool search** | Progressive tool discovery for large tool sets | :white_check_mark: [Pydantic AI](https://pydantic.dev/docs/ai/tools-toolsets/toolsets/#deferred-loading) | |
| | **File system** | Read, write, edit, search files with path traversal prevention | :white_check_mark: [Docs](pydantic_ai_harness/filesystem/) | [pydantic-ai-backend](https://github.com/vstorm-co/pydantic-ai-backend) (vstorm‑co) |
| | **Shell** | Execute commands with allowlists, denylists, and timeouts | :white_check_mark: [Docs](pydantic_ai_harness/shell/) | [pydantic-ai-backend](https://github.com/vstorm-co/pydantic-ai-backend) (vstorm‑co) |
| | **LocalStack** | Environment-backed AWS emulator with AWS CLI tools and optional Docker lifecycle | :construction: [Docs](pydantic_ai_harness/experimental/localstack/) | |
| | **LocalStack** | Environment-backed AWS emulator with AWS CLI tools and optional Docker lifecycle | :white_check_mark: [Docs](pydantic_ai_harness/localstack/) | |
| | **Repo context injection** | Auto-load CLAUDE.md/AGENTS.md and repo structure | :white_check_mark: [Docs](pydantic_ai_harness/context/) | [pydantic-deep](https://github.com/vstorm-co/pydantic-deepagents) (vstorm‑co) |
| | **Docs lookup** | On-demand `read_pyai_docs` tool for Pydantic AI docs | :white_check_mark: [Docs](pydantic_ai_harness/docs/) | |
| | **Verification loop** | Run tests after edits, auto-fix failures | :construction: [PR #169](https://github.com/pydantic/pydantic-ai-harness/pull/169) | |
docs/localstack.md
+215
View File
@@ -0,0 +1,215 @@
---
title: LocalStack
description: Give a Pydantic AI agent access to an emulated AWS environment through the AWS CLI, with an optional Docker-managed LocalStack container lifecycle.
---
# LocalStack
`LocalStack` gives an agent access to an emulated AWS environment, so it can
provision and exercise AWS services without touching a real account. It wires
the AWS CLI to a running [LocalStack](https://www.localstack.cloud/) instance --
injecting the endpoint, region, and credentials -- and can optionally start and
stop the LocalStack Docker container for each run.
[Source](https://github.com/pydantic/pydantic-ai-harness/tree/main/pydantic_ai_harness/localstack/)
> Import this capability from its submodule -- there is no top-level
> `pydantic_ai_harness` re-export:
>
> ```python
> from pydantic_ai_harness.localstack import LocalStack
> ```
>
> The API may change between releases. Where practical, breaking changes ship with a deprecation warning.
## The problem
Agents that build or test cloud infrastructure need somewhere to create buckets,
tables, queues, and functions. Pointing them at real AWS is slow, costs money,
risks leaking credentials, and is hard to reset between runs. LocalStack
emulates the AWS APIs locally, but wiring an agent to it means repeating the same
boilerplate: injecting the endpoint URL, supplying dummy credentials, shelling
out to the AWS CLI, and checking which services are up.
## Usage
`LocalStack` exposes AWS tooling wired to a running LocalStack instance. The
agent issues plain AWS CLI commands; the capability injects the endpoint, region,
and credentials, and adds a health check for the emulated services.
```python
from pydantic_ai import Agent
from pydantic_ai_harness.localstack import LocalStack
agent = Agent(
'anthropic:claude-sonnet-4-6',
capabilities=[LocalStack()],
)
result = agent.run_sync('Create an S3 bucket called reports and list all buckets.')
print(result.output)
```
By default the agent connects to a LocalStack instance you started separately --
for example with the
[`localstack` CLI](https://docs.localstack.cloud/aws/tooling/localstack-cli/)
(`localstack start`). The defaults match LocalStack's conventions: the edge
endpoint `http://localhost.localstack.cloud:4566` (which resolves to
`127.0.0.1`) and `test` / `test` credentials. Set `manage_container=True` to have
the capability start and stop a fresh Docker container per run.
## Tools
| Tool | Purpose |
|---|---|
| `aws_cli` | Run an AWS CLI command against LocalStack. Pass the command **without** the leading `aws` and **without** `--endpoint-url` -- both are injected. Returns labelled stdout/stderr plus an exit code on failure. |
| `localstack_health` | Query LocalStack's health endpoint and return the JSON of which services (s3, dynamodb, sqs, etc.) are available. |
Commands run as an argument vector (no shell), so shell operators and
redirection in the command string have no effect. Output is labelled with
`[stdout]` / `[stderr]` markers and an `[exit code: N]` line on non-zero exit.
When it exceeds `max_output_chars` the **tail** is kept (the head is dropped),
so errors survive truncation.
The AWS CLI can read from and write to local files through arguments such as
`--body`, `file://`, `fileb://`, and `s3 cp`. Treat this capability as both
AWS-emulator access and AWS CLI access to the process's filesystem.
## Service controls
| Field | Effect |
|---|---|
| `allowed_services` | If non-empty, only these AWS services may be used (allowlist), e.g. `['s3', 'dynamodb']`. |
| `denied_services` | These AWS services are always rejected (denylist). |
`allowed_services` and `denied_services` are mutually exclusive -- set one, not
both. The service is the first non-flag token of the command (`s3` in `s3 ls`).
!!! warning "Best-effort, not a security boundary"
These checks gate which commands the agent issues, not what it can reach.
For hard guarantees, configure LocalStack itself with the narrowest service
and IAM behavior the run needs, and run the agent under OS-level isolation.
## Managing the container
Set `manage_container=True` and the capability starts a LocalStack Docker
container for each run and stops it when the run ends, so the agent always gets a
fresh, isolated environment. Docker must be installed and running.
```python
from pydantic_ai_harness.localstack import LocalStack
LocalStack(
manage_container=True,
image='localstack/localstack',
container_env={'DEBUG': '1', 'PERSISTENCE': '1'},
startup_timeout=120.0,
)
```
The container's edge port (`4566`) is published on the host port from
`endpoint_url`, and the capability waits for the health endpoint before the run
starts, then stops the container when it ends (even if the run raises). Each run
gets its own container, so concurrent runs of one agent need distinct host ports
or an externally managed instance (`manage_container=False`).
Since LocalStack 2026.03.0 the default `localstack/localstack` image is a single
image that requires an auth token to start (a free Hobby/OSS token covers
community usage). When `LOCALSTACK_AUTH_TOKEN` is set in the current process it
is forwarded to the container automatically; a legacy `LOCALSTACK_API_KEY` value
is forwarded when no auth token is set. Auth values are forwarded through the
Docker CLI environment rather than embedded in the `docker run` arguments. To run
without any token, pin an image tag from before the account requirement, such as
a `localstack/localstack:4.x` release.
Docker-backed services such as Lambda need the Docker socket mounted, and some
services expose ports outside the gateway (LocalStack reserves `4510-4559`).
Enable those explicitly when a service you test requires them:
```python
LocalStack(manage_container=True, service_port_range='4510-4559', mount_docker_socket=True)
```
Mounting the Docker socket gives the container host-level Docker control. Keep
`mount_docker_socket=False` unless the emulated service requires it and the run
environment is already trusted.
The same lifecycle is available standalone as an async context manager:
```python
from pydantic_ai_harness.localstack import LocalStackContainer
async with LocalStackContainer(environment={'DEBUG': '1'}) as localstack:
... # talk to localstack.endpoint_url
```
## Configuration
```python
from pydantic_ai_harness.localstack import LocalStack
LocalStack(
endpoint_url='http://localhost.localstack.cloud:4566', # edge endpoint (host port reused when managed)
region='us-east-1', # region for the CLI and environment
access_key_id='test', # LocalStack accepts any value
secret_access_key='test', # LocalStack accepts any value
allowed_services=[], # allowlist (mutually exclusive with denied)
denied_services=[], # denylist
default_timeout=60.0, # seconds, per command and health check
max_output_chars=50_000, # output cap returned to the model
aws_cli_path='aws', # CLI executable (e.g. 'aws' or 'awslocal')
manage_container=False, # start/stop a Docker container per run
image='localstack/localstack', # image used when managing the container
host_address='127.0.0.1', # host address for Docker port publishing
service_port_range=None, # e.g. '4510-4559' for non-gateway service ports
mount_docker_socket=False, # required by Docker-backed services such as Lambda
container_name=None, # optional name for the managed container
container_env={}, # env vars for the managed container
docker_path='docker', # Docker executable
startup_timeout=120.0, # seconds to wait for the container to be ready
include_instructions=True, # add usage instructions to the prompt
)
```
The AWS CLI must be installed and on `PATH` (or point `aws_cli_path` at it). If
the binary is missing, `aws_cli` returns a clear error instead of aborting the
run. Set `include_instructions=False` to omit the capability's prompt text when
you supply your own.
## Agent spec (YAML/JSON)
`LocalStack` works with Pydantic AI's
[agent spec](/ai/core-concepts/agent-spec/):
```yaml
# agent.yaml
model: anthropic:claude-sonnet-4-6
capabilities:
- LocalStack:
endpoint_url: http://localhost.localstack.cloud:4566
allowed_services: ['s3', 'dynamodb', 'sqs']
```
```python
from pydantic_ai import Agent
from pydantic_ai_harness.localstack import LocalStack
agent = Agent.from_file('agent.yaml', custom_capability_types=[LocalStack])
```
Pass `custom_capability_types` so the spec loader knows how to instantiate
`LocalStack`.
## Further reading
- [LocalStack documentation](https://docs.localstack.cloud/)
- [Pydantic AI capabilities](/ai/core-concepts/capabilities/)
- [Toolsets](/ai/tools-toolsets/toolsets/)
## API reference
::: pydantic_ai_harness.localstack.LocalStack
::: pydantic_ai_harness.localstack.LocalStackToolset
::: pydantic_ai_harness.localstack.LocalStackContainer
docs/nav.json
+1
View File
@@ -19,6 +19,7 @@
{ "label": "Runtime Authoring", "slug": "runtime-authoring" },
{ "label": "Guardrails", "slug": "guardrails" },
{ "label": "Managed Prompt", "slug": "managed-prompt" },
{ "label": "LocalStack", "slug": "localstack" },
{ "label": "ACP", "slug": "acp" }
]
}
integration_tests/localstack/test_live_localstack.py
+1 -1
View File
@@ -15,7 +15,7 @@ import httpx
import pytest
from pydantic import TypeAdapter
from pydantic_ai_harness.experimental.localstack import LocalStack, LocalStackContainer, LocalStackToolset
from pydantic_ai_harness.localstack import LocalStack, LocalStackContainer, LocalStackToolset
_LOCALSTACK_INFO = TypeAdapter(dict[str, object])
_STARTUP_TIMEOUT = 240.0
pydantic_ai_harness/experimental/localstack/__init__.py
-10
View File
@@ -1,10 +0,0 @@
"""LocalStack capability: gives agents access to an emulated AWS environment."""
from pydantic_ai_harness.experimental._warn import warn_experimental
from pydantic_ai_harness.experimental.localstack._capability import LocalStack
from pydantic_ai_harness.experimental.localstack._container import LocalStackContainer, LocalStackError
from pydantic_ai_harness.experimental.localstack._toolset import LocalStackToolset
warn_experimental('localstack')
__all__ = ['LocalStack', 'LocalStackContainer', 'LocalStackError', 'LocalStackToolset']
pydantic_ai_harness/experimental/localstack/README.md → pydantic_ai_harness/localstack/README.md
+24 -25
View File
@@ -1,22 +1,18 @@
# LocalStack
> [!WARNING]
> **Experimental.** This capability lives under `pydantic_ai_harness.experimental`
> and may change or be removed in any release, without a deprecation period.
> Import it from the experimental path -- there is no top-level export:
> [!NOTE]
> Import this capability from its submodule -- there is no top-level `pydantic_ai_harness` re-export:
>
> ```python
> from pydantic_ai_harness.experimental.localstack import LocalStack
> from pydantic_ai_harness.localstack import LocalStack
> ```
>
> Importing it emits a `HarnessExperimentalWarning`. Silence all harness
> experimental warnings with `HarnessExperimentalWarning` from
> `pydantic_ai_harness.experimental`.
> The API may change between releases. Where practical, breaking changes ship with a deprecation warning.
Give an agent access to an emulated AWS environment, so it can provision and
exercise AWS services without touching a real account.
[Source](https://github.com/pydantic/pydantic-ai-harness/tree/main/pydantic_ai_harness/experimental/localstack/)
[Source](https://github.com/pydantic/pydantic-ai-harness/tree/main/pydantic_ai_harness/localstack/)
## The problem
@@ -36,7 +32,7 @@ region, and credentials, and adds a health check for the emulated services.
```python
from pydantic_ai import Agent
from pydantic_ai_harness.experimental.localstack import LocalStack
from pydantic_ai_harness.localstack import LocalStack
agent = Agent(
'anthropic:claude-sonnet-4-6',
@@ -135,14 +131,15 @@ starts and stops the container when it ends (even if the run raises). Each run
gets its own container, so concurrent runs of one agent need distinct host ports
or an externally managed instance (`manage_container=False`).
Managed containers bind to `127.0.0.1` by default. The default image is the
community `localstack/localstack` image. To run LocalStack Pro, set
`image='localstack/localstack-pro'`; if `LOCALSTACK_AUTH_TOKEN` is set in the
current process it is forwarded to the container automatically. A legacy
`LOCALSTACK_API_KEY` value is also forwarded when no auth token is set, but new
setups should use `LOCALSTACK_AUTH_TOKEN`. Auth values are forwarded through
the Docker CLI environment rather than embedded in the `docker run` command
arguments.
Managed containers bind to `127.0.0.1` by default. The default image is
`localstack/localstack`, which since LocalStack 2026.03.0 is a single image that
requires an auth token to start (a free Hobby/OSS token covers community usage).
When `LOCALSTACK_AUTH_TOKEN` is set in the current process it is forwarded to the
container automatically; a legacy `LOCALSTACK_API_KEY` value is forwarded when no
auth token is set, but new setups should use `LOCALSTACK_AUTH_TOKEN`. Auth values
are forwarded through the Docker CLI environment rather than embedded in the
`docker run` command arguments. To run without any token, pin an image tag from
before the account requirement, such as a `localstack/localstack:4.x` release.
Some AWS services expose ports outside the gateway. LocalStack reserves
`4510-4559` for those service endpoints, and Docker-backed services such as
@@ -164,7 +161,7 @@ it and the run environment is already trusted.
The same lifecycle is available standalone as an async context manager:
```python
from pydantic_ai_harness.experimental.localstack import LocalStackContainer
from pydantic_ai_harness.localstack import LocalStackContainer
async with LocalStackContainer(environment={'DEBUG': '1'}) as localstack:
... # talk to localstack.endpoint_url
@@ -205,14 +202,16 @@ you supply your own.
The live integration tests start LocalStack in Docker and then drive AWS CLI
commands through this capability. Local runs are skipped unless Docker and an
AWS CLI-compatible executable are available. The default target uses the
community `localstack/localstack` image and does not require an Auth Token.
AWS CLI-compatible executable are available, and unless an auth token is
configured -- the default `localstack/localstack` image requires
`LOCALSTACK_AUTH_TOKEN` to start (see [Managing the container](#managing-the-container)).
```bash
make integration-localstack
LOCALSTACK_AUTH_TOKEN=<token> make integration-localstack
```
To make a missing Auth Token fail instead of skip for Pro verification, set:
Set `LOCALSTACK_REQUIRE_AUTH_TOKEN=1` to make a missing token fail the run
instead of skipping the container tests (this is what CI uses):
```bash
LOCALSTACK_REQUIRE_AUTH_TOKEN=1 make integration-localstack
@@ -226,7 +225,7 @@ Optional environment variables:
| `LOCALSTACK_API_KEY` | Legacy fallback when `LOCALSTACK_AUTH_TOKEN` is absent. |
| `LOCALSTACK_AWS_CLI` | CLI executable for tests; defaults to `aws` when available, then `awslocal`. |
| `LOCALSTACK_IMAGE` | Docker image for live tests; defaults to `localstack/localstack`. |
| `LOCALSTACK_REQUIRE_AUTH_TOKEN` | Fail instead of skip when no Auth Token is configured. Use this for Pro/Auth Token verification. |
| `LOCALSTACK_REQUIRE_AUTH_TOKEN` | Fail instead of skip when no Auth Token is configured. Use this in CI so a missing token is a hard error. |
Store the token in your shell or CI secret store; do not commit it to the repo.
@@ -246,7 +245,7 @@ capabilities:
```python
from pydantic_ai import Agent
from pydantic_ai_harness.experimental.localstack import LocalStack
from pydantic_ai_harness.localstack import LocalStack
agent = Agent.from_file('agent.yaml', custom_capability_types=[LocalStack])
```
pydantic_ai_harness/localstack/__init__.py
+7
View File
@@ -0,0 +1,7 @@
"""LocalStack capability: gives agents access to an emulated AWS environment."""
from pydantic_ai_harness.localstack._capability import LocalStack
from pydantic_ai_harness.localstack._container import LocalStackContainer, LocalStackError
from pydantic_ai_harness.localstack._toolset import LocalStackToolset
__all__ = ['LocalStack', 'LocalStackContainer', 'LocalStackError', 'LocalStackToolset']
pydantic_ai_harness/experimental/localstack/_capability.py → pydantic_ai_harness/localstack/_capability.py
+2 -2
View File
@@ -9,7 +9,7 @@ from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.tools import AgentDepsT
from pydantic_ai.toolsets import AgentToolset
from pydantic_ai_harness.experimental.localstack._toolset import LocalStackToolset
from pydantic_ai_harness.localstack._toolset import LocalStackToolset
_INSTRUCTIONS = (
'You have access to an emulated AWS environment powered by LocalStack at {endpoint_url}. '
@@ -32,7 +32,7 @@ class LocalStack(AbstractCapability[AgentDepsT]):
```python
from pydantic_ai import Agent
from pydantic_ai_harness.experimental.localstack import LocalStack
from pydantic_ai_harness.localstack import LocalStack
agent = Agent('anthropic:claude-sonnet-4-6', capabilities=[LocalStack()])
result = agent.run_sync('Create an S3 bucket called reports and list all buckets.')
pydantic_ai_harness/experimental/localstack/_container.py → pydantic_ai_harness/localstack/_container.py
-2
View File
@@ -14,7 +14,6 @@ _EDGE_PORT = 4566
_HEALTH_PATH = '/_localstack/health'
_AUTH_TOKEN_ENV = 'LOCALSTACK_AUTH_TOKEN'
_LEGACY_API_KEY_ENV = 'LOCALSTACK_API_KEY'
_ACCOUNT_REQUIREMENT_ENV = 'LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT'
_DOCKER_SOCKET = '/var/run/docker.sock'
_GATEWAY_LISTEN_ENV = 'GATEWAY_LISTEN'
_GATEWAY_LISTEN = f'0.0.0.0:{_EDGE_PORT}'
@@ -127,7 +126,6 @@ class LocalStackContainer:
"""Return container env, forwarding LocalStack auth from the process when present."""
environment = dict(self._environment)
environment.setdefault(_GATEWAY_LISTEN_ENV, _GATEWAY_LISTEN)
environment.setdefault(_ACCOUNT_REQUIREMENT_ENV, '1')
if _AUTH_TOKEN_ENV in environment or _LEGACY_API_KEY_ENV in environment:
return environment
auth_token = os.environ.get(_AUTH_TOKEN_ENV)
pydantic_ai_harness/experimental/localstack/_toolset.py → pydantic_ai_harness/localstack/_toolset.py
+1 -1
View File
@@ -17,7 +17,7 @@ from pydantic_ai.tools import AgentDepsT
from pydantic_ai.toolsets import AbstractToolset, FunctionToolset
from typing_extensions import Self
from pydantic_ai_harness.experimental.localstack._container import LocalStackContainer
from pydantic_ai_harness.localstack._container import LocalStackContainer
_HEALTH_PATH = '/_localstack/health'
_DEFAULT_EDGE_PORT = 4566
tests/experimental/localstack/test_ci_workflow.py
-29
View File
@@ -1,29 +0,0 @@
from __future__ import annotations
from pathlib import Path
def test_localstack_ci_does_not_start_account_gated_service() -> None:
workflow = Path(__file__).parents[3] / '.github' / 'workflows' / 'main.yml'
lines = workflow.read_text().splitlines()
action_index = lines.index(
' - uses: LocalStack/setup-localstack@7c8a0cb3405bc58be4c8f763f812aa000bc46303 # v0.3.2'
)
action_block = lines[action_index : action_index + 4]
assert " skip-startup: 'true'" in action_block
def test_localstack_ci_does_not_advertise_an_external_endpoint() -> None:
workflow = Path(__file__).parents[3] / '.github' / 'workflows' / 'main.yml'
lines = workflow.read_text().splitlines()
assert not any('LOCALSTACK_ENDPOINT_URL:' in line for line in lines)
def test_localstack_ci_acknowledges_community_account_requirement() -> None:
workflow = Path(__file__).parents[3] / '.github' / 'workflows' / 'main.yml'
lines = workflow.read_text().splitlines()
assert " LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT: '1'" in lines
tests/experimental/localstack/__init__.py → tests/localstack/__init__.py
View File
tests/experimental/localstack/_http_server.py → tests/localstack/_http_server.py
View File
tests/localstack/test_ci_workflow.py
+34
View File
@@ -0,0 +1,34 @@
from __future__ import annotations
from pathlib import Path
def _workflow_lines() -> list[str]:
workflow = Path(__file__).parents[2] / '.github' / 'workflows' / 'main.yml'
return workflow.read_text().splitlines()
def test_localstack_ci_lets_the_capability_manage_the_container() -> None:
lines = _workflow_lines()
action_index = lines.index(
' - uses: LocalStack/setup-localstack@7c8a0cb3405bc58be4c8f763f812aa000bc46303 # v0.3.2'
)
action_block = lines[action_index : action_index + 4]
assert " skip-startup: 'true'" in action_block
def test_localstack_ci_does_not_advertise_an_external_endpoint() -> None:
lines = _workflow_lines()
assert not any('LOCALSTACK_ENDPOINT_URL:' in line for line in lines)
def test_localstack_ci_authenticates_with_an_auth_token() -> None:
lines = _workflow_lines()
# The single image requires a token since LocalStack 2026.03.0; the deprecated
# LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT bypass expired and must not return.
assert ' LOCALSTACK_AUTH_TOKEN: ${{ secrets.LOCALSTACK_AUTH_TOKEN }}' in lines
assert " LOCALSTACK_REQUIRE_AUTH_TOKEN: '1'" in lines
assert not any('LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT' in line for line in lines)
tests/experimental/localstack/test_container.py → tests/localstack/test_container.py
+12 -43
View File
@@ -14,14 +14,13 @@ from pathlib import Path
import pytest
from pydantic_ai_harness.experimental.localstack import LocalStackContainer, LocalStackError
from pydantic_ai_harness.localstack import LocalStackContainer, LocalStackError
from ._http_server import HttpResponse, http_server, unused_tcp_port
_LOCALSTACK_ENV_NAMES = (
'LOCALSTACK_AUTH_TOKEN',
'LOCALSTACK_API_KEY',
'LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT',
)
@@ -50,9 +49,9 @@ def _docker_stub(tmp_path: Path) -> tuple[str, Path]:
'#!/bin/sh\n'
f'echo "$@" >> {log}\n'
'if [ "$1" = run ]; then\n'
f' printf "env SERVICES=%s GATEWAY_LISTEN=%s LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT=%s '
f' printf "env SERVICES=%s GATEWAY_LISTEN=%s '
f'LOCALSTACK_AUTH_TOKEN=%s LOCALSTACK_API_KEY=%s\\n" '
f'"$SERVICES" "$GATEWAY_LISTEN" "$LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT" '
f'"$SERVICES" "$GATEWAY_LISTEN" '
f'"$LOCALSTACK_AUTH_TOKEN" "$LOCALSTACK_API_KEY" >> {log}\n'
' echo container-abc123\n'
'fi\n'
@@ -121,14 +120,8 @@ class TestLifecycle:
assert health.paths == ['/_localstack/health', '/_localstack/health']
assert container.container_id is None
log_text = log.read_text()
assert (
f'run -d --rm -p 127.0.0.1:{health.port}:4566 '
'-e GATEWAY_LISTEN -e LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT localstack/localstack'
) in log_text
assert (
'env SERVICES= GATEWAY_LISTEN=0.0.0.0:4566 LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT=1 '
'LOCALSTACK_AUTH_TOKEN= LOCALSTACK_API_KEY='
) in log_text
assert (f'run -d --rm -p 127.0.0.1:{health.port}:4566 -e GATEWAY_LISTEN localstack/localstack') in log_text
assert 'env SERVICES= GATEWAY_LISTEN=0.0.0.0:4566 LOCALSTACK_AUTH_TOKEN= LOCALSTACK_API_KEY=' in log_text
assert 'stop container-abc123' in log_text
async def test_run_command_includes_env_and_name(self, tmp_path: Path) -> None:
@@ -144,14 +137,8 @@ class TestLifecycle:
):
pass
log_text = log.read_text()
assert (
f'run -d --rm -p 127.0.0.1:{health.port}:4566 '
'-e SERVICES -e GATEWAY_LISTEN -e LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT --name ls img'
) in log_text
assert (
'env SERVICES=s3 GATEWAY_LISTEN=0.0.0.0:4566 LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT=1 '
'LOCALSTACK_AUTH_TOKEN= LOCALSTACK_API_KEY='
) in log_text
assert (f'run -d --rm -p 127.0.0.1:{health.port}:4566 -e SERVICES -e GATEWAY_LISTEN --name ls img') in log_text
assert 'env SERVICES=s3 GATEWAY_LISTEN=0.0.0.0:4566 LOCALSTACK_AUTH_TOKEN= LOCALSTACK_API_KEY=' in log_text
async def test_explicit_gateway_listen_environment_takes_precedence(self, tmp_path: Path) -> None:
docker, log = _docker_stub(tmp_path)
@@ -164,22 +151,9 @@ class TestLifecycle:
):
pass
assert (
'env SERVICES= GATEWAY_LISTEN=127.0.0.1:4566 LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT=1 '
'LOCALSTACK_AUTH_TOKEN= LOCALSTACK_API_KEY=' in log.read_text()
'env SERVICES= GATEWAY_LISTEN=127.0.0.1:4566 LOCALSTACK_AUTH_TOKEN= LOCALSTACK_API_KEY=' in log.read_text()
)
async def test_explicit_account_requirement_environment_takes_precedence(self, tmp_path: Path) -> None:
docker, log = _docker_stub(tmp_path)
with _localstack_env(), http_server([HttpResponse(200)]) as health:
async with LocalStackContainer(
docker_path=docker,
host_port=health.port,
environment={'LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT': '0'},
poll_interval=0.01,
):
pass
assert 'LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT=0' in log.read_text()
async def test_mounts_service_ports_and_docker_socket(self, tmp_path: Path) -> None:
docker, log = _docker_stub(tmp_path)
docker_socket = tmp_path / 'docker.sock'
@@ -220,11 +194,9 @@ class TestLifecycle:
argv_line, env_line, *_ = log.read_text().splitlines()
assert '-e LOCALSTACK_AUTH_TOKEN ' in f'{argv_line} '
assert '-e GATEWAY_LISTEN ' in f'{argv_line} '
assert '-e LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT ' in f'{argv_line} '
assert 'auth-token' not in argv_line
assert (
env_line == 'env SERVICES= GATEWAY_LISTEN=0.0.0.0:4566 LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT=1 '
'LOCALSTACK_AUTH_TOKEN=auth-token LOCALSTACK_API_KEY='
env_line == 'env SERVICES= GATEWAY_LISTEN=0.0.0.0:4566 LOCALSTACK_AUTH_TOKEN=auth-token LOCALSTACK_API_KEY='
)
async def test_forwards_legacy_api_key_when_auth_token_is_absent(self, tmp_path: Path) -> None:
@@ -235,11 +207,9 @@ class TestLifecycle:
argv_line, env_line, *_ = log.read_text().splitlines()
assert '-e LOCALSTACK_API_KEY ' in f'{argv_line} '
assert '-e GATEWAY_LISTEN ' in f'{argv_line} '
assert '-e LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT ' in f'{argv_line} '
assert 'legacy-key' not in argv_line
assert (
env_line == 'env SERVICES= GATEWAY_LISTEN=0.0.0.0:4566 LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT=1 '
'LOCALSTACK_AUTH_TOKEN= LOCALSTACK_API_KEY=legacy-key'
env_line == 'env SERVICES= GATEWAY_LISTEN=0.0.0.0:4566 LOCALSTACK_AUTH_TOKEN= LOCALSTACK_API_KEY=legacy-key'
)
async def test_explicit_auth_environment_takes_precedence(self, tmp_path: Path) -> None:
@@ -255,12 +225,11 @@ class TestLifecycle:
argv_line, env_line, *_ = log.read_text().splitlines()
assert '-e LOCALSTACK_AUTH_TOKEN ' in f'{argv_line} '
assert '-e GATEWAY_LISTEN ' in f'{argv_line} '
assert '-e LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT ' in f'{argv_line} '
assert 'process-token' not in argv_line
assert 'explicit-token' not in argv_line
assert (
env_line == 'env SERVICES= GATEWAY_LISTEN=0.0.0.0:4566 LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT=1 '
'LOCALSTACK_AUTH_TOKEN=explicit-token LOCALSTACK_API_KEY='
env_line
== 'env SERVICES= GATEWAY_LISTEN=0.0.0.0:4566 LOCALSTACK_AUTH_TOKEN=explicit-token LOCALSTACK_API_KEY='
)
async def test_readiness_timeout_stops_container(self, tmp_path: Path) -> None:
tests/experimental/localstack/test_localstack.py → tests/localstack/test_localstack.py
+5 -8
View File
@@ -16,7 +16,7 @@ from pydantic_ai.exceptions import ModelRetry
from pydantic_ai.models.test import TestModel
from pydantic_ai.usage import RunUsage
from pydantic_ai_harness.experimental.localstack import LocalStack, LocalStackError, LocalStackToolset
from pydantic_ai_harness.localstack import LocalStack, LocalStackError, LocalStackToolset
from ._http_server import HttpResponse, http_server, unused_tcp_port
@@ -289,12 +289,12 @@ class TestLocalStackCapability:
result = await agent.run('create a bucket')
assert result.output == 'done'
def test_exported_from_experimental_namespace(self) -> None:
def test_exported_from_submodule(self) -> None:
import pydantic_ai_harness
from pydantic_ai_harness.experimental.localstack import LocalStack as Exported
from pydantic_ai_harness.localstack import LocalStack as Exported
assert Exported is LocalStack
# Experimental capabilities are reached via the experimental namespace, not the package root.
# Capabilities are reached via their submodule, not the package root, so each keeps its own optional deps.
assert 'LocalStack' not in pydantic_ai_harness.__all__
@@ -328,10 +328,7 @@ class TestContainerManagement:
async with _toolset(endpoint_url=health.endpoint_url, manage_container=True, docker_path=docker):
pass
log_text = log.read_text()
assert (
f'run -d --rm -p 127.0.0.1:{health.port}:4566 '
'-e GATEWAY_LISTEN -e LOCALSTACK_ACKNOWLEDGE_ACCOUNT_REQUIREMENT localstack/localstack'
) in log_text
assert (f'run -d --rm -p 127.0.0.1:{health.port}:4566 -e GATEWAY_LISTEN localstack/localstack') in log_text
assert 'stop managed-xyz' in log_text
async def test_managed_defaults_to_edge_port_when_endpoint_has_no_port(self, tmp_path: Path) -> None:
tests/test_docs_parity.py
+1
View File
@@ -127,6 +127,7 @@ _CAPABILITY_PAGE_META = {
'planning.md': ('planning', 'Planning'),
'runtime-authoring.md': ('runtime_authoring', 'Runtime Authoring'),
'guardrails.md': ('guardrails', 'Input & Output Guardrails'),
'localstack.md': ('localstack', 'LocalStack'),
'acp.md': ('experimental/acp', 'ACP (Agent Client Protocol)'),
}