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
145
146
147
148
149
150
151
152
153
154
|
import httpx
import os
import pathlib
import pytest
import re
import subprocess
import tempfile
import time
import trustme
try:
import pysqlite3 as sqlite3
except ImportError:
import sqlite3
UNDOCUMENTED_PERMISSIONS = {
"this_is_allowed",
"this_is_denied",
"this_is_allowed_async",
"this_is_denied_async",
"no_match",
}
def wait_until_responds(url, timeout=5.0, client=httpx, **kwargs):
start = time.time()
while time.time() - start < timeout:
try:
client.get(url, **kwargs)
return
except httpx.ConnectError:
time.sleep(0.1)
raise AssertionError("Timed out waiting for {} to respond".format(url))
def pytest_report_header(config):
return "SQLite: {}".format(
sqlite3.connect(":memory:").execute("select sqlite_version()").fetchone()[0]
)
def pytest_configure(config):
import sys
sys._called_from_test = True
def pytest_unconfigure(config):
import sys
del sys._called_from_test
def pytest_collection_modifyitems(items):
# Ensure test_cli.py and test_black.py and test_inspect.py run first before any asyncio code kicks in
move_to_front(items, "test_cli")
move_to_front(items, "test_black")
move_to_front(items, "test_inspect_cli")
move_to_front(items, "test_serve_with_get")
move_to_front(items, "test_serve_with_get_exit_code_for_error")
move_to_front(items, "test_inspect_cli_writes_to_file")
move_to_front(items, "test_spatialite_error_if_attempt_to_open_spatialite")
move_to_front(items, "test_package")
move_to_front(items, "test_package_with_port")
def move_to_front(items, test_name):
test = [fn for fn in items if fn.name == test_name]
if test:
items.insert(0, items.pop(items.index(test[0])))
@pytest.fixture
def restore_working_directory(tmpdir, request):
try:
previous_cwd = os.getcwd()
except OSError:
# https://github.com/simonw/datasette/issues/1361
previous_cwd = None
tmpdir.chdir()
def return_to_previous():
os.chdir(previous_cwd)
if previous_cwd is not None:
request.addfinalizer(return_to_previous)
@pytest.fixture(scope="session", autouse=True)
def check_permission_actions_are_documented():
from datasette.plugins import pm
content = (
pathlib.Path(__file__).parent.parent / "docs" / "authentication.rst"
).read_text()
permissions_re = re.compile(r"\.\. _permissions_([^\s:]+):")
documented_permission_actions = set(permissions_re.findall(content)).union(
UNDOCUMENTED_PERMISSIONS
)
def before(hook_name, hook_impls, kwargs):
if hook_name == "permission_allowed":
action = kwargs.get("action").replace("-", "_")
assert (
action in documented_permission_actions
), "Undocumented permission action: {}, resource: {}".format(
action, kwargs["resource"]
)
pm.add_hookcall_monitoring(
before=before, after=lambda outcome, hook_name, hook_impls, kwargs: None
)
@pytest.fixture(scope="session")
def ds_localhost_http_server():
ds_proc = subprocess.Popen(
["datasette", "--memory", "-p", "8041"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
# Avoid FileNotFoundError: [Errno 2] No such file or directory:
cwd=tempfile.gettempdir(),
)
wait_until_responds("http://localhost:8041/")
# Check it started successfully
assert not ds_proc.poll(), ds_proc.stdout.read().decode("utf-8")
yield ds_proc
# Shut it down at the end of the pytest session
ds_proc.terminate()
@pytest.fixture(scope="session")
def ds_unix_domain_socket_server(tmp_path_factory):
# This used to use tmp_path_factory.mktemp("uds") but that turned out to
# produce paths that were too long to use as UDS on macOS, see
# https://github.com/simonw/datasette/issues/1407 - so I switched to
# using tempfile.gettempdir()
uds = str(pathlib.Path(tempfile.gettempdir()) / "datasette.sock")
ds_proc = subprocess.Popen(
["datasette", "--memory", "--uds", uds],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
cwd=tempfile.gettempdir(),
)
# Poll until available
transport = httpx.HTTPTransport(uds=uds)
client = httpx.Client(transport=transport)
wait_until_responds("http://localhost/_memory.json", client=client)
# Check it started successfully
assert not ds_proc.poll(), ds_proc.stdout.read().decode("utf-8")
yield ds_proc, uds
# Shut it down at the end of the pytest session
ds_proc.terminate()
|
