datasette-edit-schema/tests/test_edit_schema.py
import datasette
from datasette.app import Datasette
import sqlite_utils
import pytest
import json
import httpx
import re
whitespace = re.compile(r"\s+")
@pytest.fixture
def db_path(tmpdir):
path = str(tmpdir / "data.db")
db = sqlite_utils.Database(path)
db["creatures"].insert_all(
[
{"name": "Cleo", "description": "A medium sized dog"},
{"name": "Siroco", "description": "A troublesome Kakapo"},
]
)
db["other_table"].insert({"foo": "bar"})
return path
@pytest.mark.asyncio
async def test_csrf_required(db_path):
ds = Datasette([db_path])
async with httpx.AsyncClient(app=ds.app()) as client:
response = await client.post(
"http://localhost/-/edit-schema/data/creatures",
data={"delete_table": "1"},
allow_redirects=False,
cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
)
assert 403 == response.status_code
@pytest.mark.parametrize("authenticate", [True, False])
@pytest.mark.asyncio
async def test_table_actions(db_path, authenticate):
ds = Datasette([db_path])
async with httpx.AsyncClient(app=ds.app()) as client:
cookies = None
if authenticate:
cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
response = await client.get("http://localhost/data/creatures", cookies=cookies)
assert response.status_code == 200
fragment = (
'<li><a href="/-/edit-schema/data/creatures">Edit table schema</a></li>'
)
if authenticate:
# Should have column actions
assert fragment in response.text
else:
assert fragment not in response.text
@pytest.mark.asyncio
async def test_post_without_operation_raises_error(db_path):
ds = Datasette([db_path])
cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
async with httpx.AsyncClient(app=ds.app()) as client:
# Get a csrftoken
csrftoken = (
await client.get(
"http://localhost/-/edit-schema/data/creatures", cookies=cookies
)
).cookies["ds_csrftoken"]
cookies["ds_csrftoken"] = csrftoken
response = await client.post(
"http://localhost/-/edit-schema/data/creatures",
data={"csrftoken": csrftoken},
allow_redirects=False,
cookies=cookies,
)
assert 400 == response.status_code
@pytest.mark.asyncio
async def test_delete_table(db_path):
ds = Datasette([db_path])
cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
db = sqlite_utils.Database(db_path)
assert "creatures" in db.table_names()
async with httpx.AsyncClient(app=ds.app()) as client:
# Get a csrftoken
csrftoken = (
await client.get(
"http://localhost/-/edit-schema/data/creatures", cookies=cookies
)
).cookies["ds_csrftoken"]
response = await client.post(
"http://localhost/-/edit-schema/data/creatures",
data={"delete_table": "1", "csrftoken": csrftoken},
allow_redirects=False,
cookies=cookies,
)
assert 302 == response.status_code
assert "creatures" not in db.table_names()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"col_type,expected_type",
[("text", str), ("integer", int), ("real", float), ("blob", bytes)],
)
async def test_add_column(db_path, col_type, expected_type):
ds = Datasette([db_path])
db = sqlite_utils.Database(db_path)
cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
table = db["creatures"]
assert {"name": str, "description": str} == table.columns_dict
async with httpx.AsyncClient(app=ds.app()) as client:
# Get a csrftoken
csrftoken = (
await client.get(
"http://localhost/-/edit-schema/data/creatures", cookies=cookies
)
).cookies["ds_csrftoken"]
response = await client.post(
"http://localhost/-/edit-schema/data/creatures",
data={
"add_column": "1",
"csrftoken": csrftoken,
"name": "new_col",
"type": col_type,
},
allow_redirects=False,
cookies=cookies,
)
assert 302 == response.status_code
if "ds_messages" in response.cookies:
messages = ds.unsign(response.cookies["ds_messages"], "messages")
# None of these should be errors
assert all(m[1] == Datasette.INFO for m in messages), "Got an error: {}".format(
messages
)
assert {
"name": str,
"description": str,
"new_col": expected_type,
} == table.columns_dict
@pytest.mark.asyncio
@pytest.mark.parametrize(
"name,type,expected_error",
[
("name", "text", "A column called 'name' already exists"),
("", "text", "Column name is required"),
("]]]", "integer", 'unrecognized token: "]"'),
("name", "blop", "Invalid type: blop"),
],
)
async def test_add_column_errors(db_path, name, type, expected_error):
ds = Datasette([db_path])
cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
async with httpx.AsyncClient(app=ds.app()) as client:
csrftoken = (
await client.get(
"http://localhost/-/edit-schema/data/creatures", cookies=cookies
)
).cookies["ds_csrftoken"]
response = await client.post(
"http://localhost/-/edit-schema/data/creatures",
data={
"add_column": "1",
"name": name,
"type": type,
"csrftoken": csrftoken,
},
allow_redirects=False,
cookies=cookies,
)
response.status_code == 302
assert response.headers["location"] == "/-/edit-schema/data/creatures"
messages = ds.unsign(response.cookies["ds_messages"], "messages")
assert len(messages) == 1
assert messages[0][0] == expected_error
@pytest.mark.asyncio
@pytest.mark.parametrize(
"post_data,expected_columns_dict,expected_order",
[
# Change column type
(
{
"type.name": "REAL",
},
{"rowid": int, "name": float, "description": str},
["rowid", "name", "description"],
),
(
{
"type.name": "INTEGER",
},
{"rowid": int, "name": int, "description": str},
["rowid", "name", "description"],
),
# Changing order
(
{
"sort.description": "0",
"sort.name": "2",
},
{"rowid": int, "name": str, "description": str},
["rowid", "description", "name"],
),
# Change names
(
{
"name.name": "name2",
"name.description": "description2",
},
{"rowid": int, "name2": str, "description2": str},
["rowid", "name2", "description2"],
),
],
)
async def test_transform_table(
db_path, post_data, expected_columns_dict, expected_order
):
ds = Datasette([db_path])
cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
db = sqlite_utils.Database(db_path)
table = db["creatures"]
assert table.columns_dict == {"name": str, "description": str}
async with httpx.AsyncClient(app=ds.app()) as client:
csrftoken = (
await client.get(
"http://localhost/-/edit-schema/data/creatures", cookies=cookies
)
).cookies["ds_csrftoken"]
post_data["csrftoken"] = csrftoken
post_data["action"] = "update_columns"
response = await client.post(
"http://localhost/-/edit-schema/data/creatures",
data=post_data,
allow_redirects=False,
cookies=cookies,
)
assert 302 == response.status_code
assert table.columns_dict == expected_columns_dict
assert [c.name for c in table.columns] == expected_order
@pytest.mark.asyncio
async def test_static_assets(db_path):
ds = Datasette([db_path])
async with httpx.AsyncClient(app=ds.app()) as client:
for path in (
"/-/static-plugins/datasette-edit-schema/draggable.1.0.0-beta.11.bundle.min.js",
):
response = await client.post(
"http://localhost" + path,
)
assert response.status_code == 200
@pytest.mark.asyncio
@pytest.mark.parametrize(
"path", ["/-/edit-schema", "/-/edit-schema/data", "/-/edit-schema/data/creatures"]
)
async def test_permission_check(db_path, path):
ds = Datasette([db_path])
someuser_cookies = {"ds_actor": ds.sign({"a": {"id": "someuser"}}, "actor")}
root_cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
async with httpx.AsyncClient(app=ds.app()) as client:
response = await client.get(
"http://localhost" + path,
)
assert response.status_code == 403
# Should deny with someuser cookie
response = await client.get("http://localhost" + path, cookies=someuser_cookies)
assert response.status_code == 403
# Should allow with root cookies
response = await client.get(
"http://localhost" + path, cookies=root_cookies, allow_redirects=False
)
assert response.status_code in (200, 302)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"new_name,should_work,expected_message",
[
("valid", True, "Table renamed to 'valid'"),
("]]]", False, 'Error renaming table: unrecognized token: "]"'),
("creatures", True, "Table name was the same"),
("", False, "New table name is required"),
("other_table", False, "A table called 'other_table' already exists"),
],
)
async def test_rename_table(db_path, new_name, should_work, expected_message):
ds = Datasette([db_path])
cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
async with httpx.AsyncClient(app=ds.app()) as client:
csrftoken = (
await client.get(
"http://localhost/-/edit-schema/data/creatures", cookies=cookies
)
).cookies["ds_csrftoken"]
response = await client.post(
"http://localhost/-/edit-schema/data/creatures",
data={
"rename_table": "1",
"name": new_name,
"csrftoken": csrftoken,
},
allow_redirects=False,
cookies=cookies,
)
response.status_code == 302
if should_work:
expected_path = "/-/edit-schema/data/{}".format(new_name)
else:
expected_path = "/-/edit-schema/data/creatures"
assert response.headers["location"] == expected_path
messages = ds.unsign(response.cookies["ds_messages"], "messages")
assert len(messages) == 1
assert messages[0][0] == expected_message