mirror of
https://github.com/hanzoai/flow.git
synced 2026-10-17 01:29:18 +00:00
zeekay
d650dd60f0
Docs corpus (Langflow->Hanzo Flow, upstream URLs langflow-ai/langflow->hanzoai/ flow, docs.langflow.org->docs.hanzo.ai), starter-project template help-URLs, CI workflows, and test files. Remaining 'langflow' repo-wide is now dominated by LANGFLOW_* env-var names + langflow_ settings fields — the config-boundary layer deliberately preserved pending a coordinated FLOW_/HANZO_ alias-with-fallback (renaming them blind breaks every deployment).
181 lines
6.7 KiB
Python
181 lines
6.7 KiB
Python
"""Tests for gp_client.py — HMAC auth and API operations."""
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import importlib
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import gp_client
|
|
import pytest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _expected_signature(password: str, message: str) -> str:
|
|
sig = hmac.new(
|
|
bytes(password, "ISO-8859-1"),
|
|
msg=bytes(message, "ISO-8859-1"),
|
|
digestmod=hashlib.sha1,
|
|
).digest()
|
|
return base64.b64encode(sig).decode()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_headers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetHeaders:
|
|
def test_get_request_authorization_format(self):
|
|
with patch.dict("os.environ", {"GP_ADMIN_USER_ID": "user123", "GP_ADMIN_PASSWORD": "secret"}):
|
|
importlib.reload(gp_client)
|
|
headers = gp_client.get_headers("https://example.com/api", "GET")
|
|
|
|
assert headers["Authorization"].startswith("GP-HMAC user123:")
|
|
assert "GP-Date" in headers
|
|
assert headers["accept"] == "application/json"
|
|
assert "Content-Type" not in headers
|
|
|
|
def test_put_request_includes_content_type(self):
|
|
with patch.dict("os.environ", {"GP_ADMIN_USER_ID": "user123", "GP_ADMIN_PASSWORD": "secret"}):
|
|
importlib.reload(gp_client)
|
|
headers = gp_client.get_headers("https://example.com/api", "PUT", {"key": "val"})
|
|
|
|
assert headers["Content-Type"] == "application/json"
|
|
|
|
def test_patch_request_uses_merge_patch_content_type(self):
|
|
with patch.dict("os.environ", {"GP_ADMIN_USER_ID": "user123", "GP_ADMIN_PASSWORD": "secret"}):
|
|
importlib.reload(gp_client)
|
|
headers = gp_client.get_headers("https://example.com/api", "PATCH", {})
|
|
|
|
assert headers["Content-Type"] == "application/merge-patch+json"
|
|
|
|
def test_hmac_signature_is_deterministic_for_same_inputs(self):
|
|
with patch.dict("os.environ", {"GP_ADMIN_USER_ID": "user123", "GP_ADMIN_PASSWORD": "secret"}):
|
|
importlib.reload(gp_client)
|
|
with patch("gp_client.datetime") as mock_dt:
|
|
mock_dt.now.return_value.strftime.return_value = "Thu, 26 Mar 2026 12:00:00 UTC"
|
|
h1 = gp_client.get_headers("https://example.com", "GET")
|
|
h2 = gp_client.get_headers("https://example.com", "GET")
|
|
|
|
assert h1["Authorization"] == h2["Authorization"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list_bundles
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListBundles:
|
|
def test_returns_parsed_json_on_success(self):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"bundleIds": ["flow-ui"]}
|
|
|
|
with (
|
|
patch.dict("os.environ", {"GP_ADMIN_USER_ID": "u", "GP_ADMIN_PASSWORD": "p"}),
|
|
patch("gp_client.requests.get", return_value=mock_response) as mock_get,
|
|
):
|
|
importlib.reload(gp_client)
|
|
result = gp_client.list_bundles()
|
|
|
|
mock_get.assert_called_once()
|
|
mock_response.raise_for_status.assert_called_once()
|
|
assert result == {"bundleIds": ["flow-ui"]}
|
|
|
|
def test_raises_on_http_error(self):
|
|
import requests as req
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status.side_effect = req.exceptions.HTTPError("401 Unauthorized")
|
|
|
|
with (
|
|
patch.dict("os.environ", {"GP_ADMIN_USER_ID": "u", "GP_ADMIN_PASSWORD": "p"}),
|
|
patch("gp_client.requests.get", return_value=mock_response),
|
|
):
|
|
importlib.reload(gp_client)
|
|
with pytest.raises(req.exceptions.HTTPError):
|
|
gp_client.list_bundles()
|
|
|
|
def test_raises_on_timeout(self):
|
|
import requests as req
|
|
|
|
with (
|
|
patch.dict("os.environ", {"GP_ADMIN_USER_ID": "u", "GP_ADMIN_PASSWORD": "p"}),
|
|
patch("gp_client.requests.get", side_effect=req.exceptions.Timeout),
|
|
):
|
|
importlib.reload(gp_client)
|
|
with pytest.raises(req.exceptions.Timeout):
|
|
gp_client.list_bundles()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# upload_strings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUploadStrings:
|
|
def test_successful_upload(self):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"status": "ok"}
|
|
|
|
with (
|
|
patch.dict("os.environ", {"GP_ADMIN_USER_ID": "u", "GP_ADMIN_PASSWORD": "p"}),
|
|
patch("gp_client.requests.put", return_value=mock_response) as mock_put,
|
|
):
|
|
importlib.reload(gp_client)
|
|
result = gp_client.upload_strings({"hello": "Hello"})
|
|
|
|
mock_put.assert_called_once()
|
|
assert result == {"status": "ok"}
|
|
|
|
def test_raises_on_server_error(self):
|
|
import requests as req
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status.side_effect = req.exceptions.HTTPError("500")
|
|
|
|
with (
|
|
patch.dict("os.environ", {"GP_ADMIN_USER_ID": "u", "GP_ADMIN_PASSWORD": "p"}),
|
|
patch("gp_client.requests.put", return_value=mock_response),
|
|
):
|
|
importlib.reload(gp_client)
|
|
with pytest.raises(req.exceptions.HTTPError):
|
|
gp_client.upload_strings({"hello": "Hello"})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_strings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetStrings:
|
|
def test_returns_parsed_json(self):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"resourceStrings": {"hello": {"value": "Bonjour"}}}
|
|
|
|
with (
|
|
patch.dict("os.environ", {"GP_ADMIN_USER_ID": "u", "GP_ADMIN_PASSWORD": "p"}),
|
|
patch("gp_client.requests.get", return_value=mock_response),
|
|
):
|
|
importlib.reload(gp_client)
|
|
result = gp_client.get_strings("fr")
|
|
|
|
assert result["resourceStrings"]["hello"]["value"] == "Bonjour"
|
|
|
|
def test_raises_on_403(self):
|
|
import requests as req
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status.side_effect = req.exceptions.HTTPError("403 Forbidden")
|
|
|
|
with (
|
|
patch.dict("os.environ", {"GP_ADMIN_USER_ID": "u", "GP_ADMIN_PASSWORD": "p"}),
|
|
patch("gp_client.requests.get", return_value=mock_response),
|
|
):
|
|
importlib.reload(gp_client)
|
|
with pytest.raises(req.exceptions.HTTPError):
|
|
gp_client.get_strings("fr")
|