datasette-configure-fts/tests/test_configure_fts.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"},
]
)
return path
@pytest.fixture
def db_path2(tmpdir):
path = str(tmpdir / "data2.db")
db = sqlite_utils.Database(path)
db["creatures"].insert({"name": "Pancakes"})
db["dogs"].insert({"name": "Pancakes"})
db["mammals"].insert({"name": "Pancakes"})
return path
@pytest.mark.asyncio
async def test_initial_db_is_not_searchable(db_path):
ds = Datasette([db_path])
response = await ds.client.get("/data.json")
assert response.status_code == 200
tables = json.loads(response.content)["tables"]
assert len(tables) == 1
@pytest.mark.parametrize("path", ["/-/configure-fts", "/-/configure-fts/data"])
@pytest.mark.asyncio
async def test_permissions(db_path, path):
ds = Datasette([db_path])
response = await ds.client.get(path)
response.status_code == 403
# Now try with a root actor
response2 = await ds.client.get(
path,
cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
)
assert response2.status_code != 403
@pytest.mark.asyncio
async def test_redirects_to_database_if_only_one(db_path):
ds = Datasette([db_path])
response = await ds.client.get(
"/-/configure-fts",
cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
)
assert response.status_code == 302
assert response.headers["location"] == "/-/configure-fts/data"
@pytest.mark.asyncio
async def test_database_page_sets_cookie(db_path):
ds = Datasette([db_path])
response = await ds.client.get(
"/-/configure-fts/data",
cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
)
assert "ds_csrftoken" in response.cookies
@pytest.mark.asyncio
async def test_lists_databases_if_more_than_one(db_path, db_path2):
ds = Datasette([db_path, db_path2])
response = await ds.client.get(
"/-/configure-fts",
cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
)
assert response.status_code == 200
assert '<a href="/-/configure-fts/data">data</a>' in response.text
assert '<a href="/-/configure-fts/data2">data2</a>' in response.text
@pytest.mark.asyncio
async def test_lists_tables_in_database(db_path2):
ds = Datasette([db_path2])
response = await ds.client.get(
"/-/configure-fts/data2",
cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
)
assert response.status_code == 200
assert "<h2>creatures</h2>" in response.text
assert "<h2>dogs</h2>" in response.text
assert "<h2>mammals</h2>" in response.text
# If we select just two tables, only those two
response2 = await ds.client.get(
"/-/configure-fts/data2?table=dogs&table=mammals",
cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
)
assert "<h2>creatures</h2>" not in response2.text
assert "<h2>dogs</h2>" in response2.text
assert "<h2>mammals</h2>" in response2.text
@pytest.mark.asyncio
async def test_text_columns_only(db_path):
ds = Datasette([db_path])
sqlite_utils.Database(db_path)["mixed_types"].insert(
{
"name": "text",
"age": 5,
"height": 1.4,
"description": "description",
}
)
response = await ds.client.get(
"/-/configure-fts/data?table=mixed_types",
cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
)
assert response.status_code == 200
content = response.text
assert 'name="column.name"' in content
assert 'name="column.age"' not in content
assert 'name="column.height"' not in content
assert 'name="column.description"' in content
@pytest.mark.asyncio
async def test_make_table_searchable(db_path):
ds = Datasette([db_path])
response1 = await ds.client.get(
"/-/configure-fts/data",
cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
)
csrftoken = response1.cookies["ds_csrftoken"]
response2 = await ds.client.post(
"/-/configure-fts/data",
data={
"csrftoken": csrftoken,
"table": "creatures",
"column.name": "on",
"column.description": "on",
},
cookies={
"ds_actor": ds.sign({"a": {"id": "root"}}, "actor"),
"ds_csrftoken": csrftoken,
},
)
assert response2.status_code == 302
assert response2.headers["location"] == "/data/creatures"
db = sqlite_utils.Database(db_path)
assert [
"creatures",
"creatures_fts",
"creatures_fts_data",
"creatures_fts_idx",
"creatures_fts_docsize",
"creatures_fts_config",
] == db.table_names()
assert (
"CREATE VIRTUAL TABLE [creatures_fts] USING FTS5 ( [name], [description], content=[creatures] )"
== whitespace.sub(" ", db["creatures_fts"].schema)
)
# It should have set up triggers
rows = db.conn.execute(
'select name from sqlite_master where type = "trigger" order by name'
).fetchall()
assert rows == [("creatures_ad",), ("creatures_ai",), ("creatures_au",)]
@pytest.mark.asyncio
async def test_uncheck_all_columns(db_path):
ds = Datasette([db_path])
db = sqlite_utils.Database(db_path)
db["creatures"].enable_fts(["name"])
response1 = await ds.client.get(
"/-/configure-fts/data",
cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
)
csrftoken = response1.cookies["ds_csrftoken"]
response2 = await ds.client.post(
"/-/configure-fts/data",
data={
"csrftoken": csrftoken,
"table": "creatures",
},
cookies={
"ds_actor": ds.sign({"a": {"id": "root"}}, "actor"),
"ds_csrftoken": csrftoken,
},
)
assert response2.status_code == 302
assert response2.headers["location"] == "/data/creatures"
db = sqlite_utils.Database(db_path)
assert db.table_names() == ["creatures"]
@pytest.mark.parametrize("authenticate", [True, False])
@pytest.mark.asyncio
async def test_table_actions(db_path, authenticate):
ds = Datasette([db_path])
cookies = None
if authenticate:
cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
response = await ds.client.get("/data/creatures", cookies=cookies)
assert response.status_code == 200
fragment = '<li><a href="/-/configure-fts/data?table=creatures">Configure full-text search</a></li>'
if authenticate:
# Should have column actions
assert fragment in response.text
else:
assert fragment not in response.text