openai/openai-python
Publicmirrored from https://github.com/openai/openai-pythonAvailable
tests/lib/bedrock_live.py
100lines · modecode
| 1 | """Opt-in live test for the Amazon Bedrock provider. |
| 2 | |
| 3 | This file is intentionally named ``bedrock_live.py`` so the standard pytest |
| 4 | suite does not collect it. Run it explicitly with: |
| 5 | |
| 6 | rye run pytest -q -s tests/lib/bedrock_live.py |
| 7 | |
| 8 | The test loads ``.env`` from the repository root when it exists. Set |
| 9 | ``BEDROCK_LIVE_ENV_FILE`` to load a different file. Existing environment |
| 10 | variables take precedence over values from the file. |
| 11 | |
| 12 | Useful environment variables: |
| 13 | |
| 14 | - ``BEDROCK_LIVE_MODEL`` (defaults to ``openai.gpt-oss-120b``) |
| 15 | - ``BEDROCK_LIVE_REGION`` (otherwise uses the normal AWS region chain) |
| 16 | - ``BEDROCK_LIVE_PROFILE`` (otherwise uses the normal AWS credential chain) |
| 17 | - ``AWS_BEDROCK_BASE_URL`` (required; GPT-OSS uses the ``/v1`` endpoint) |
| 18 | |
| 19 | The normal AWS environment variables, shared config, credential-process, SSO, |
| 20 | and workload credential sources are resolved by botocore. The test explicitly |
| 21 | disables the ``AWS_BEARER_TOKEN_BEDROCK`` fallback so a passing run proves that |
| 22 | AWS credentials and SigV4 signing worked. |
| 23 | """ |
| 24 | |
| 25 | from __future__ import annotations |
| 26 | |
| 27 | import os |
| 28 | import re |
| 29 | from pathlib import Path |
| 30 | |
| 31 | from openai import OpenAI |
| 32 | from openai.providers import bedrock |
| 33 | |
| 34 | _REPOSITORY_ROOT = Path(__file__).resolve().parents[2] |
| 35 | _ENV_NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") |
| 36 | |
| 37 | |
| 38 | def _load_env_file() -> None: |
| 39 | configured_path = os.environ.get("BEDROCK_LIVE_ENV_FILE") |
| 40 | path = Path(configured_path).expanduser() if configured_path else _REPOSITORY_ROOT / ".env" |
| 41 | if not path.exists(): |
| 42 | if configured_path: |
| 43 | raise RuntimeError(f"BEDROCK_LIVE_ENV_FILE does not exist: {path}") |
| 44 | return |
| 45 | |
| 46 | for line_number, raw_line in enumerate(path.read_text().splitlines(), start=1): |
| 47 | line = raw_line.strip() |
| 48 | if not line or line.startswith("#"): |
| 49 | continue |
| 50 | if line.startswith("export "): |
| 51 | line = line.removeprefix("export ").lstrip() |
| 52 | |
| 53 | name, separator, raw_value = line.partition("=") |
| 54 | name = name.strip() |
| 55 | if not separator or _ENV_NAME.fullmatch(name) is None: |
| 56 | raise RuntimeError(f"Invalid environment assignment at {path}:{line_number}") |
| 57 | |
| 58 | value = raw_value.strip() |
| 59 | if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: |
| 60 | value = value[1:-1] |
| 61 | elif " #" in value: |
| 62 | value = value.split(" #", 1)[0].rstrip() |
| 63 | |
| 64 | os.environ.setdefault(name, value) |
| 65 | |
| 66 | |
| 67 | _load_env_file() |
| 68 | |
| 69 | |
| 70 | def test_bedrock_live_response() -> None: |
| 71 | model = os.environ.get("BEDROCK_LIVE_MODEL") or "openai.gpt-oss-120b" |
| 72 | region = os.environ.get("BEDROCK_LIVE_REGION") or None |
| 73 | profile = os.environ.get("BEDROCK_LIVE_PROFILE") or None |
| 74 | base_url = os.environ.get("AWS_BEDROCK_BASE_URL") or None |
| 75 | if base_url is None: |
| 76 | raise RuntimeError( |
| 77 | "Set AWS_BEDROCK_BASE_URL to the Bedrock GPT-OSS endpoint, for example " |
| 78 | "https://bedrock-mantle.us-west-2.api.aws/v1." |
| 79 | ) |
| 80 | |
| 81 | with OpenAI( |
| 82 | provider=bedrock( |
| 83 | region=region, |
| 84 | profile=profile, |
| 85 | base_url=base_url, |
| 86 | api_key=None, |
| 87 | ), |
| 88 | timeout=60, |
| 89 | max_retries=2, |
| 90 | ) as client: |
| 91 | response = client.responses.create( |
| 92 | model=model, |
| 93 | input="Reply with exactly: bedrock live test ok", |
| 94 | store=False, |
| 95 | ) |
| 96 | |
| 97 | output_text = response.output_text.strip() |
| 98 | assert output_text, f"Bedrock returned no output text for response {response.id}" |
| 99 | assert "bedrock live test ok" in output_text.lower() |
| 100 | print(f"Bedrock live response {response.id}: {output_text}") |
| 101 | |