update LocalAgent tests

docs/orchestration/execution/storage_options.md

@@ -140,6 +140,59 @@ If you do not specify a `registry_url` for your Docker Storage then the image wi
 Docker Storage uses the [Docker SDK for Python](https://docker-py.readthedocs.io/en/stable/index.html) to build the image and push to a registry. Make sure you have the Docker daemon running locally and you are configured to push to your desired container registry. Additionally make sure whichever platform Agent deploys the container also has permissions to pull from that same registry.
 :::
 
+## WebHook
+
+[WebHook Storage](/api/latest/environments/storage.html#webhook) is a storage option that stores and retrieves flows with HTTP requests. This type of storage can be used with any type of agent, and is intended to be a flexible way to integrate Prefect with your existing ecosystem, including your own file storage services.
+
+For example, the following code could be used to store flows in DropBox.
+
+```python
+from prefect import Flow
+from prefect.environments.storage import WebHook
+
+flow = Flow(
+    "dropbox-flow",
+    storage=WebHoook(
+        build_kwargs={
+            "url": "https://content.dropboxapi.com/2/files/upload",
+            "headers": {
+                "Content-Type": "application/octet-stream",
+                "Dropbox-API-Arg": json.dumps(
+                    {
+                        "path": "/Apps/prefect-test-app/dropbox-flow.flow",
+                        "mode": "overwrite",
+                        "autorename": False,
+                        "strict_conflict": True,
+                    }
+                ),
+            },
+        },
+        build_http_method="POST",
+        get_flow_kwargs={
+            "url": "https://content.dropboxapi.com/2/files/download",
+            "headers": {
+                "Accept": "application/octet-stream",
+                "Dropbox-API-Arg": json.dumps(
+                    {"path": "/Apps/prefect-test-app/dropbox-flow.flow"}
+                ),
+            },
+        },
+        get_flow_http_method="POST",
+        build_secret_config={
+            "Authorization": {"value": "DBOX_OAUTH2_TOKEN", "type": "environment"}
+        },
+    )
+)
+
+flow.storage.build()
+```
+
+The `build_secret_config` is used to resolve environment variables to fill in request headers with sensitive information. Because this resolution is at runtime, this storage option never has your sensitive information stored in it and that sensitive information is never sent to Prefect Cloud. That config supports environment variables and [Prefect secrets](/core/concepts/secrets.html).
+
+::: tip Sensible Defaults
+Flows registered with this storage option will automatically be labeled with `"webhook-flow-storage"`. Add that label to an agent to tell Prefect Cloud that that agent should run flows with `WebHook` storage.
+:::
+
 ### Non-Docker Storage for Containerized Environments
 
 Prefect allows for flows to be stored in cloud storage services and executed in containerized environments. This has the added benefit of rapidly deploying new versions of flows without having to rebuild images each time. To enable this functionality add an image name to the flow's Environment metadata.

src/prefect/agent/local/agent.py

@@ -6,7 +6,7 @@
 
 from prefect import config
 from prefect.agent import Agent
-from prefect.environments.storage import GCS, S3, Azure, Local, GitHub
+from prefect.environments.storage import GCS, S3, Azure, Local, GitHub, WebHook
 from prefect.serialization.storage import StorageSchema
 from prefect.utilities.graphql import GraphQLResult
 
@@ -91,6 +91,7 @@ def __init__(
                 "gcs-flow-storage",
                 "s3-flow-storage",
                 "github-flow-storage",
+                "webhook-flow-storage",
             ]
         )
 
@@ -125,7 +126,8 @@ def deploy_flow(self, flow_run: GraphQLResult) -> str:
         )
 
         if not isinstance(
-            StorageSchema().load(flow_run.flow.storage), (Local, Azure, GCS, S3, GitHub)
+            StorageSchema().load(flow_run.flow.storage),
+            (Local, Azure, GCS, S3, GitHub, WebHook),
         ):
             self.logger.error(
                 "Storage for flow run {} is not a supported type.".format(flow_run.id)

src/prefect/environments/storage/webhook.py

@@ -4,9 +4,6 @@
 #   until after build() is done
 # * test with Google Drive
 # * test with something custom on API Gateway
-# * test that creds work with Prefect Cloud secrets
-# * test that creds work with environment variables
-# * different secret_config for get_flow() vs. build()
 # * add to all the relevant docs
 
 import cloudpickle
@@ -82,7 +79,7 @@ class WebHook(Storage):
         ...
         build_secret_config={
             "X-Api-Key": {
-                "value": "MY_COOL_ENV_VARIABLE",
+                "name": "MY_COOL_ENV_VARIABLE",
                 "type": "environment"
             }
         }
@@ -259,10 +256,13 @@ def _render_headers(
             - headers (dict): A dictionary of headers.
             - secret_config (dict): A secret config. See `help(WebHook)` for
                 details on how this should be constructed.
+        Raises:
+            - KeyError if referencing an environment variable that does not exist
+            - ValueError if referencing a Secret that does not exist
         """
         out_headers = deepcopy(headers)
         for header, details in secret_config.items():
-            name = details["value"]
+            name = details["name"]
             if details["type"] == "environment":
                 out_headers[header] = os.environ[name]
             elif details["type"] == "secret":

tests/agent/test_local_agent.py

@@ -7,7 +7,7 @@
 from testfixtures import compare, LogCapture
 
 from prefect.agent.local import LocalAgent
-from prefect.environments.storage import Docker, Local, Azure, GCS, S3
+from prefect.environments.storage import Docker, Local, Azure, GCS, S3, WebHook
 from prefect.utilities.configuration import set_temporary_config
 from prefect.utilities.graphql import GraphQLResult
 
@@ -21,6 +21,7 @@ def test_local_agent_init(runner_token):
         "s3-flow-storage",
         "gcs-flow-storage",
         "github-flow-storage",
+        "webhook-flow-storage",
     }
     assert agent.name == "agent"
 
@@ -46,6 +47,7 @@ def test_local_agent_config_options(runner_token):
             "s3-flow-storage",
             "gcs-flow-storage",
             "github-flow-storage",
+            "webhook-flow-storage",
             "test_label",
         }
 
@@ -71,6 +73,7 @@ def test_local_agent_config_options_hostname(runner_token):
             "s3-flow-storage",
             "gcs-flow-storage",
             "github-flow-storage",
+            "webhook-flow-storage",
         }
 
 
@@ -140,6 +143,7 @@ def test_populate_env_vars(runner_token):
                     "gcs-flow-storage",
                     "s3-flow-storage",
                     "github-flow-storage",
+                    "webhook-flow-storage",
                 ]
             ),
             "PREFECT__CONTEXT__FLOW_RUN_ID": "id",
@@ -180,6 +184,7 @@ def test_populate_env_vars_includes_agent_labels(runner_token):
                     "gcs-flow-storage",
                     "s3-flow-storage",
                     "github-flow-storage",
+                    "webhook-flow-storage",
                 ]
             ),
             "PREFECT__CONTEXT__FLOW_RUN_ID": "id",
@@ -278,6 +283,31 @@ def test_local_agent_deploy_processes_azure_storage(monkeypatch, runner_token):
     assert len(agent.processes) == 1
 
 
+def test_local_agent_deploy_processes_webhook_storage(monkeypatch, runner_token):
+
+    popen = MagicMock()
+    monkeypatch.setattr("prefect.agent.local.agent.Popen", popen)
+
+    agent = LocalAgent()
+    webhook = WebHook(
+        build_kwargs={"url": "test-service/upload"},
+        build_http_method="POST",
+        get_flow_kwargs={"url": "test-service/download"},
+        get_flow_http_method="GET",
+    )
+    agent.deploy_flow(
+        flow_run=GraphQLResult(
+            {
+                "flow": GraphQLResult({"storage": webhook.serialize(), "id": "foo"}),
+                "id": "id",
+            }
+        )
+    )
+
+    assert popen.called
+    assert len(agent.processes) == 1
+
+
 def test_local_agent_deploy_storage_raises_not_supported_storage(
     monkeypatch, runner_token
 ):

tests/environments/storage/test_webhook_storage.py

@@ -6,6 +6,7 @@
 from requests.exceptions import HTTPError
 from typing import Any, Dict, Optional
 
+from prefect import context
 from prefect import task, Flow
 from prefect.environments.storage import WebHook
 
@@ -215,7 +216,77 @@ def _mock_successful_post(*args, **kwargs):
         "POST": _mock_successful_post,
     }
     webhook.add_flow(sample_flow)
-    res = webhook.build()
+    webhook.build()
 
     with pytest.raises(HTTPError, match="test-error-message"):
         webhook.get_flow()
+
+
+def test_render_headers_gets_env_variables(monkeypatch):
+    some_cred = str(uuid.uuid4())
+    another_secret = str(uuid.uuid4())
+    monkeypatch.setenv("SOME_CRED", some_cred)
+    webhook = WebHook(
+        build_kwargs={"url": "https://content.dropboxapi.com/2/files"},
+        build_http_method="POST",
+        build_secret_config={
+            "X-Api-Key": {"name": "SOME_CRED", "type": "environment"},
+            "X-Custom-Key": {"name": "ANOTHER_SECRET", "type": "secret"},
+        },
+        get_flow_kwargs={"url": "https://content.dropboxapi.com/2/files"},
+        get_flow_http_method="GET",
+    )
+
+    # set a local secret
+    context.setdefault("secrets", {})
+    context.secrets["ANOTHER_SECRET"] = another_secret
+
+    initial_headers = {"X-Api-Key": "abc"}
+    new_headers = webhook._render_headers(
+        headers=initial_headers, secret_config=webhook.build_secret_config
+    )
+
+    # _render_headers should not have side effects
+    assert initial_headers == {"X-Api-Key": "abc"}
+
+    # env variables and secrets should have been filled in
+    assert new_headers["X-Api-Key"] == some_cred
+    assert new_headers["X-Custom-Key"] == another_secret
+
+
+def test_render_headers_raises_expected_exception_on_missing_env_var(monkeypatch):
+    monkeypatch.delenv("SOME_CRED", raising=False)
+    webhook = WebHook(
+        build_kwargs={"url": "https://content.dropboxapi.com/2/files"},
+        build_http_method="POST",
+        build_secret_config={
+            "X-Api-Key": {"name": "SOME_CRED", "type": "environment"},
+        },
+        get_flow_kwargs={"url": "https://content.dropboxapi.com/2/files"},
+        get_flow_http_method="GET",
+    )
+
+    with pytest.raises(KeyError, match="SOME_CRED"):
+        initial_headers = {"X-Api-Key": "abc"}
+        webhook._render_headers(
+            headers=initial_headers, secret_config=webhook.build_secret_config
+        )
+
+
+def test_render_headers_raises_expected_exception_on_missing_secret(monkeypatch):
+    monkeypatch.delenv("ANOTHER_SECRET", raising=False)
+    webhook = WebHook(
+        build_kwargs={"url": "https://content.dropboxapi.com/2/files"},
+        build_http_method="POST",
+        build_secret_config={
+            "X-Custom-Key": {"name": "ANOTHER_SECRET", "type": "secret"},
+        },
+        get_flow_kwargs={"url": "https://content.dropboxapi.com/2/files"},
+        get_flow_http_method="GET",
+    )
+
+    with pytest.raises(ValueError, match='Local Secret "ANOTHER_SECRET" was not found'):
+        initial_headers = {"X-Api-Key": "abc"}
+        webhook._render_headers(
+            headers=initial_headers, secret_config=webhook.build_secret_config
+        )

tests/serialization/test_storage.py

@@ -321,7 +321,7 @@ def test_webhook_full_serialize():
         get_flow_kwargs={
             "url": get_url,
             "headers": {
-                "Content-Type": content_type,
+                "Accept": content_type,
                 "Dropbox-API-Arg": json.dumps({"path": test_file}),
             },
         },
@@ -350,7 +350,7 @@ def test_webhook_full_serialize():
     assert serialized["get_flow_kwargs"] == {
         "url": get_url,
         "headers": {
-            "Content-Type": content_type,
+            "Accept": content_type,
             "Dropbox-API-Arg": json.dumps({"path": test_file}),
         },
     }