mirror of
https://github.com/bitechdev/ResolveSpec.git
synced 2026-10-01 12:31:59 +00:00
211 lines
7.0 KiB
Python
211 lines
7.0 KiB
Python
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from resolvespec import (
|
|
AsyncHeaderSpecClient,
|
|
AsyncResolveSpecClient,
|
|
HeaderSpecClient,
|
|
ResolveSpecClient,
|
|
ResolveSpecError,
|
|
get_headerspec_client,
|
|
get_resolvespec_client,
|
|
)
|
|
|
|
CFG = dict(base_url="http://localhost:3000", token="test-token")
|
|
|
|
|
|
def make(handler, **kw):
|
|
return ResolveSpecClient(**{**CFG, **kw}, transport=httpx.MockTransport(handler))
|
|
|
|
|
|
def ok(_req):
|
|
return httpx.Response(200, json={"success": True, "data": [{"id": 1}]})
|
|
|
|
|
|
def capture():
|
|
seen = []
|
|
|
|
def handler(req):
|
|
seen.append(req)
|
|
return httpx.Response(200, json={"success": True, "data": {"id": 1, "name": "Test"}})
|
|
|
|
return seen, handler
|
|
|
|
|
|
def body(req):
|
|
return json.loads(req.content)
|
|
|
|
|
|
def test_read_with_numeric_id():
|
|
seen, h = capture()
|
|
with make(h) as c:
|
|
assert c.read("public", "users", 1)["success"] is True
|
|
r = seen[0]
|
|
assert str(r.url) == "http://localhost:3000/public/users/1"
|
|
assert r.method == "POST"
|
|
assert r.headers["authorization"] == "Bearer test-token"
|
|
assert r.headers["content-type"] == "application/json"
|
|
assert body(r) == {"operation": "read"}
|
|
|
|
|
|
def test_read_array_id_goes_in_body():
|
|
seen, h = capture()
|
|
with make(h) as c:
|
|
c.read("public", "users", ["1", "2"])
|
|
assert str(seen[0].url) == "http://localhost:3000/public/users"
|
|
assert body(seen[0])["id"] == ["1", "2"]
|
|
|
|
|
|
def test_read_options_passthrough():
|
|
seen, h = capture()
|
|
opts = {
|
|
"columns": ["id", "name"], "omit_columns": ["secret"],
|
|
"filters": [{"column": "active", "operator": "eq", "value": True}],
|
|
"sort": [{"column": "name", "direction": "asc"}],
|
|
"limit": 10, "offset": 0, "cursor_forward": "cursor1", "fetch_row_number": "5",
|
|
"customOperators": [{"name": "x", "sql": "a = 1"}],
|
|
}
|
|
with make(h) as c:
|
|
c.read("public", "users", options=opts)
|
|
assert body(seen[0])["options"] == opts
|
|
|
|
|
|
def test_create():
|
|
seen, h = capture()
|
|
with make(h) as c:
|
|
res = c.create("public", "users", {"name": "Test"})
|
|
assert res["data"]["name"] == "Test"
|
|
assert body(seen[0]) == {"operation": "create", "data": {"name": "Test"}}
|
|
|
|
|
|
def test_create_batch():
|
|
seen, h = capture()
|
|
with make(h) as c:
|
|
c.create("public", "users", [{"a": 1}, {"a": 2}])
|
|
assert body(seen[0])["data"] == [{"a": 1}, {"a": 2}]
|
|
|
|
|
|
def test_update_with_id_in_url_and_array():
|
|
seen, h = capture()
|
|
with make(h) as c:
|
|
c.update("public", "users", {"name": "X"}, 5)
|
|
c.update("public", "users", {"name": "X"}, ["1", "2"])
|
|
assert str(seen[0].url).endswith("/public/users/5")
|
|
assert body(seen[0]) == {"operation": "update", "data": {"name": "X"}}
|
|
assert str(seen[1].url).endswith("/public/users")
|
|
assert body(seen[1])["id"] == ["1", "2"]
|
|
|
|
|
|
def test_update_preserves_empty_string_and_null():
|
|
seen, h = capture()
|
|
with make(h) as c:
|
|
c.update("public", "users", {"a": "", "b": None}, 1)
|
|
assert body(seen[0])["data"] == {"a": "", "b": None}
|
|
|
|
|
|
def test_delete():
|
|
seen, h = capture()
|
|
with make(h) as c:
|
|
c.delete("public", "users", 1)
|
|
assert str(seen[0].url).endswith("/public/users/1")
|
|
assert body(seen[0]) == {"operation": "delete"}
|
|
|
|
|
|
def test_get_metadata():
|
|
seen, h = capture()
|
|
with make(h) as c:
|
|
c.get_metadata("public", "users")
|
|
assert seen[0].method == "GET"
|
|
assert str(seen[0].url) == "http://localhost:3000/public/users"
|
|
assert not seen[0].content
|
|
|
|
|
|
def test_error_uses_server_message():
|
|
with make(lambda r: httpx.Response(404, json={"success": False, "error": {"code": "not_found", "message": "nope"}})) as c:
|
|
with pytest.raises(ResolveSpecError, match="nope") as ei:
|
|
c.read("public", "users", 1)
|
|
assert ei.value.status_code == 404 and ei.value.code == "not_found"
|
|
|
|
|
|
def test_id_is_url_quoted():
|
|
seen, h = capture()
|
|
with make(h) as c:
|
|
c.read("public", "users", "a/b")
|
|
assert str(seen[0].url).endswith("/public/users/a%2Fb")
|
|
|
|
|
|
def test_trailing_slash_base_url():
|
|
seen, h = capture()
|
|
with make(h, base_url="http://localhost:3000/") as c:
|
|
c.read("public", "users")
|
|
assert str(seen[0].url) == "http://localhost:3000/public/users"
|
|
|
|
|
|
async def test_async_client():
|
|
async def handler(req):
|
|
return httpx.Response(200, json={"success": True, "data": [1]})
|
|
|
|
async with AsyncResolveSpecClient(**CFG, transport=httpx.MockTransport(handler)) as c:
|
|
assert (await c.read("public", "users"))["data"] == [1]
|
|
assert (await c.create("public", "users", {}))["success"]
|
|
assert (await c.update("public", "users", {}, 1))["success"]
|
|
assert (await c.delete("public", "users", 1))["success"]
|
|
assert (await c.get_metadata("public", "users"))["success"]
|
|
|
|
|
|
# ---- custom headers (ported from custom-headers.test.ts) ----
|
|
|
|
@pytest.mark.parametrize("cls", [ResolveSpecClient, HeaderSpecClient])
|
|
def test_custom_headers_on_every_op_case_insensitive(cls):
|
|
seen = []
|
|
|
|
def handler(req):
|
|
seen.append(req)
|
|
return httpx.Response(200, json={"success": True, "data": []})
|
|
|
|
headers = {"X-Tenant": "acme", "authorization": "Basic ignored",
|
|
"content-type": "application/custom+json", "x-limit": "99"}
|
|
with cls("http://localhost:3000", "tok", headers, transport=httpx.MockTransport(handler)) as c:
|
|
c.read("public", "users", options={"limit": 10})
|
|
c.create("public", "users", {})
|
|
if cls is ResolveSpecClient:
|
|
c.update("public", "users", {}, "1")
|
|
c.get_metadata("public", "users")
|
|
else:
|
|
c.update("public", "users", "1", {})
|
|
c.delete("public", "users", "1")
|
|
for r in seen:
|
|
assert r.headers["x-tenant"] == "acme"
|
|
assert r.headers["authorization"] == "Bearer tok"
|
|
assert r.headers["content-type"] == "application/custom+json"
|
|
if cls is HeaderSpecClient:
|
|
assert seen[0].headers["x-limit"] == "10"
|
|
assert headers["authorization"] == "Basic ignored"
|
|
assert headers["x-limit"] == "99"
|
|
|
|
|
|
@pytest.mark.parametrize("cls", [ResolveSpecClient, HeaderSpecClient])
|
|
def test_custom_auth_without_token(cls):
|
|
seen = []
|
|
|
|
def handler(req):
|
|
seen.append(req)
|
|
return httpx.Response(200, json={"success": True, "data": []})
|
|
|
|
with cls("http://localhost:3000", headers={"Authorization": "Basic custom"}, transport=httpx.MockTransport(handler)) as c:
|
|
c.read("public", "users")
|
|
assert seen[0].headers["authorization"] == "Basic custom"
|
|
|
|
|
|
@pytest.mark.parametrize("factory", [get_resolvespec_client, get_headerspec_client])
|
|
def test_cache_isolation_and_snapshot(factory):
|
|
headers = {"X-Tenant": "acme", "X-App": "grid"}
|
|
first = factory("http://tenant-cache", "one", headers)
|
|
assert factory("http://tenant-cache", "one", {"x-app": "grid", "x-tenant": "acme"}) is first
|
|
assert factory("http://tenant-cache", "two", headers) is not first
|
|
headers["X-Tenant"] = "other"
|
|
assert factory("http://tenant-cache", "one", headers) is not first
|
|
assert first.headers["X-Tenant"] == "acme"
|