1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
"""
Tests for the datasette.app.Datasette class
"""
from datasette import Forbidden
from datasette.app import Datasette, Database
from itsdangerous import BadSignature
from .fixtures import app_client
import pytest
@pytest.fixture
def datasette(app_client):
return app_client.ds
def test_get_database(datasette):
db = datasette.get_database("fixtures")
assert "fixtures" == db.name
with pytest.raises(KeyError):
datasette.get_database("missing")
def test_get_database_no_argument(datasette):
# Returns the first available database:
db = datasette.get_database()
assert "fixtures" == db.name
@pytest.mark.parametrize("value", ["hello", 123, {"key": "value"}])
@pytest.mark.parametrize("namespace", [None, "two"])
def test_sign_unsign(datasette, value, namespace):
extra_args = [namespace] if namespace else []
signed = datasette.sign(value, *extra_args)
assert value != signed
assert value == datasette.unsign(signed, *extra_args)
with pytest.raises(BadSignature):
datasette.unsign(signed[:-1] + ("!" if signed[-1] != "!" else ":"))
@pytest.mark.parametrize(
"setting,expected",
(
("base_url", "/"),
("max_csv_mb", 100),
("allow_csv_stream", True),
),
)
def test_datasette_setting(datasette, setting, expected):
assert datasette.setting(setting) == expected
@pytest.mark.asyncio
async def test_datasette_constructor():
ds = Datasette()
databases = (await ds.client.get("/-/databases.json")).json()
assert databases == [
{
"name": "_memory",
"route": "_memory",
"path": None,
"size": 0,
"is_mutable": True,
"is_memory": True,
"hash": None,
}
]
@pytest.mark.asyncio
async def test_num_sql_threads_zero():
ds = Datasette([], memory=True, settings={"num_sql_threads": 0})
db = ds.add_database(Database(ds, memory_name="test_num_sql_threads_zero"))
await db.execute_write("create table t(id integer primary key)")
await db.execute_write("insert into t (id) values (1)")
response = await ds.client.get("/-/threads.json")
assert response.json() == {"num_threads": 0, "threads": []}
response2 = await ds.client.get("/test_num_sql_threads_zero/t.json?_shape=array")
assert response2.json() == [{"id": 1}]
ROOT = {"id": "root"}
ALLOW_ROOT = {"allow": {"id": "root"}}
@pytest.mark.asyncio
@pytest.mark.parametrize(
"actor,metadata,permissions,should_allow,expected_private",
(
(None, ALLOW_ROOT, ["view-instance"], False, False),
(ROOT, ALLOW_ROOT, ["view-instance"], True, True),
(
None,
{"databases": {"_memory": ALLOW_ROOT}},
[("view-database", "_memory")],
False,
False,
),
(
ROOT,
{"databases": {"_memory": ALLOW_ROOT}},
[("view-database", "_memory")],
True,
True,
),
# Check private is false for non-protected instance check
(
ROOT,
{"allow": True},
["view-instance"],
True,
False,
),
),
)
async def test_datasette_ensure_permissions_check_visibility(
actor, metadata, permissions, should_allow, expected_private
):
ds = Datasette([], memory=True, metadata=metadata)
if not should_allow:
with pytest.raises(Forbidden):
await ds.ensure_permissions(actor, permissions)
else:
await ds.ensure_permissions(actor, permissions)
# And try check_visibility too:
visible, private = await ds.check_visibility(actor, permissions=permissions)
assert visible == should_allow
assert private == expected_private
@pytest.mark.asyncio
async def test_datasette_render_template_no_request():
# https://github.com/simonw/datasette/issues/1849
ds = Datasette(memory=True)
await ds.invoke_startup()
rendered = await ds.render_template("error.html")
assert "Error " in rendered
def test_datasette_error_if_string_not_list(tmpdir):
# https://github.com/simonw/datasette/issues/1985
db_path = str(tmpdir / "data.db")
with pytest.raises(ValueError):
ds = Datasette(db_path)
|
