datasette-auth-tokens/tests/test_auth_tokens.py
from datasette.app import Datasette
import pytest
import pytest_asyncio
import sqlite_utils
@pytest_asyncio.fixture
async def ds(tmp_path_factory):
db_directory = tmp_path_factory.mktemp("dbs")
db_path1 = db_directory / "demo.db"
sqlite_utils.Database(db_path1)["foo"].insert({"bar": 1})
db_path2 = db_directory / "tokens.db"
db = sqlite_utils.Database(db_path2)
db["tokens"].insert_all(
[
{
"id": 1,
"actor_id": "one",
"actor_name": "Cleo",
"token_secret": "oneone",
},
{
"id": 2,
"actor_id": "two",
"actor_name": "Pancakes",
"token_secret": "twotwo",
},
],
pk="id",
)
return Datasette(
[db_path1, db_path2],
metadata={
"plugins": {
"datasette-auth-tokens": {
"query": {
"sql": (
"select actor_id, actor_name, token_secret "
"from tokens where id = :token_id"
),
"database": "tokens",
},
"tokens": [
{"token": "one", "actor": {"id": "one"}},
{"token": "two", "actor": {"id": "two"}},
],
"param": "_auth_token",
}
},
"databases": {
"demo": {"allow_sql": {"id": "one"}},
"tokens": {"allow": {}},
},
},
)
@pytest.mark.parametrize(
"token,path,expected_status",
[
("", "/", 200),
("", "/demo?sql=select+1", 403),
("one", "/", 200),
("one", "/demo?sql=select+1", 200),
("two", "/", 200),
("two", "/demo?sql=select+1", 403),
],
)
@pytest.mark.asyncio
async def test_token(ds, token, path, expected_status):
response = await ds.client.get(
path,
headers={"Authorization": "Bearer {}".format(token)},
)
assert response.status_code == expected_status
@pytest.mark.parametrize(
"token,path,expected_status",
[
("", "/?", 200),
("", "/demo?sql=select+1", 403),
("one", "/?", 200),
("one", "/demo?sql=select+1", 200),
("two", "/?", 200),
("two", "/demo?sql=select+1", 403),
],
)
@pytest.mark.asyncio
async def test_query_param(ds, token, path, expected_status):
response = await ds.client.get(
"{}&_auth_token={}".format(path, token),
)
assert response.status_code == expected_status
@pytest.mark.parametrize(
"token,path,expected_status",
[
("", "/", 200),
("", "/demo?sql=select+1", 403),
("1-oneone", "/", 200),
("1-oneone", "/demo?sql=select+1", 200),
("2-twotwo", "/", 200),
("2-twotwo", "/demo?sql=select+1", 403),
("invalid", "/", 200),
("invalid", "/demo?sql=select+1", 403),
],
)
@pytest.mark.asyncio
async def test_query(ds, token, path, expected_status):
response = await ds.client.get(
path,
headers={"Authorization": "Bearer {}".format(token)},
)
assert response.status_code == expected_status
@pytest.mark.parametrize(
"token,expected_actor",
[
("1-oneone", {"id": "one", "name": "Cleo"}),
("2-twotwo", {"id": "two", "name": "Pancakes"}),
("invalid", None),
("invalid", None),
],
)
@pytest.mark.asyncio
async def test_actor(ds, token, expected_actor):
response = await ds.client.get(
"/-/actor.json",
headers={"Authorization": "Bearer {}".format(token)},
)
assert response.json() == {"actor": expected_actor}
@pytest.mark.parametrize(
"path",
[
"/tokens",
"/tokens/tokens",
"/tokens?sql=select+*+from+tokens",
],
)
@pytest.mark.asyncio
async def test_tokens_table_not_visible(ds, path):
response = await ds.client.get(path)
assert response.status_code == 403