2019-07-06 00:05:56 +00:00
|
|
|
import os
|
2020-06-07 05:30:36 +00:00
|
|
|
import pathlib
|
2019-07-06 00:05:56 +00:00
|
|
|
import pytest
|
2020-06-07 05:30:36 +00:00
|
|
|
import re
|
2021-02-12 00:52:16 +00:00
|
|
|
import subprocess
|
|
|
|
import tempfile
|
|
|
|
import time
|
|
|
|
import trustme
|
|
|
|
|
2020-06-07 05:30:36 +00:00
|
|
|
|
2020-11-30 21:29:57 +00:00
|
|
|
try:
|
|
|
|
import pysqlite3 as sqlite3
|
|
|
|
except ImportError:
|
|
|
|
import sqlite3
|
|
|
|
|
2020-06-07 05:30:36 +00:00
|
|
|
UNDOCUMENTED_PERMISSIONS = {
|
|
|
|
"this_is_allowed",
|
|
|
|
"this_is_denied",
|
|
|
|
"this_is_allowed_async",
|
|
|
|
"this_is_denied_async",
|
|
|
|
"no_match",
|
|
|
|
}
|
2019-07-06 00:05:56 +00:00
|
|
|
|
|
|
|
|
2020-11-30 21:29:57 +00:00
|
|
|
def pytest_report_header(config):
|
|
|
|
return "SQLite: {}".format(
|
|
|
|
sqlite3.connect(":memory:").execute("select sqlite_version()").fetchone()[0]
|
|
|
|
)
|
|
|
|
|
|
|
|
|
2019-05-02 05:10:23 +00:00
|
|
|
def pytest_configure(config):
|
|
|
|
import sys
|
|
|
|
|
|
|
|
sys._called_from_test = True
|
|
|
|
|
|
|
|
|
|
|
|
def pytest_unconfigure(config):
|
|
|
|
import sys
|
|
|
|
|
|
|
|
del sys._called_from_test
|
2019-05-04 02:15:14 +00:00
|
|
|
|
|
|
|
|
|
|
|
def pytest_collection_modifyitems(items):
|
2020-08-15 20:38:15 +00:00
|
|
|
# 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")
|
2019-05-11 21:45:59 +00:00
|
|
|
move_to_front(items, "test_black")
|
|
|
|
move_to_front(items, "test_inspect_cli")
|
2020-08-15 20:38:15 +00:00
|
|
|
move_to_front(items, "test_serve_with_get")
|
2020-09-11 22:04:23 +00:00
|
|
|
move_to_front(items, "test_serve_with_get_exit_code_for_error")
|
2019-05-11 22:03:52 +00:00
|
|
|
move_to_front(items, "test_inspect_cli_writes_to_file")
|
2019-05-11 23:22:55 +00:00
|
|
|
move_to_front(items, "test_spatialite_error_if_attempt_to_open_spatialite")
|
2020-01-29 22:46:43 +00:00
|
|
|
move_to_front(items, "test_package")
|
|
|
|
move_to_front(items, "test_package_with_port")
|
2019-05-11 21:45:59 +00:00
|
|
|
|
|
|
|
|
|
|
|
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])))
|
2019-07-06 00:05:56 +00:00
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
def restore_working_directory(tmpdir, request):
|
|
|
|
previous_cwd = os.getcwd()
|
|
|
|
tmpdir.chdir()
|
|
|
|
|
|
|
|
def return_to_previous():
|
|
|
|
os.chdir(previous_cwd)
|
|
|
|
|
|
|
|
request.addfinalizer(return_to_previous)
|
2020-06-07 05:30:36 +00:00
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
|
|
def check_permission_actions_are_documented():
|
|
|
|
from datasette.plugins import pm
|
|
|
|
|
|
|
|
content = (
|
2021-03-11 16:15:49 +00:00
|
|
|
pathlib.Path(__file__).parent.parent / "docs" / "authentication.rst"
|
|
|
|
).read_text()
|
2020-06-07 05:30:36 +00:00
|
|
|
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
|
2020-06-08 18:59:11 +00:00
|
|
|
), "Undocumented permission action: {}, resource: {}".format(
|
|
|
|
action, kwargs["resource"]
|
2020-06-07 05:30:36 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
pm.add_hookcall_monitoring(
|
|
|
|
before=before, after=lambda outcome, hook_name, hook_impls, kwargs: None
|
|
|
|
)
|
2021-02-12 00:52:16 +00:00
|
|
|
|
|
|
|
|
|
|
|
@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(),
|
|
|
|
)
|
|
|
|
# Give the server time to start
|
|
|
|
time.sleep(1.5)
|
|
|
|
# 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_localhost_https_server(tmp_path_factory):
|
|
|
|
cert_directory = tmp_path_factory.mktemp("certs")
|
|
|
|
ca = trustme.CA()
|
|
|
|
server_cert = ca.issue_cert("localhost")
|
|
|
|
keyfile = str(cert_directory / "server.key")
|
|
|
|
certfile = str(cert_directory / "server.pem")
|
|
|
|
client_cert = str(cert_directory / "client.pem")
|
|
|
|
server_cert.private_key_pem.write_to_path(path=keyfile)
|
|
|
|
for blob in server_cert.cert_chain_pems:
|
|
|
|
blob.write_to_path(path=certfile, append=True)
|
|
|
|
ca.cert_pem.write_to_path(path=client_cert)
|
|
|
|
|
|
|
|
ds_proc = subprocess.Popen(
|
|
|
|
[
|
|
|
|
"datasette",
|
|
|
|
"--memory",
|
|
|
|
"-p",
|
|
|
|
"8042",
|
|
|
|
"--ssl-keyfile",
|
|
|
|
keyfile,
|
|
|
|
"--ssl-certfile",
|
|
|
|
certfile,
|
|
|
|
],
|
|
|
|
stdout=subprocess.PIPE,
|
|
|
|
stderr=subprocess.STDOUT,
|
|
|
|
cwd=tempfile.gettempdir(),
|
|
|
|
)
|
|
|
|
# Give the server time to start
|
|
|
|
time.sleep(1.5)
|
|
|
|
# Check it started successfully
|
|
|
|
assert not ds_proc.poll(), ds_proc.stdout.read().decode("utf-8")
|
|
|
|
yield ds_proc, client_cert
|
|
|
|
# Shut it down at the end of the pytest session
|
|
|
|
ds_proc.terminate()
|