from datasette.app import Datasette from datasette.utils import sqlite3 import pytest import time @pytest.fixture def ds_write(tmp_path_factory): db_directory = tmp_path_factory.mktemp("dbs") db_path = str(db_directory / "data.db") db_path_immutable = str(db_directory / "immutable.db") db1 = sqlite3.connect(str(db_path)) db2 = sqlite3.connect(str(db_path_immutable)) for db in (db1, db2): db.execute("vacuum") db.execute( "create table docs (id integer primary key, title text, score float, age integer)" ) ds = Datasette([db_path], immutables=[db_path_immutable]) yield ds db.close() def write_token(ds, actor_id="root", permissions=None): to_sign = {"a": actor_id, "token": "dstok", "t": int(time.time())} if permissions: to_sign["_r"] = {"a": permissions} return "dstok_{}".format(ds.sign(to_sign, namespace="token")) def _headers(token): return { "Authorization": "Bearer {}".format(token), "Content-Type": "application/json", } @pytest.mark.asyncio async def test_insert_row(ds_write): token = write_token(ds_write) response = await ds_write.client.post( "/data/docs/-/insert", json={"row": {"title": "Test", "score": 1.2, "age": 5}}, headers=_headers(token), ) expected_row = {"id": 1, "title": "Test", "score": 1.2, "age": 5} assert response.status_code == 201 assert response.json()["ok"] is True assert response.json()["rows"] == [expected_row] rows = (await ds_write.get_database("data").execute("select * from docs")).rows assert dict(rows[0]) == expected_row @pytest.mark.asyncio @pytest.mark.parametrize("return_rows", (True, False)) async def test_insert_rows(ds_write, return_rows): token = write_token(ds_write) data = { "rows": [ {"title": "Test {}".format(i), "score": 1.0, "age": 5} for i in range(20) ] } if return_rows: data["return"] = True response = await ds_write.client.post( "/data/docs/-/insert", json=data, headers=_headers(token), ) assert response.status_code == 201 actual_rows = [ dict(r) for r in ( await ds_write.get_database("data").execute("select * from docs") ).rows ] assert len(actual_rows) == 20 assert actual_rows == [ {"id": i + 1, "title": "Test {}".format(i), "score": 1.0, "age": 5} for i in range(20) ] assert response.json()["ok"] is True if return_rows: assert response.json()["rows"] == actual_rows @pytest.mark.asyncio @pytest.mark.parametrize( "path,input,special_case,expected_status,expected_errors", ( ( "/data2/docs/-/insert", {}, None, 404, ["Database not found: data2"], ), ( "/data/docs2/-/insert", {}, None, 404, ["Table not found: docs2"], ), ( "/data/docs/-/insert", {"rows": [{"title": "Test"} for i in range(10)]}, "bad_token", 403, ["Permission denied"], ), ( "/data/docs/-/insert", {}, "invalid_json", 400, [ "Invalid JSON: Expecting property name enclosed in double quotes: line 1 column 2 (char 1)" ], ), ( "/data/docs/-/insert", {}, "invalid_content_type", 400, ["Invalid content-type, must be application/json"], ), ( "/data/docs/-/insert", [], None, 400, ["JSON must be a dictionary"], ), ( "/data/docs/-/insert", {"row": "blah"}, None, 400, ['"row" must be a dictionary'], ), ( "/data/docs/-/insert", {"blah": "blah"}, None, 400, ['JSON must have one or other of "row" or "rows"'], ), ( "/data/docs/-/insert", {"rows": "blah"}, None, 400, ['"rows" must be a list'], ), ( "/data/docs/-/insert", {"rows": ["blah"]}, None, 400, ['"rows" must be a list of dictionaries'], ), ( "/data/docs/-/insert", {"rows": [{"title": "Test"} for i in range(101)]}, None, 400, ["Too many rows, maximum allowed is 100"], ), ( "/data/docs/-/insert", {"rows": [{"id": 1, "title": "Test"}, {"id": 2, "title": "Test"}]}, "duplicate_id", 400, ["UNIQUE constraint failed: docs.id"], ), ( "/data/docs/-/insert", {"rows": [{"title": "Test"}], "ignore": True, "replace": True}, None, 400, ['Cannot use "ignore" and "replace" at the same time'], ), ( "/data/docs/-/insert", {"rows": [{"title": "Test"}], "invalid_param": True}, None, 400, ['Invalid parameter: "invalid_param"'], ), ( "/data/docs/-/insert", {"rows": [{"title": "Test"}], "one": True, "two": True}, None, 400, ['Invalid parameter: "one", "two"'], ), ( "/immutable/docs/-/insert", {"rows": [{"title": "Test"}]}, None, 403, ["Database is immutable"], ), # Validate columns of each row ( "/data/docs/-/insert", {"rows": [{"title": "Test", "bad": 1, "worse": 2} for i in range(2)]}, None, 400, [ "Row 0 has invalid columns: bad, worse", "Row 1 has invalid columns: bad, worse", ], ), ## UPSERT ERRORS: ( "/immutable/docs/-/upsert", {"rows": [{"title": "Test"}]}, None, 403, ["Database is immutable"], ), ( "/data/badtable/-/upsert", {"rows": [{"title": "Test"}]}, None, 404, ["Table not found: badtable"], ), # missing primary key ( "/data/docs/-/upsert", {"rows": [{"title": "Missing PK"}]}, None, 400, ['Row 0 is missing primary key column(s): "id"'], ), # Upsert does not support ignore or replace ( "/data/docs/-/upsert", {"rows": [{"id": 1, "title": "Bad"}], "ignore": True}, None, 400, ["Upsert does not support ignore or replace"], ), # Upsert permissions ( "/data/docs/-/upsert", {"rows": [{"id": 1, "title": "Disallowed"}]}, "insert-but-not-update", 403, ["Permission denied: need both insert-row and update-row"], ), ( "/data/docs/-/upsert", {"rows": [{"id": 1, "title": "Disallowed"}]}, "update-but-not-insert", 403, ["Permission denied: need both insert-row and update-row"], ), ), ) async def test_insert_or_upsert_row_errors( ds_write, path, input, special_case, expected_status, expected_errors ): token = write_token(ds_write) if special_case == "insert-but-not-update": token = write_token(ds_write, permissions=["ir", "vi"]) if special_case == "update-but-not-insert": token = write_token(ds_write, permissions=["ur", "vi"]) if special_case == "duplicate_id": await ds_write.get_database("data").execute_write( "insert into docs (id) values (1)" ) if special_case == "bad_token": token += "bad" kwargs = dict( json=input, headers={ "Authorization": "Bearer {}".format(token), "Content-Type": "text/plain" if special_case == "invalid_content_type" else "application/json", }, ) actor_response = ( await ds_write.client.get("/-/actor.json", headers=kwargs["headers"]) ).json() print(actor_response) if special_case == "invalid_json": del kwargs["json"] kwargs["content"] = "{bad json" before_count = ( await ds_write.get_database("data").execute("select count(*) from docs") ).rows[0][0] == 0 response = await ds_write.client.post( path, **kwargs, ) assert response.status_code == expected_status assert response.json()["ok"] is False assert response.json()["errors"] == expected_errors # Check that no rows were inserted after_count = ( await ds_write.get_database("data").execute("select count(*) from docs") ).rows[0][0] == 0 assert before_count == after_count @pytest.mark.asyncio @pytest.mark.parametrize( "ignore,replace,expected_rows", ( ( True, False, [ {"id": 1, "title": "Exists", "score": None, "age": None}, ], ), ( False, True, [ {"id": 1, "title": "One", "score": None, "age": None}, ], ), ), ) @pytest.mark.parametrize("should_return", (True, False)) async def test_insert_ignore_replace( ds_write, ignore, replace, expected_rows, should_return ): await ds_write.get_database("data").execute_write( "insert into docs (id, title) values (1, 'Exists')" ) token = write_token(ds_write) data = {"rows": [{"id": 1, "title": "One"}]} if ignore: data["ignore"] = True if replace: data["replace"] = True if should_return: data["return"] = True response = await ds_write.client.post( "/data/docs/-/insert", json=data, headers=_headers(token), ) assert response.status_code == 201 actual_rows = [ dict(r) for r in ( await ds_write.get_database("data").execute("select * from docs") ).rows ] assert actual_rows == expected_rows assert response.json()["ok"] is True if should_return: assert response.json()["rows"] == expected_rows @pytest.mark.asyncio @pytest.mark.parametrize( "initial,input,expected_rows", ( ( # Simple primary key update {"rows": [{"id": 1, "title": "One"}], "pk": "id"}, {"rows": [{"id": 1, "title": "Two"}]}, [ {"id": 1, "title": "Two"}, ], ), ( # Multiple rows update one of them { "rows": [{"id": 1, "title": "One"}, {"id": 2, "title": "Two"}], "pk": "id", }, {"rows": [{"id": 1, "title": "Three"}]}, [ {"id": 1, "title": "Three"}, {"id": 2, "title": "Two"}, ], ), ( # rowid update {"rows": [{"title": "One"}]}, {"rows": [{"rowid": 1, "title": "Two"}]}, [ {"rowid": 1, "title": "Two"}, ], ), ( # Compound primary key update {"rows": [{"id": 1, "title": "One", "score": 1}], "pks": ["id", "score"]}, {"rows": [{"id": 1, "title": "Two", "score": 1}]}, [ {"id": 1, "title": "Two", "score": 1}, ], ), ), ) @pytest.mark.parametrize("should_return", (False, True)) async def test_upsert(ds_write, initial, input, expected_rows, should_return): token = write_token(ds_write) # Insert initial data initial["table"] = "upsert_test" create_response = await ds_write.client.post( "/data/-/create", json=initial, headers=_headers(token), ) assert create_response.status_code == 201 if should_return: input["return"] = True response = await ds_write.client.post( "/data/upsert_test/-/upsert", json=input, headers=_headers(token), ) assert response.status_code == 200 assert response.json()["ok"] is True if should_return: # We only expect it to return rows corresponding to those we sent expected_returned_rows = expected_rows[: len(input["rows"])] assert response.json()["rows"] == expected_returned_rows # Check the database too actual_rows = ( await ds_write.client.get("/data/upsert_test.json?_shape=array") ).json() assert actual_rows == expected_rows # Drop the upsert_test table await ds_write.get_database("data").execute_write("drop table upsert_test") async def _insert_row(ds): insert_response = await ds.client.post( "/data/docs/-/insert", json={"row": {"title": "Row one", "score": 1.2, "age": 5}, "return": True}, headers=_headers(write_token(ds)), ) assert insert_response.status_code == 201 return insert_response.json()["rows"][0]["id"] @pytest.mark.asyncio @pytest.mark.parametrize("scenario", ("no_token", "no_perm", "bad_table")) async def test_delete_row_errors(ds_write, scenario): if scenario == "no_token": token = "bad_token" elif scenario == "no_perm": token = write_token(ds_write, actor_id="not-root") else: token = write_token(ds_write) pk = await _insert_row(ds_write) path = "/data/{}/{}/-/delete".format( "docs" if scenario != "bad_table" else "bad_table", pk ) response = await ds_write.client.post( path, headers=_headers(token), ) assert response.status_code == 403 if scenario in ("no_token", "bad_token") else 404 assert response.json()["ok"] is False assert ( response.json()["errors"] == ["Permission denied"] if scenario == "no_token" else ["Table not found: bad_table"] ) assert len((await ds_write.client.get("/data/docs.json?_shape=array")).json()) == 1 @pytest.mark.asyncio @pytest.mark.parametrize( "table,row_for_create,pks,delete_path", ( ("rowid_table", {"name": "rowid row"}, None, None), ("pk_table", {"id": 1, "name": "ID table"}, "id", "1"), ( "compound_pk_table", {"type": "article", "key": "k"}, ["type", "key"], "article,k", ), ), ) async def test_delete_row(ds_write, table, row_for_create, pks, delete_path): # First create the table with that example row create_data = { "table": table, "row": row_for_create, } if pks: if isinstance(pks, str): create_data["pk"] = pks else: create_data["pks"] = pks create_response = await ds_write.client.post( "/data/-/create", json=create_data, headers=_headers(write_token(ds_write)), ) assert create_response.status_code == 201, create_response.json() # Should be a single row assert ( await ds_write.client.get( "/data.json?_shape=arrayfirst&sql=select+count(*)+from+{}".format(table) ) ).json() == [1] # Now delete the row if delete_path is None: # Special case for that rowid table delete_path = ( await ds_write.client.get( "/data.json?_shape=arrayfirst&sql=select+rowid+from+{}".format(table) ) ).json()[0] delete_response = await ds_write.client.post( "/data/{}/{}/-/delete".format(table, delete_path), headers=_headers(write_token(ds_write)), ) assert delete_response.status_code == 200 assert ( await ds_write.client.get( "/data.json?_shape=arrayfirst&sql=select+count(*)+from+{}".format(table) ) ).json() == [0] @pytest.mark.asyncio @pytest.mark.parametrize("scenario", ("no_token", "no_perm", "bad_table")) async def test_update_row_check_permission(ds_write, scenario): if scenario == "no_token": token = "bad_token" elif scenario == "no_perm": token = write_token(ds_write, actor_id="not-root") else: token = write_token(ds_write) pk = await _insert_row(ds_write) path = "/data/{}/{}/-/delete".format( "docs" if scenario != "bad_table" else "bad_table", pk ) response = await ds_write.client.post( path, json={"update": {"title": "New title"}}, headers=_headers(token), ) assert response.status_code == 403 if scenario in ("no_token", "bad_token") else 404 assert response.json()["ok"] is False assert ( response.json()["errors"] == ["Permission denied"] if scenario == "no_token" else ["Table not found: bad_table"] ) @pytest.mark.asyncio @pytest.mark.parametrize( "input,expected_errors", ( ({"title": "New title"}, None), ({"title": None}, None), ({"score": 1.6}, None), ({"age": 10}, None), ({"title": "New title", "score": 1.6}, None), ({"title2": "New title"}, ["no such column: title2"]), ), ) @pytest.mark.parametrize("use_return", (True, False)) async def test_update_row(ds_write, input, expected_errors, use_return): token = write_token(ds_write) pk = await _insert_row(ds_write) path = "/data/docs/{}/-/update".format(pk) data = {"update": input} if use_return: data["return"] = True response = await ds_write.client.post( path, json=data, headers=_headers(token), ) if expected_errors: assert response.status_code == 400 assert response.json()["ok"] is False assert response.json()["errors"] == expected_errors return assert response.json()["ok"] is True if not use_return: assert "row" not in response.json() else: returned_row = response.json()["row"] assert returned_row["id"] == pk for k, v in input.items(): assert returned_row[k] == v # And fetch the row to check it's updated response = await ds_write.client.get( "/data/docs/{}.json?_shape=array".format(pk), ) assert response.status_code == 200 row = response.json()[0] assert row["id"] == pk for k, v in input.items(): assert row[k] == v @pytest.mark.asyncio @pytest.mark.parametrize( "scenario", ("no_token", "no_perm", "bad_table", "has_perm", "immutable") ) async def test_drop_table(ds_write, scenario): if scenario == "no_token": token = "bad_token" elif scenario == "no_perm": token = write_token(ds_write, actor_id="not-root") else: token = write_token(ds_write) should_work = scenario == "has_perm" await ds_write.get_database("data").execute_write( "insert into docs (id, title) values (1, 'Row 1')" ) path = "/{database}/{table}/-/drop".format( database="immutable" if scenario == "immutable" else "data", table="docs" if scenario != "bad_table" else "bad_table", ) response = await ds_write.client.post( path, headers=_headers(token), ) if not should_work: assert ( response.status_code == 403 if scenario in ("no_token", "bad_token") else 404 ) assert response.json()["ok"] is False expected_error = "Permission denied" if scenario == "bad_table": expected_error = "Table not found: bad_table" elif scenario == "immutable": expected_error = "Database is immutable" assert response.json()["errors"] == [expected_error] assert (await ds_write.client.get("/data/docs")).status_code == 200 else: # It should show a confirmation page assert response.status_code == 200 assert response.json() == { "ok": True, "database": "data", "table": "docs", "row_count": 1, "message": 'Pass "confirm": true to confirm', } assert (await ds_write.client.get("/data/docs")).status_code == 200 # Now send confirm: true response2 = await ds_write.client.post( path, json={"confirm": True}, headers=_headers(token), ) assert response2.json() == {"ok": True} assert (await ds_write.client.get("/data/docs")).status_code == 404 @pytest.mark.asyncio @pytest.mark.parametrize( "input,expected_status,expected_response", ( # Permission error with a bad token ( {"table": "bad", "row": {"id": 1}}, 403, {"ok": False, "errors": ["Permission denied"]}, ), # Successful creation with columns: ( { "table": "one", "columns": [ { "name": "id", "type": "integer", }, { "name": "title", "type": "text", }, { "name": "score", "type": "integer", }, { "name": "weight", "type": "float", }, { "name": "thumbnail", "type": "blob", }, ], "pk": "id", }, 201, { "ok": True, "database": "data", "table": "one", "table_url": "http://localhost/data/one", "table_api_url": "http://localhost/data/one.json", "schema": ( "CREATE TABLE [one] (\n" " [id] INTEGER PRIMARY KEY,\n" " [title] TEXT,\n" " [score] INTEGER,\n" " [weight] FLOAT,\n" " [thumbnail] BLOB\n" ")" ), }, ), # Successful creation with rows: ( { "table": "two", "rows": [ { "id": 1, "title": "Row 1", "score": 1.5, }, { "id": 2, "title": "Row 2", "score": 1.5, }, ], "pk": "id", }, 201, { "ok": True, "database": "data", "table": "two", "table_url": "http://localhost/data/two", "table_api_url": "http://localhost/data/two.json", "schema": ( "CREATE TABLE [two] (\n" " [id] INTEGER PRIMARY KEY,\n" " [title] TEXT,\n" " [score] FLOAT\n" ")" ), "row_count": 2, }, ), # Successful creation with row: ( { "table": "three", "row": { "id": 1, "title": "Row 1", "score": 1.5, }, "pk": "id", }, 201, { "ok": True, "database": "data", "table": "three", "table_url": "http://localhost/data/three", "table_api_url": "http://localhost/data/three.json", "schema": ( "CREATE TABLE [three] (\n" " [id] INTEGER PRIMARY KEY,\n" " [title] TEXT,\n" " [score] FLOAT\n" ")" ), "row_count": 1, }, ), # Create with row and no primary key ( { "table": "four", "row": { "name": "Row 1", }, }, 201, { "ok": True, "database": "data", "table": "four", "table_url": "http://localhost/data/four", "table_api_url": "http://localhost/data/four.json", "schema": ("CREATE TABLE [four] (\n" " [name] TEXT\n" ")"), "row_count": 1, }, ), # Create table with compound primary key ( { "table": "five", "row": {"type": "article", "key": 123, "title": "Article 1"}, "pks": ["type", "key"], }, 201, { "ok": True, "database": "data", "table": "five", "table_url": "http://localhost/data/five", "table_api_url": "http://localhost/data/five.json", "schema": ( "CREATE TABLE [five] (\n [type] TEXT,\n [key] INTEGER,\n" " [title] TEXT,\n PRIMARY KEY ([type], [key])\n)" ), "row_count": 1, }, ), # Error: Table is required ( { "row": {"id": 1}, }, 400, { "ok": False, "errors": ["Table is required"], }, ), # Error: Invalid table name ( { "table": "sqlite_bad_name", "row": {"id": 1}, }, 400, { "ok": False, "errors": ["Invalid table name"], }, ), # Error: JSON must be an object ( [], 400, { "ok": False, "errors": ["JSON must be an object"], }, ), # Error: Cannot specify columns with rows or row ( { "table": "bad", "columns": [{"name": "id", "type": "integer"}], "rows": [{"id": 1}], }, 400, { "ok": False, "errors": ["Cannot specify columns with rows or row"], }, ), # Error: columns, rows or row is required ( { "table": "bad", }, 400, { "ok": False, "errors": ["columns, rows or row is required"], }, ), # Error: columns must be a list ( { "table": "bad", "columns": {"name": "id", "type": "integer"}, }, 400, { "ok": False, "errors": ["columns must be a list"], }, ), # Error: columns must be a list of objects ( { "table": "bad", "columns": ["id"], }, 400, { "ok": False, "errors": ["columns must be a list of objects"], }, ), # Error: Column name is required ( { "table": "bad", "columns": [{"type": "integer"}], }, 400, { "ok": False, "errors": ["Column name is required"], }, ), # Error: Unsupported column type ( { "table": "bad", "columns": [{"name": "id", "type": "bad"}], }, 400, { "ok": False, "errors": ["Unsupported column type: bad"], }, ), # Error: Duplicate column name ( { "table": "bad", "columns": [ {"name": "id", "type": "integer"}, {"name": "id", "type": "integer"}, ], }, 400, { "ok": False, "errors": ["Duplicate column name: id"], }, ), # Error: rows must be a list ( { "table": "bad", "rows": {"id": 1}, }, 400, { "ok": False, "errors": ["rows must be a list"], }, ), # Error: rows must be a list of objects ( { "table": "bad", "rows": ["id"], }, 400, { "ok": False, "errors": ["rows must be a list of objects"], }, ), # Error: pk must be a string ( { "table": "bad", "row": {"id": 1}, "pk": 1, }, 400, { "ok": False, "errors": ["pk must be a string"], }, ), # Error: Cannot specify both pk and pks ( { "table": "bad", "row": {"id": 1, "name": "Row 1"}, "pk": "id", "pks": ["id", "name"], }, 400, { "ok": False, "errors": ["Cannot specify both pk and pks"], }, ), # Error: pks must be a list ( { "table": "bad", "row": {"id": 1, "name": "Row 1"}, "pks": "id", }, 400, { "ok": False, "errors": ["pks must be a list"], }, ), # Error: pks must be a list of strings ( {"table": "bad", "row": {"id": 1, "name": "Row 1"}, "pks": [1, 2]}, 400, {"ok": False, "errors": ["pks must be a list of strings"]}, ), # Error: ignore and replace are mutually exclusive ( { "table": "bad", "row": {"id": 1, "name": "Row 1"}, "pk": "id", "ignore": True, "replace": True, }, 400, { "ok": False, "errors": ["ignore and replace are mutually exclusive"], }, ), # ignore and replace require row or rows ( { "table": "bad", "columns": [{"name": "id", "type": "integer"}], "ignore": True, }, 400, { "ok": False, "errors": ["ignore and replace require row or rows"], }, ), # ignore and replace require pk or pks ( { "table": "bad", "row": {"id": 1}, "ignore": True, }, 400, { "ok": False, "errors": ["ignore and replace require pk or pks"], }, ), ( { "table": "bad", "row": {"id": 1}, "replace": True, }, 400, { "ok": False, "errors": ["ignore and replace require pk or pks"], }, ), ), ) async def test_create_table(ds_write, input, expected_status, expected_response): # Special case for expected status of 403 if expected_status == 403: token = "bad_token" else: token = write_token(ds_write) response = await ds_write.client.post( "/data/-/create", json=input, headers=_headers(token), ) assert response.status_code == expected_status data = response.json() assert data == expected_response @pytest.mark.asyncio @pytest.mark.parametrize( "permissions,body,expected_status,expected_errors", ( (["create-table"], {"table": "t", "columns": [{"name": "c"}]}, 201, None), # Need insert-row too if you use "rows": ( ["create-table"], {"table": "t", "rows": [{"name": "c"}]}, 403, ["Permission denied - need insert-row"], ), # This should work: ( ["create-table", "insert-row"], {"table": "t", "rows": [{"name": "c"}]}, 201, None, ), # If you use replace: true you need update-row too: ( ["create-table", "insert-row"], {"table": "t", "rows": [{"id": 1}], "pk": "id", "replace": True}, 403, ["Permission denied - need update-row"], ), ), ) async def test_create_table_permissions( ds_write, permissions, body, expected_status, expected_errors ): token = ds_write.create_token("root", restrict_all=["view-instance"] + permissions) response = await ds_write.client.post( "/data/-/create", json=body, headers=_headers(token), ) assert response.status_code == expected_status if expected_errors: data = response.json() assert data["ok"] is False assert data["errors"] == expected_errors @pytest.mark.asyncio @pytest.mark.parametrize( "input,expected_rows_after", ( ( { "table": "test_insert_replace", "rows": [ {"id": 1, "name": "Row 1 new"}, {"id": 3, "name": "Row 3 new"}, ], "pk": "id", "ignore": True, }, [ {"id": 1, "name": "Row 1"}, {"id": 2, "name": "Row 2"}, {"id": 3, "name": "Row 3 new"}, ], ), ( { "table": "test_insert_replace", "rows": [ {"id": 1, "name": "Row 1 new"}, {"id": 3, "name": "Row 3 new"}, ], "pk": "id", "replace": True, }, [ {"id": 1, "name": "Row 1 new"}, {"id": 2, "name": "Row 2"}, {"id": 3, "name": "Row 3 new"}, ], ), ), ) async def test_create_table_ignore_replace(ds_write, input, expected_rows_after): # Create table with two rows token = write_token(ds_write) first_response = await ds_write.client.post( "/data/-/create", json={ "rows": [{"id": 1, "name": "Row 1"}, {"id": 2, "name": "Row 2"}], "table": "test_insert_replace", "pk": "id", }, headers=_headers(token), ) assert first_response.status_code == 201 # Try a second time second_response = await ds_write.client.post( "/data/-/create", json=input, headers=_headers(token), ) assert second_response.status_code == 201 # Check that the rows are as expected rows = await ds_write.client.get("/data/test_insert_replace.json?_shape=array") assert rows.json() == expected_rows_after @pytest.mark.asyncio async def test_create_table_error_if_pk_changed(ds_write): token = write_token(ds_write) first_response = await ds_write.client.post( "/data/-/create", json={ "rows": [{"id": 1, "name": "Row 1"}, {"id": 2, "name": "Row 2"}], "table": "test_insert_replace", "pk": "id", }, headers=_headers(token), ) assert first_response.status_code == 201 # Try a second time with a different pk second_response = await ds_write.client.post( "/data/-/create", json={ "rows": [{"id": 1, "name": "Row 1"}, {"id": 2, "name": "Row 2"}], "table": "test_insert_replace", "pk": "name", "replace": True, }, headers=_headers(token), ) assert second_response.status_code == 400 assert second_response.json() == { "ok": False, "errors": ["pk cannot be changed for existing table"], } @pytest.mark.asyncio async def test_create_table_error_rows_twice_with_duplicates(ds_write): # Error if you don't send ignore: True or replace: True token = write_token(ds_write) input = { "rows": [{"id": 1, "name": "Row 1"}, {"id": 2, "name": "Row 2"}], "table": "test_create_twice", "pk": "id", } first_response = await ds_write.client.post( "/data/-/create", json=input, headers=_headers(token), ) assert first_response.status_code == 201 second_response = await ds_write.client.post( "/data/-/create", json=input, headers=_headers(token), ) assert second_response.status_code == 400 assert second_response.json() == { "ok": False, "errors": ["UNIQUE constraint failed: test_create_twice.id"], } @pytest.mark.asyncio @pytest.mark.parametrize( "path", ( "/data/-/create", "/data/docs/-/drop", "/data/docs/-/insert", ), ) async def test_method_not_allowed(ds_write, path): response = await ds_write.client.get( path, headers={ "Content-Type": "application/json", }, ) assert response.status_code == 405 assert response.json() == { "ok": False, "error": "Method not allowed", }