67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
"""Flows and dashboards over HTTP, the way most tests want them there."""
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastapi.testclient import TestClient
|
|
|
|
from fluksio.core.config import settings
|
|
|
|
FLOWS = f"{settings.API_V1_STR}/flows"
|
|
DASHBOARDS = f"{settings.API_V1_STR}/dashboards"
|
|
|
|
|
|
def put_flow(
|
|
client: TestClient,
|
|
headers: dict[str, str],
|
|
definition: dict[str, Any],
|
|
flow_id: str = "",
|
|
) -> httpx.Response:
|
|
"""Create the flow under *flow_id*, its key unless given, or save over it.
|
|
|
|
Creating pins the id, so the test can address the flow by what it knows,
|
|
which needs a superuser's *headers*. Saving goes from whatever version is
|
|
stored. What a test meant by ``PUT`` before ``PUT`` stopped creating.
|
|
"""
|
|
flow_id = flow_id or definition["name"]
|
|
current = client.get(f"{FLOWS}/{flow_id}", headers=headers)
|
|
if current.status_code == 404:
|
|
return client.post(
|
|
f"{FLOWS}/", headers=headers, json={**definition, "id": flow_id}
|
|
)
|
|
version = current.json()["definition"]["version"]
|
|
return client.put(
|
|
f"{FLOWS}/{flow_id}",
|
|
headers=headers,
|
|
json={**definition, "id": flow_id, "version": version},
|
|
)
|
|
|
|
|
|
def put_dashboard(
|
|
client: TestClient,
|
|
headers: dict[str, str],
|
|
document: dict[str, Any],
|
|
dashboard_id: str = "",
|
|
) -> httpx.Response:
|
|
"""Create the dashboard under *dashboard_id*, its key unless given, and
|
|
save *document* over it, as `put_flow` does for a flow."""
|
|
dashboard_id = dashboard_id or document["name"]
|
|
current = client.get(f"{DASHBOARDS}/{dashboard_id}?draft=true", headers=headers)
|
|
if current.status_code == 404:
|
|
created = client.post(
|
|
f"{DASHBOARDS}/",
|
|
headers=headers,
|
|
json={"id": dashboard_id, "name": document["name"]},
|
|
)
|
|
assert created.status_code == 201, created.text
|
|
current = created
|
|
return client.put(
|
|
f"{DASHBOARDS}/{dashboard_id}",
|
|
headers=headers,
|
|
json={
|
|
**document,
|
|
"id": dashboard_id,
|
|
"version": current.json()["version"],
|
|
},
|
|
)
|