KG
09/08/2025, 5:56 PMMarvin
09/08/2025, 5:56 PMMarvin
09/08/2025, 5:57 PMimport requests
import responses
from prefect import flow
from prefect.blocks.system import Secret
# A simple flow that loads a secret and calls an external API with it
@flow
def call_protected_endpoint() -> int:
token = Secret.load("my-api-token").get()
resp = requests.get(
"<https://api.example.com/protected>",
headers={"Authorization": f"Bearer {token}"},
timeout=10,
)
resp.raise_for_status()
return resp.status_code
@responses.activate
def test_secret_auth_success(monkeypatch):
# If you have a Prefect API (Cloud/Server) available, you can save the block:
# Secret(value="TEST_TOKEN").save("my-api-token", overwrite=True)
#
# If you don't have a Prefect API running in tests, patch the load instead:
monkeypatch.setattr(
Secret,
"load",
staticmethod(lambda name: Secret(value="TEST_TOKEN"))
)
# Mock the protected endpoint to require the expected auth header
def auth_check(request):
auth = request.headers.get("Authorization")
if auth == "Bearer TEST_TOKEN":
return (200, {}, "OK")
return (401, {}, "Unauthorized")
responses.add_callback(
responses.GET,
"<https://api.example.com/protected>",
callback=auth_check,
content_type="text/plain",
)
# Run the flow and assert we authenticated successfully
status = call_protected_endpoint()
assert status == 200
Notes:
- If you want to test against a real Secret block saved in your workspace and a real endpoint, remove the monkeypatch
and ensure a Secret named my-api-token
exists with a valid value. Saving requires a running Prefect API (Prefect Cloud or a local Prefect Server).
- Install deps:
- pip install pytest responses requests prefect
- Run: pytest -q
Docs for reference:
- Blocks and Secrets overview: https://docs.prefect.io/latest/concepts/blocks/