mirror of
https://github.com/simonw/datasette.git
synced 2026-10-26 18:26:59 +01:00
fecbed82b4Stored write queries converted every JSON value with str(), so null was
stored as 'None', true as 'True' and lists as Python repr. execute-write
kept JSON types but rejected arrays and objects and silently bound
missing parameters as empty strings.
Both endpoints now share one function for JSON parameters:
- strings, numbers and null pass through, true/false become 1/0, and
arrays and objects are stored as JSON text, matching the insert API
- missing parameters return a 400 listing them
- unknown parameters return a 400
Stored write queries keep their top-level body shape, with _json and
csrftoken as the only reserved keys. Parameters declared in "params"
are accepted, magic parameters are not required, and a body wrapped in
{"params": ...} gets an error explaining the difference.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012CXYAsBaNidHVk4GipkDjs
647 lines
22 KiB
Python
647 lines
22 KiB
Python
import json
|
|
import re
|
|
|
|
import pytest
|
|
from asgiref.sync import async_to_sync
|
|
from bs4 import BeautifulSoup as Soup
|
|
|
|
from datasette.app import Datasette
|
|
|
|
from .fixtures import make_app_client
|
|
|
|
|
|
def update_query(client, name, **kwargs):
|
|
async_to_sync(client.ds.invoke_startup)()
|
|
async_to_sync(client.ds.update_query)("data", name, **kwargs)
|
|
|
|
|
|
@pytest.fixture
|
|
def stored_write_client(tmpdir):
|
|
template_dir = tmpdir / "stored_write_templates"
|
|
template_dir.mkdir()
|
|
(template_dir / "query-data-update_name.html").write_text(
|
|
"""
|
|
{% extends "query.html" %}
|
|
{% block content %}!!!CUSTOM_UPDATE_NAME_TEMPLATE!!!{{ super() }}{% endblock %}
|
|
""",
|
|
"utf-8",
|
|
)
|
|
with make_app_client(
|
|
extra_databases={"data.db": "create table names (name text)"},
|
|
template_dir=str(template_dir),
|
|
config={
|
|
"databases": {
|
|
"data": {
|
|
"queries": {
|
|
"stored_read": {"sql": "select * from names"},
|
|
"add_name": {
|
|
"sql": "insert into names (name) values (:name)",
|
|
"write": True,
|
|
"on_success_redirect": "/data/add_name?success",
|
|
},
|
|
"add_name_specify_id": {
|
|
"sql": "insert into names (rowid, name) values (:rowid, :name)",
|
|
"on_success_message_sql": "select 'Name added: ' || :name || ' with rowid ' || :rowid",
|
|
"write": True,
|
|
"on_error_redirect": "/data/add_name_specify_id?error",
|
|
},
|
|
"add_name_specify_id_with_error_in_on_success_message_sql": {
|
|
"sql": "insert into names (rowid, name) values (:rowid, :name)",
|
|
"on_success_message_sql": "select this is bad SQL",
|
|
"write": True,
|
|
},
|
|
"delete_name": {
|
|
"sql": "delete from names where rowid = :rowid",
|
|
"write": True,
|
|
"on_success_message": "Name deleted",
|
|
"allow": {"id": "root"},
|
|
},
|
|
"update_name": {
|
|
"sql": "update names set name = :name where rowid = :rowid",
|
|
"params": ["rowid", "name", "extra"],
|
|
"write": True,
|
|
},
|
|
}
|
|
}
|
|
}
|
|
},
|
|
) as client:
|
|
yield client
|
|
|
|
|
|
@pytest.fixture
|
|
def stored_write_immutable_client():
|
|
with make_app_client(
|
|
is_immutable=True,
|
|
config={
|
|
"databases": {
|
|
"fixtures": {
|
|
"queries": {
|
|
"add": {
|
|
"sql": "insert into sortable (text) values (:text)",
|
|
"write": True,
|
|
},
|
|
}
|
|
}
|
|
}
|
|
},
|
|
) as client:
|
|
yield client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stored_query_with_named_parameter(ds_client):
|
|
response = await ds_client.get(
|
|
"/fixtures/neighborhood_search.json?text=town&_shape=arrays"
|
|
)
|
|
assert response.json()["rows"] == [
|
|
["Corktown", "Detroit", "MI"],
|
|
["Downtown", "Los Angeles", "CA"],
|
|
["Downtown", "Detroit", "MI"],
|
|
["Greektown", "Detroit", "MI"],
|
|
["Koreatown", "Los Angeles", "CA"],
|
|
["Mexicantown", "Detroit", "MI"],
|
|
]
|
|
|
|
|
|
def test_insert(stored_write_client):
|
|
response = stored_write_client.post(
|
|
"/data/add_name",
|
|
{"name": "Hello"},
|
|
csrftoken_from=True,
|
|
cookies={"foo": "bar"},
|
|
)
|
|
messages = stored_write_client.ds.unsign(
|
|
response.cookies["ds_messages"], "messages"
|
|
)
|
|
assert messages == [["Query executed, 1 row affected", 1]]
|
|
assert response.status == 302
|
|
assert response.headers["Location"] == "/data/add_name?success"
|
|
|
|
|
|
def test_insert_blocked_cross_site(stored_write_client):
|
|
# A cross-site POST (browser-originated) must be blocked
|
|
response = stored_write_client.post(
|
|
"/data/add_name",
|
|
{"name": "Hello"},
|
|
headers={"sec-fetch-site": "cross-site"},
|
|
)
|
|
assert 403 == response.status
|
|
|
|
|
|
def test_insert_no_cookies_no_csrf(stored_write_client):
|
|
response = stored_write_client.post("/data/add_name", {"name": "Hello"})
|
|
assert 302 == response.status
|
|
assert "/data/add_name?success" == response.headers["Location"]
|
|
|
|
|
|
def test_custom_success_message(stored_write_client):
|
|
response = stored_write_client.post(
|
|
"/data/delete_name",
|
|
{"rowid": 1},
|
|
cookies={"ds_actor": stored_write_client.actor_cookie({"id": "root"})},
|
|
csrftoken_from=True,
|
|
)
|
|
assert 302 == response.status
|
|
messages = stored_write_client.ds.unsign(
|
|
response.cookies["ds_messages"], "messages"
|
|
)
|
|
assert [["Name deleted", 1]] == messages
|
|
|
|
|
|
def test_insert_error(stored_write_client):
|
|
stored_write_client.post("/data/add_name", {"name": "Hello"}, csrftoken_from=True)
|
|
response = stored_write_client.post(
|
|
"/data/add_name_specify_id",
|
|
{"rowid": 1, "name": "Should fail"},
|
|
csrftoken_from=True,
|
|
)
|
|
assert 302 == response.status
|
|
assert "/data/add_name_specify_id?error" == response.headers["Location"]
|
|
messages = stored_write_client.ds.unsign(
|
|
response.cookies["ds_messages"], "messages"
|
|
)
|
|
assert [["UNIQUE constraint failed: names.rowid", 3]] == messages
|
|
# How about with a custom error message?
|
|
update_query(stored_write_client, "add_name_specify_id", on_error_message="ERROR")
|
|
response = stored_write_client.post(
|
|
"/data/add_name_specify_id",
|
|
{"rowid": 1, "name": "Should fail"},
|
|
csrftoken_from=True,
|
|
)
|
|
assert [["ERROR", 3]] == stored_write_client.ds.unsign(
|
|
response.cookies["ds_messages"], "messages"
|
|
)
|
|
|
|
|
|
def test_on_success_message_sql(stored_write_client):
|
|
response = stored_write_client.post(
|
|
"/data/add_name_specify_id",
|
|
{"rowid": 5, "name": "Should be OK"},
|
|
csrftoken_from=True,
|
|
)
|
|
assert response.status == 302
|
|
assert response.headers["Location"] == "/data/add_name_specify_id"
|
|
messages = stored_write_client.ds.unsign(
|
|
response.cookies["ds_messages"], "messages"
|
|
)
|
|
assert messages == [["Name added: Should be OK with rowid 5", 1]]
|
|
|
|
|
|
def test_error_in_on_success_message_sql(stored_write_client):
|
|
response = stored_write_client.post(
|
|
"/data/add_name_specify_id_with_error_in_on_success_message_sql",
|
|
{"rowid": 1, "name": "Should fail"},
|
|
csrftoken_from=True,
|
|
)
|
|
messages = stored_write_client.ds.unsign(
|
|
response.cookies["ds_messages"], "messages"
|
|
)
|
|
assert messages == [
|
|
["Error running on_success_message_sql: no such column: bad", 3]
|
|
]
|
|
|
|
|
|
def test_custom_params(stored_write_client):
|
|
response = stored_write_client.get("/data/update_name?extra=foo")
|
|
assert (
|
|
'<input type="text" id="qp3" name="extra" value="foo" data-parameter-control data-parameter-name="extra">'
|
|
in response.text
|
|
)
|
|
|
|
|
|
def test_stored_query_pages_vary_by_credentials(stored_write_client):
|
|
# Even without per-cookie CSRF tokens, anonymous pages must not be reused
|
|
# for authenticated users whose permissions or navigation can differ.
|
|
for path in ("/data", "/data/update_name"):
|
|
response = stored_write_client.get(path)
|
|
assert response.headers["vary"] == "Cookie, Authorization"
|
|
|
|
|
|
def test_json_post_body(stored_write_client):
|
|
response = stored_write_client.post(
|
|
"/data/add_name",
|
|
body=json.dumps({"name": ["Hello", "there"]}),
|
|
)
|
|
assert 302 == response.status
|
|
assert "/data/add_name?success" == response.headers["Location"]
|
|
rows = stored_write_client.get("/data/names.json?_shape=array").json
|
|
assert rows == [{"rowid": 1, "name": '["Hello", "there"]'}]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"headers,body,querystring",
|
|
(
|
|
(None, "name=NameGoesHere", "?_json=1"),
|
|
({"Accept": "application/json"}, "name=NameGoesHere", None),
|
|
(None, "name=NameGoesHere&_json=1", None),
|
|
(None, '{"name": "NameGoesHere", "_json": 1}', None),
|
|
),
|
|
)
|
|
def test_json_response(stored_write_client, headers, body, querystring):
|
|
response = stored_write_client.post(
|
|
"/data/add_name" + (querystring or ""),
|
|
body=body,
|
|
headers=headers,
|
|
)
|
|
assert 200 == response.status
|
|
assert response.headers["content-type"] == "application/json; charset=utf-8"
|
|
assert response.json == {
|
|
"ok": True,
|
|
"message": "Query executed, 1 row affected",
|
|
"redirect": "/data/add_name?success",
|
|
}
|
|
rows = stored_write_client.get("/data/names.json?_shape=array").json
|
|
assert rows == [{"rowid": 1, "name": "NameGoesHere"}]
|
|
|
|
|
|
def test_stored_query_permissions_on_database_page(stored_write_client):
|
|
# Without auth shows the five public queries
|
|
anon_response = stored_write_client.get("/data.json")
|
|
query_names = {q["name"] for q in anon_response.json["queries"]}
|
|
assert query_names == {
|
|
"add_name_specify_id_with_error_in_on_success_message_sql",
|
|
"update_name",
|
|
"add_name_specify_id",
|
|
"stored_read",
|
|
"add_name",
|
|
}
|
|
assert anon_response.json["queries_more"] is False
|
|
|
|
# With auth the database page preview shows the first five queries
|
|
response = stored_write_client.get(
|
|
"/data.json",
|
|
cookies={"ds_actor": stored_write_client.actor_cookie({"id": "root"})},
|
|
)
|
|
assert response.status == 200
|
|
query_names_and_private = sorted(
|
|
[
|
|
{"name": q["name"], "private": q["private"]}
|
|
for q in response.json["queries"]
|
|
],
|
|
key=lambda q: q["name"],
|
|
)
|
|
assert query_names_and_private == [
|
|
{"name": "add_name", "private": False},
|
|
{"name": "add_name_specify_id", "private": False},
|
|
{
|
|
"name": "add_name_specify_id_with_error_in_on_success_message_sql",
|
|
"private": False,
|
|
},
|
|
{"name": "delete_name", "private": True},
|
|
{"name": "stored_read", "private": False},
|
|
]
|
|
assert response.json["queries_more"] is True
|
|
|
|
# The full query list endpoint includes the remaining query
|
|
response = stored_write_client.get(
|
|
"/data/-/queries.json?_size=10",
|
|
cookies={"ds_actor": stored_write_client.actor_cookie({"id": "root"})},
|
|
)
|
|
assert response.status == 200
|
|
query_names_and_private = sorted(
|
|
[
|
|
{"name": q["name"], "private": q["private"]}
|
|
for q in response.json["queries"]
|
|
],
|
|
key=lambda q: q["name"],
|
|
)
|
|
assert query_names_and_private == [
|
|
{"name": "add_name", "private": False},
|
|
{"name": "add_name_specify_id", "private": False},
|
|
{
|
|
"name": "add_name_specify_id_with_error_in_on_success_message_sql",
|
|
"private": False,
|
|
},
|
|
{"name": "delete_name", "private": True},
|
|
{"name": "stored_read", "private": False},
|
|
{"name": "update_name", "private": False},
|
|
]
|
|
|
|
|
|
def test_stored_query_permissions(stored_write_client):
|
|
assert 403 == stored_write_client.get("/data/delete_name").status
|
|
assert 200 == stored_write_client.get("/data/update_name").status
|
|
cookies = {"ds_actor": stored_write_client.actor_cookie({"id": "root"})}
|
|
assert 200 == stored_write_client.get("/data/delete_name", cookies=cookies).status
|
|
assert 200 == stored_write_client.get("/data/update_name", cookies=cookies).status
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def magic_parameters_client():
|
|
with make_app_client(
|
|
extra_databases={"data.db": "create table logs (line text)"},
|
|
config={
|
|
"databases": {
|
|
"data": {
|
|
"queries": {
|
|
"runme_post": {"sql": "", "write": True},
|
|
"runme_get": {"sql": ""},
|
|
}
|
|
}
|
|
}
|
|
},
|
|
) as client:
|
|
yield client
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"magic_parameter,expected_re",
|
|
[
|
|
("_actor_id", "root"),
|
|
("_header_host", "localhost"),
|
|
("_header_not_a_thing", ""),
|
|
("_cookie_foo", "bar"),
|
|
("_now_epoch", r"^\d+$"),
|
|
("_now_date_utc", r"^\d{4}-\d{2}-\d{2}$"),
|
|
("_now_datetime_utc", r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$"),
|
|
("_random_chars_1", r"^\w$"),
|
|
("_random_chars_10", r"^\w{10}$"),
|
|
],
|
|
)
|
|
def test_magic_parameters(magic_parameters_client, magic_parameter, expected_re):
|
|
update_query(
|
|
magic_parameters_client,
|
|
"runme_post",
|
|
sql=f"insert into logs (line) values (:{magic_parameter})",
|
|
)
|
|
update_query(
|
|
magic_parameters_client,
|
|
"runme_get",
|
|
sql=f"select :{magic_parameter} as result",
|
|
)
|
|
cookies = {
|
|
"ds_actor": magic_parameters_client.actor_cookie({"id": "root"}),
|
|
"foo": "bar",
|
|
}
|
|
# Test the GET version
|
|
get_response = magic_parameters_client.get(
|
|
"/data/runme_get.json?_shape=array", cookies=cookies
|
|
)
|
|
get_actual = get_response.json[0]["result"]
|
|
assert re.match(expected_re, str(get_actual))
|
|
# Test the form
|
|
form_response = magic_parameters_client.get("/data/runme_post")
|
|
soup = Soup(form_response.body, "html.parser")
|
|
# The magic parameter should not be represented as a form field
|
|
assert None is soup.find("input", {"name": magic_parameter})
|
|
# Submit the form to create a log line
|
|
response = magic_parameters_client.post(
|
|
"/data/runme_post?_json=1", {}, csrftoken_from=True, cookies=cookies
|
|
)
|
|
assert response.json == {
|
|
"ok": True,
|
|
"message": "Query executed, 1 row affected",
|
|
"redirect": None,
|
|
}
|
|
post_actual = magic_parameters_client.get(
|
|
"/data/logs.json?_sort_desc=rowid&_shape=array"
|
|
).json[0]["line"]
|
|
assert re.match(expected_re, post_actual)
|
|
|
|
|
|
@pytest.mark.parametrize("use_csrf", [True, False])
|
|
@pytest.mark.parametrize("return_json", [True, False])
|
|
def test_magic_parameters_csrf_json(magic_parameters_client, use_csrf, return_json):
|
|
update_query(
|
|
magic_parameters_client,
|
|
"runme_post",
|
|
sql="insert into logs (line) values (:_header_host)",
|
|
)
|
|
qs = ""
|
|
if return_json:
|
|
qs = "?_json=1"
|
|
response = magic_parameters_client.post(
|
|
f"/data/runme_post{qs}",
|
|
{},
|
|
csrftoken_from=use_csrf or None,
|
|
)
|
|
if return_json:
|
|
assert response.status == 200
|
|
assert response.json["ok"], response.json
|
|
else:
|
|
assert response.status == 302
|
|
messages = magic_parameters_client.ds.unsign(
|
|
response.cookies["ds_messages"], "messages"
|
|
)
|
|
assert [["Query executed, 1 row affected", 1]] == messages
|
|
post_actual = magic_parameters_client.get(
|
|
"/data/logs.json?_sort_desc=rowid&_shape=array"
|
|
).json[0]["line"]
|
|
assert post_actual == "localhost"
|
|
|
|
|
|
def test_magic_parameters_cannot_be_used_in_arbitrary_queries(magic_parameters_client):
|
|
response = magic_parameters_client.get(
|
|
"/data/-/query.json?sql=select+:_header_host&_shape=array"
|
|
)
|
|
assert 400 == response.status
|
|
assert response.json["error"].startswith("You did not supply a value for binding")
|
|
|
|
|
|
def test_stored_write_custom_template(stored_write_client):
|
|
response = stored_write_client.get("/data/update_name")
|
|
assert response.status == 200
|
|
assert "!!!CUSTOM_UPDATE_NAME_TEMPLATE!!!" in response.text
|
|
assert (
|
|
"<!-- Templates considered: *query-data-update_name.html, query-data.html, query.html -->"
|
|
in response.text
|
|
)
|
|
# And test for link rel=alternate while we're here:
|
|
assert (
|
|
'<link rel="alternate" type="application/json+datasette" href="http://localhost/data/update_name.json">'
|
|
in response.text
|
|
)
|
|
assert (
|
|
response.headers["link"]
|
|
== '<http://localhost/data/update_name.json>; rel="alternate"; type="application/json+datasette"'
|
|
)
|
|
|
|
|
|
def test_stored_write_query_disabled_for_immutable_database(
|
|
stored_write_immutable_client,
|
|
):
|
|
response = stored_write_immutable_client.get("/fixtures/add")
|
|
assert response.status == 200
|
|
assert (
|
|
"This query cannot be executed because the database is immutable."
|
|
in response.text
|
|
)
|
|
assert '<input type="submit" value="Run SQL" disabled>' in response.text
|
|
# Submitting form should get a forbidden error
|
|
response = stored_write_immutable_client.post(
|
|
"/fixtures/add",
|
|
{"text": "text"},
|
|
csrftoken_from=True,
|
|
)
|
|
assert response.status == 403
|
|
assert "Database is immutable" in response.text
|
|
|
|
|
|
JSON_EVENT_SQL = (
|
|
"insert into events (value, flag, extra, tags, meta) "
|
|
"values (:value, :flag, :extra, :tags, :meta)"
|
|
)
|
|
JSON_EVENT_PARAMS = {
|
|
"value": 5,
|
|
"flag": True,
|
|
"extra": None,
|
|
"tags": ["a", "b"],
|
|
"meta": {"k": 1},
|
|
}
|
|
|
|
|
|
async def _json_params_datasette():
|
|
ds = Datasette(
|
|
memory=True,
|
|
config={
|
|
"databases": {
|
|
"data": {
|
|
"queries": {
|
|
"add_event": {"sql": JSON_EVENT_SQL, "write": True},
|
|
"add_event_declared": {
|
|
"sql": "insert into events (value) values (:value)",
|
|
"params": ["value", "note"],
|
|
"write": True,
|
|
},
|
|
"add_event_actor": {
|
|
"sql": "insert into events (value, extra) values (:value, :_actor_id)",
|
|
"write": True,
|
|
},
|
|
}
|
|
}
|
|
}
|
|
},
|
|
)
|
|
ds.root_enabled = True
|
|
db = ds.add_memory_database("stored_query_json_params", name="data")
|
|
# Columns without a declared type store values exactly as they are bound
|
|
await db.execute_write(
|
|
"create table if not exists events (value, flag, extra, tags, meta)"
|
|
)
|
|
await db.execute_write("delete from events")
|
|
await ds.invoke_startup()
|
|
return ds, db
|
|
|
|
|
|
async def _post_json_params(ds, endpoint, params):
|
|
if endpoint == "stored":
|
|
return await ds.client.post(
|
|
"/data/add_event",
|
|
actor={"id": "root"},
|
|
json=params,
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
return await ds.client.post(
|
|
"/data/-/execute-write",
|
|
actor={"id": "root"},
|
|
json={"sql": JSON_EVENT_SQL, "params": params},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("endpoint", ("stored", "execute-write"))
|
|
async def test_json_parameters_keep_types(endpoint):
|
|
# https://github.com/simonw/datasette/issues/2946
|
|
ds, db = await _json_params_datasette()
|
|
response = await _post_json_params(ds, endpoint, JSON_EVENT_PARAMS)
|
|
assert response.status_code == 200
|
|
assert response.json()["ok"] is True
|
|
rows = (
|
|
await db.execute(
|
|
"select value, typeof(value) as value_type, flag, typeof(flag) as flag_type, "
|
|
"extra, typeof(extra) as extra_type, tags, meta from events"
|
|
)
|
|
).dicts()
|
|
assert rows == [
|
|
{
|
|
"value": 5,
|
|
"value_type": "integer",
|
|
"flag": 1,
|
|
"flag_type": "integer",
|
|
"extra": None,
|
|
"extra_type": "null",
|
|
"tags": '["a", "b"]',
|
|
"meta": '{"k": 1}',
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("endpoint", ("stored", "execute-write"))
|
|
@pytest.mark.parametrize(
|
|
"params,expected_error",
|
|
(
|
|
({"value": 1}, "Missing parameters: flag, extra, tags, meta"),
|
|
({**JSON_EVENT_PARAMS, "bogus": 1}, "Unknown parameters: bogus"),
|
|
),
|
|
)
|
|
async def test_json_parameters_missing_or_unknown(endpoint, params, expected_error):
|
|
ds, db = await _json_params_datasette()
|
|
response = await _post_json_params(ds, endpoint, params)
|
|
assert response.status_code == 400
|
|
assert response.json()["errors"] == [expected_error]
|
|
assert (await db.execute("select count(*) from events")).single_value() == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stored_query_json_params_wrapper_hint():
|
|
ds, db = await _json_params_datasette()
|
|
response = await ds.client.post(
|
|
"/data/add_event",
|
|
actor={"id": "root"},
|
|
json={"params": JSON_EVENT_PARAMS},
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["errors"] == [
|
|
(
|
|
"Unknown parameters: params. Stored queries take parameters as "
|
|
'top-level keys, not inside "params"'
|
|
)
|
|
]
|
|
assert (await db.execute("select count(*) from events")).single_value() == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stored_query_json_declared_and_magic_parameters():
|
|
ds, db = await _json_params_datasette()
|
|
# Parameters declared in "params" are accepted even if the SQL does not use them
|
|
declared = await ds.client.post(
|
|
"/data/add_event_declared",
|
|
actor={"id": "root"},
|
|
json={"value": 1, "note": "ignored"},
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
assert declared.status_code == 200
|
|
# Magic parameters are filled in by Datasette, not required from the client
|
|
magic = await ds.client.post(
|
|
"/data/add_event_actor",
|
|
actor={"id": "root"},
|
|
json={"value": 2},
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
assert magic.status_code == 200
|
|
rows = (await db.execute("select value, extra from events order by value")).dicts()
|
|
assert rows == [{"value": 1, "extra": None}, {"value": 2, "extra": "root"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"body,expected_error",
|
|
(
|
|
("[1, 2]", "JSON must be a dictionary"),
|
|
("{bad json}", "Invalid JSON: "),
|
|
),
|
|
)
|
|
async def test_stored_query_invalid_json_body(body, expected_error):
|
|
ds, db = await _json_params_datasette()
|
|
response = await ds.client.post(
|
|
"/data/add_event",
|
|
actor={"id": "root"},
|
|
content=body,
|
|
headers={"Content-Type": "application/json", "Accept": "application/json"},
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["errors"][0].startswith(expected_error)
|
|
assert (await db.execute("select count(*) from events")).single_value() == 0
|