github/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2025-12-10 16:51:24 +01:00
Code Issues Projects Releases Packages Wiki Activity

datasette.pm property, closes #2595

Browse source
This commit is contained in:
Simon Willison 2025-11-13 10:31:03 -08:00
commit 4b4add4d31
11 changed files with 101 additions and 89 deletions
Download patch file Download diff file Expand all files Collapse all files

16
datasette/app.py
View file

@ -631,6 +631,17 @@ class Datasette:
def urls(self): def urls(self):
return Urls(self) return Urls(self)
@property
def pm(self):
"""
Return the global plugin manager instance.
This provides access to the pluggy PluginManager that manages all
Datasette plugins and hooks. Use datasette.pm.hook.hook_name() to
call plugin hooks.
"""
return pm
async def invoke_startup(self): async def invoke_startup(self):
# This must be called for Datasette to be in a usable state # This must be called for Datasette to be in a usable state
if self._startup_invoked: if self._startup_invoked:
@ -2415,7 +2426,10 @@ class DatasetteClient:
def __init__(self, ds): def __init__(self, ds):
self.ds = ds self.ds = ds
self.app = ds.app()
@property
def app(self):
return self.ds.app()
def actor_cookie(self, actor): def actor_cookie(self, actor):
# Utility method, mainly for tests # Utility method, mainly for tests

19
datasette/plugins.py
View file

@ -94,21 +94,24 @@ def get_plugins():
for plugin in pm.get_plugins(): for plugin in pm.get_plugins():
static_path = None static_path = None
templates_path = None templates_path = None
if plugin.__name__ not in DEFAULT_PLUGINS: plugin_name = (
plugin.__name__
if hasattr(plugin, "__name__")
else plugin.__class__.__name__
)
if plugin_name not in DEFAULT_PLUGINS:
try: try:
if (importlib_resources.files(plugin.__name__) / "static").is_dir(): if (importlib_resources.files(plugin_name) / "static").is_dir():
static_path = str( static_path = str(importlib_resources.files(plugin_name) / "static")
importlib_resources.files(plugin.__name__) / "static" if (importlib_resources.files(plugin_name) / "templates").is_dir():
)
if (importlib_resources.files(plugin.__name__) / "templates").is_dir():
templates_path = str( templates_path = str(
importlib_resources.files(plugin.__name__) / "templates" importlib_resources.files(plugin_name) / "templates"
) )
except (TypeError, ModuleNotFoundError): except (TypeError, ModuleNotFoundError):
# Caused by --plugins_dir= plugins # Caused by --plugins_dir= plugins
pass pass
plugin_info = { plugin_info = {
"name": plugin.__name__, "name": plugin_name,
"static_path": static_path, "static_path": static_path,
"templates_path": templates_path, "templates_path": templates_path,
"hooks": [h.name for h in pm.get_hookcallers(plugin)], "hooks": [h.name for h in pm.get_hookcallers(plugin)],

2
docs/internals.rst
View file

@ -1093,7 +1093,7 @@ Example usage:
if not datasette.in_client(): if not datasette.in_client():
return Response.text( return Response.text(
"Only available via internal client requests", "Only available via internal client requests",
status=403 status=403,
) )
... ...

9
docs/testing_plugins.rst
View file

@ -283,13 +283,12 @@ Here's a test for that plugin that mocks the HTTPX outbound request:
Registering a plugin for the duration of a test Registering a plugin for the duration of a test
----------------------------------------------- -----------------------------------------------
When writing tests for plugins you may find it useful to register a test plugin just for the duration of a single test. You can do this using ``pm.register()`` and ``pm.unregister()`` like this: When writing tests for plugins you may find it useful to register a test plugin just for the duration of a single test. You can do this using ``datasette.pm.register()`` and ``datasette.pm.unregister()`` like this:
.. code-block:: python .. code-block:: python
from datasette import hookimpl from datasette import hookimpl
from datasette.app import Datasette from datasette.app import Datasette
from datasette.plugins import pm
import pytest import pytest
@ -305,14 +304,14 @@ When writing tests for plugins you may find it useful to register a test plugin
(r"^/error$", lambda: 1 / 0), (r"^/error$", lambda: 1 / 0),
] ]
pm.register(TestPlugin(), name="undo") datasette = Datasette()
try: try:
# The test implementation goes here # The test implementation goes here
datasette = Datasette() datasette.pm.register(TestPlugin(), name="undo")
response = await datasette.client.get("/error") response = await datasette.client.get("/error")
assert response.status_code == 500 assert response.status_code == 500
finally: finally:
pm.unregister(name="undo") datasette.pm.unregister(name="undo")
To reuse the same temporary plugin in multiple tests, you can register it inside a fixture in your ``conftest.py`` file like this: To reuse the same temporary plugin in multiple tests, you can register it inside a fixture in your ``conftest.py`` file like this:

25
tests/test_actions_sql.py
View file

@ -11,7 +11,6 @@ These tests verify:
import pytest import pytest
import pytest_asyncio import pytest_asyncio
from datasette.app import Datasette from datasette.app import Datasette
from datasette.plugins import pm
from datasette.permissions import PermissionSQL from datasette.permissions import PermissionSQL
from datasette.resources import TableResource from datasette.resources import TableResource
from datasette import hookimpl from datasette import hookimpl
@ -67,7 +66,7 @@ async def test_allowed_resources_global_allow(test_ds):
return None return None
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
# Use the new allowed_resources() method # Use the new allowed_resources() method
@ -87,7 +86,7 @@ async def test_allowed_resources_global_allow(test_ds):
assert ("production", "orders") in table_set assert ("production", "orders") in table_set
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -106,7 +105,7 @@ async def test_allowed_specific_resource(test_ds):
return None return None
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
actor = {"id": "bob", "role": "analyst"} actor = {"id": "bob", "role": "analyst"}
@ -130,7 +129,7 @@ async def test_allowed_specific_resource(test_ds):
) )
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -148,7 +147,7 @@ async def test_allowed_resources_include_reasons(test_ds):
return None return None
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
# Use allowed_resources with include_reasons to get debugging info # Use allowed_resources with include_reasons to get debugging info
@ -170,7 +169,7 @@ async def test_allowed_resources_include_reasons(test_ds):
assert "analyst access" in reasons_text assert "analyst access" in reasons_text
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -190,7 +189,7 @@ async def test_child_deny_overrides_parent_allow(test_ds):
return None return None
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
actor = {"id": "bob", "role": "analyst"} actor = {"id": "bob", "role": "analyst"}
@ -219,7 +218,7 @@ async def test_child_deny_overrides_parent_allow(test_ds):
) )
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -239,7 +238,7 @@ async def test_child_allow_overrides_parent_deny(test_ds):
return None return None
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
actor = {"id": "carol"} actor = {"id": "carol"}
@ -264,7 +263,7 @@ async def test_child_allow_overrides_parent_deny(test_ds):
) )
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -288,7 +287,7 @@ async def test_sql_does_filtering_not_python(test_ds):
return PermissionSQL(sql=sql) return PermissionSQL(sql=sql)
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
actor = {"id": "dave"} actor = {"id": "dave"}
@ -314,4 +313,4 @@ async def test_sql_does_filtering_not_python(test_ds):
assert tables[0].child == "users" assert tables[0].child == "users"
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")

25
tests/test_allowed_resources.py
View file

@ -8,7 +8,6 @@ based on permission rules from plugins and configuration.
import pytest import pytest
import pytest_asyncio import pytest_asyncio
from datasette.app import Datasette from datasette.app import Datasette
from datasette.plugins import pm
from datasette.permissions import PermissionSQL from datasette.permissions import PermissionSQL
from datasette import hookimpl from datasette import hookimpl
@ -62,7 +61,7 @@ async def test_tables_endpoint_global_access(test_ds):
return None return None
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
# Use the allowed_resources API directly # Use the allowed_resources API directly
@ -87,7 +86,7 @@ async def test_tables_endpoint_global_access(test_ds):
assert "production/orders" in table_names assert "production/orders" in table_names
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -102,7 +101,7 @@ async def test_tables_endpoint_database_restriction(test_ds):
return None return None
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
page = await test_ds.allowed_resources( page = await test_ds.allowed_resources(
@ -130,7 +129,7 @@ async def test_tables_endpoint_database_restriction(test_ds):
# Note: default_permissions.py provides default allows, so we just check analytics are present # Note: default_permissions.py provides default allows, so we just check analytics are present
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -149,7 +148,7 @@ async def test_tables_endpoint_table_exception(test_ds):
return None return None
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
page = await test_ds.allowed_resources("view-table", {"id": "carol"}) page = await test_ds.allowed_resources("view-table", {"id": "carol"})
@ -172,7 +171,7 @@ async def test_tables_endpoint_table_exception(test_ds):
assert "analytics/sensitive" not in table_names assert "analytics/sensitive" not in table_names
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -191,7 +190,7 @@ async def test_tables_endpoint_deny_overrides_allow(test_ds):
return None return None
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
page = await test_ds.allowed_resources( page = await test_ds.allowed_resources(
@ -214,7 +213,7 @@ async def test_tables_endpoint_deny_overrides_allow(test_ds):
assert "analytics/sensitive" not in table_names assert "analytics/sensitive" not in table_names
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -257,7 +256,7 @@ async def test_tables_endpoint_specific_table_only(test_ds):
return None return None
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
page = await test_ds.allowed_resources("view-table", {"id": "dave"}) page = await test_ds.allowed_resources("view-table", {"id": "dave"})
@ -280,7 +279,7 @@ async def test_tables_endpoint_specific_table_only(test_ds):
assert "production/orders" in table_names assert "production/orders" in table_names
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -295,7 +294,7 @@ async def test_tables_endpoint_empty_result(test_ds):
return None return None
plugin = PermissionRulesPlugin(rules_callback) plugin = PermissionRulesPlugin(rules_callback)
pm.register(plugin, name="test_plugin") test_ds.pm.register(plugin, name="test_plugin")
try: try:
page = await test_ds.allowed_resources("view-table", {"id": "blocked"}) page = await test_ds.allowed_resources("view-table", {"id": "blocked"})
@ -311,7 +310,7 @@ async def test_tables_endpoint_empty_result(test_ds):
assert len(result) == 0 assert len(result) == 0
finally: finally:
pm.unregister(plugin, name="test_plugin") test_ds.pm.unregister(plugin, name="test_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio

8
tests/test_docs_plugins.py
View file

@ -2,7 +2,6 @@
# -- start datasette_with_plugin_fixture -- # -- start datasette_with_plugin_fixture --
from datasette import hookimpl from datasette import hookimpl
from datasette.app import Datasette from datasette.app import Datasette
from datasette.plugins import pm
import pytest import pytest
import pytest_asyncio import pytest_asyncio
@ -18,11 +17,12 @@ async def datasette_with_plugin():
(r"^/error$", lambda: 1 / 0), (r"^/error$", lambda: 1 / 0),
] ]
pm.register(TestPlugin(), name="undo") datasette = Datasette()
datasette.pm.register(TestPlugin(), name="undo")
try: try:
yield Datasette() yield datasette
finally: finally:
pm.unregister(name="undo") datasette.pm.unregister(name="undo")
# -- end datasette_with_plugin_fixture -- # -- end datasette_with_plugin_fixture --

18
tests/test_internals_datasette_client.py
View file

@ -239,7 +239,6 @@ async def test_in_client_returns_false_outside_request(datasette):
async def test_in_client_returns_true_inside_request(): async def test_in_client_returns_true_inside_request():
"""Test that datasette.in_client() returns True inside a client request""" """Test that datasette.in_client() returns True inside a client request"""
from datasette import hookimpl, Response from datasette import hookimpl, Response
from datasette.plugins import pm
class TestPlugin: class TestPlugin:
__name__ = "test_in_client_plugin" __name__ = "test_in_client_plugin"
@ -255,10 +254,10 @@ async def test_in_client_returns_true_inside_request():
(r"^/-/test-in-client$", test_view), (r"^/-/test-in-client$", test_view),
] ]
pm.register(TestPlugin(), name="test_in_client_plugin") ds = Datasette()
await ds.invoke_startup()
ds.pm.register(TestPlugin(), name="test_in_client_plugin")
try: try:
ds = Datasette()
await ds.invoke_startup()
# Outside of a client request, should be False # Outside of a client request, should be False
assert ds.in_client() is False assert ds.in_client() is False
@ -271,14 +270,13 @@ async def test_in_client_returns_true_inside_request():
# After the request, should be False again # After the request, should be False again
assert ds.in_client() is False assert ds.in_client() is False
finally: finally:
pm.unregister(name="test_in_client_plugin") ds.pm.unregister(name="test_in_client_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_in_client_with_skip_permission_checks(): async def test_in_client_with_skip_permission_checks():
"""Test that in_client() works regardless of skip_permission_checks value""" """Test that in_client() works regardless of skip_permission_checks value"""
from datasette import hookimpl from datasette import hookimpl
from datasette.plugins import pm
from datasette.utils.asgi import Response from datasette.utils.asgi import Response
in_client_values = [] in_client_values = []
@ -296,10 +294,10 @@ async def test_in_client_with_skip_permission_checks():
(r"^/-/test-in-client$", test_view), (r"^/-/test-in-client$", test_view),
] ]
pm.register(TestPlugin(), name="test_in_client_skip_plugin") ds = Datasette(config={"databases": {"test_db": {"allow": {"id": "admin"}}}})
await ds.invoke_startup()
ds.pm.register(TestPlugin(), name="test_in_client_skip_plugin")
try: try:
ds = Datasette(config={"databases": {"test_db": {"allow": {"id": "admin"}}}})
await ds.invoke_startup()
# Request without skip_permission_checks # Request without skip_permission_checks
await ds.client.get("/-/test-in-client") await ds.client.get("/-/test-in-client")
@ -312,4 +310,4 @@ async def test_in_client_with_skip_permission_checks():
), f"Expected 2 values, got {len(in_client_values)}" ), f"Expected 2 values, got {len(in_client_values)}"
assert all(in_client_values), f"Expected all True, got {in_client_values}" assert all(in_client_values), f"Expected all True, got {in_client_values}"
finally: finally:
pm.unregister(name="test_in_client_skip_plugin") ds.pm.unregister(name="test_in_client_skip_plugin")

10
tests/test_permission_endpoints.py
View file

@ -439,7 +439,6 @@ async def test_execute_sql_requires_view_database():
be able to execute SQL on that database. be able to execute SQL on that database.
""" """
from datasette.permissions import PermissionSQL from datasette.permissions import PermissionSQL
from datasette.plugins import pm
from datasette import hookimpl from datasette import hookimpl
class TestPermissionPlugin: class TestPermissionPlugin:
@ -464,11 +463,12 @@ async def test_execute_sql_requires_view_database():
return [] return []
plugin = TestPermissionPlugin() plugin = TestPermissionPlugin()
pm.register(plugin, name="test_plugin")
ds = Datasette()
await ds.invoke_startup()
ds.pm.register(plugin, name="test_plugin")
try: try:
ds = Datasette()
await ds.invoke_startup()
ds.add_memory_database("secret") ds.add_memory_database("secret")
await ds.refresh_schemas() await ds.refresh_schemas()
@ -498,4 +498,4 @@ async def test_execute_sql_requires_view_database():
f"but got {response.status_code}" f"but got {response.status_code}"
) )
finally: finally:
pm.unregister(plugin) ds.pm.unregister(plugin)

38
tests/test_plugins.py
View file

@ -691,7 +691,7 @@ async def test_hook_permission_resources_sql():
await ds.invoke_startup() await ds.invoke_startup()
collected = [] collected = []
for block in pm.hook.permission_resources_sql( for block in ds.pm.hook.permission_resources_sql(
datasette=ds, datasette=ds,
actor={"id": "alice"}, actor={"id": "alice"},
action="view-table", action="view-table",
@ -1161,12 +1161,12 @@ async def test_hook_filters_from_request(ds_client):
if request.args.get("_nothing"): if request.args.get("_nothing"):
return FilterArguments(["1 = 0"], human_descriptions=["NOTHING"]) return FilterArguments(["1 = 0"], human_descriptions=["NOTHING"])
pm.register(ReturnNothingPlugin(), name="ReturnNothingPlugin") ds_client.ds.pm.register(ReturnNothingPlugin(), name="ReturnNothingPlugin")
response = await ds_client.get("/fixtures/facetable?_nothing=1") response = await ds_client.get("/fixtures/facetable?_nothing=1")
assert "0 rows\n where NOTHING" in response.text assert "0 rows\n where NOTHING" in response.text
json_response = await ds_client.get("/fixtures/facetable.json?_nothing=1") json_response = await ds_client.get("/fixtures/facetable.json?_nothing=1")
assert json_response.json()["rows"] == [] assert json_response.json()["rows"] == []
pm.unregister(name="ReturnNothingPlugin") ds_client.ds.pm.unregister(name="ReturnNothingPlugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -1327,7 +1327,7 @@ async def test_hook_actors_from_ids():
return inner return inner
try: try:
pm.register(ActorsFromIdsPlugin(), name="ActorsFromIdsPlugin") ds.pm.register(ActorsFromIdsPlugin(), name="ActorsFromIdsPlugin")
actors2 = await ds.actors_from_ids(["3", "5", "7"]) actors2 = await ds.actors_from_ids(["3", "5", "7"])
assert actors2 == { assert actors2 == {
"3": {"id": "3", "name": "Cate Blanchett"}, "3": {"id": "3", "name": "Cate Blanchett"},
@ -1335,7 +1335,7 @@ async def test_hook_actors_from_ids():
"7": {"id": "7", "name": "Sarah Paulson"}, "7": {"id": "7", "name": "Sarah Paulson"},
} }
finally: finally:
pm.unregister(name="ReturnNothingPlugin") ds.pm.unregister(name="ReturnNothingPlugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -1350,14 +1350,14 @@ async def test_plugin_is_installed():
return {} return {}
try: try:
pm.register(DummyPlugin(), name="DummyPlugin") datasette.pm.register(DummyPlugin(), name="DummyPlugin")
response = await datasette.client.get("/-/plugins.json") response = await datasette.client.get("/-/plugins.json")
assert response.status_code == 200 assert response.status_code == 200
installed_plugins = {p["name"] for p in response.json()} installed_plugins = {p["name"] for p in response.json()}
assert "DummyPlugin" in installed_plugins assert "DummyPlugin" in installed_plugins
finally: finally:
pm.unregister(name="DummyPlugin") datasette.pm.unregister(name="DummyPlugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -1384,7 +1384,7 @@ async def test_hook_jinja2_environment_from_request(tmpdir):
datasette = Datasette(memory=True) datasette = Datasette(memory=True)
try: try:
pm.register(EnvironmentPlugin(), name="EnvironmentPlugin") datasette.pm.register(EnvironmentPlugin(), name="EnvironmentPlugin")
response = await datasette.client.get("/") response = await datasette.client.get("/")
assert response.status_code == 200 assert response.status_code == 200
assert "Hello museums!" not in response.text assert "Hello museums!" not in response.text
@ -1395,7 +1395,7 @@ async def test_hook_jinja2_environment_from_request(tmpdir):
assert response2.status_code == 200 assert response2.status_code == 200
assert "Hello museums!" in response2.text assert "Hello museums!" in response2.text
finally: finally:
pm.unregister(name="EnvironmentPlugin") datasette.pm.unregister(name="EnvironmentPlugin")
class SlotPlugin: class SlotPlugin:
@ -1433,48 +1433,48 @@ class SlotPlugin:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_hook_top_homepage(): async def test_hook_top_homepage():
datasette = Datasette(memory=True)
try: try:
pm.register(SlotPlugin(), name="SlotPlugin") datasette.pm.register(SlotPlugin(), name="SlotPlugin")
datasette = Datasette(memory=True)
response = await datasette.client.get("/?z=foo") response = await datasette.client.get("/?z=foo")
assert response.status_code == 200 assert response.status_code == 200
assert "Xtop_homepage:foo" in response.text assert "Xtop_homepage:foo" in response.text
finally: finally:
pm.unregister(name="SlotPlugin") datasette.pm.unregister(name="SlotPlugin")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_hook_top_database(): async def test_hook_top_database():
datasette = Datasette(memory=True)
try: try:
pm.register(SlotPlugin(), name="SlotPlugin") datasette.pm.register(SlotPlugin(), name="SlotPlugin")
datasette = Datasette(memory=True)
response = await datasette.client.get("/_memory?z=bar") response = await datasette.client.get("/_memory?z=bar")
assert response.status_code == 200 assert response.status_code == 200
assert "Xtop_database:_memory:bar" in response.text assert "Xtop_database:_memory:bar" in response.text
finally: finally:
pm.unregister(name="SlotPlugin") datasette.pm.unregister(name="SlotPlugin")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_hook_top_table(ds_client): async def test_hook_top_table(ds_client):
try: try:
pm.register(SlotPlugin(), name="SlotPlugin") ds_client.ds.pm.register(SlotPlugin(), name="SlotPlugin")
response = await ds_client.get("/fixtures/facetable?z=baz") response = await ds_client.get("/fixtures/facetable?z=baz")
assert response.status_code == 200 assert response.status_code == 200
assert "Xtop_table:fixtures:facetable:baz" in response.text assert "Xtop_table:fixtures:facetable:baz" in response.text
finally: finally:
pm.unregister(name="SlotPlugin") ds_client.ds.pm.unregister(name="SlotPlugin")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_hook_top_row(ds_client): async def test_hook_top_row(ds_client):
try: try:
pm.register(SlotPlugin(), name="SlotPlugin") ds_client.ds.pm.register(SlotPlugin(), name="SlotPlugin")
response = await ds_client.get("/fixtures/facet_cities/1?z=bax") response = await ds_client.get("/fixtures/facet_cities/1?z=bax")
assert response.status_code == 200 assert response.status_code == 200
assert "Xtop_row:fixtures:facet_cities:San Francisco:bax" in response.text assert "Xtop_row:fixtures:facet_cities:San Francisco:bax" in response.text
finally: finally:
pm.unregister(name="SlotPlugin") ds_client.ds.pm.unregister(name="SlotPlugin")
@pytest.mark.asyncio @pytest.mark.asyncio

20
tests/test_restriction_sql.py
View file

@ -13,7 +13,6 @@ async def test_multiple_restriction_sources_intersect():
provide restriction_sql - both must pass for access to be granted. provide restriction_sql - both must pass for access to be granted.
""" """
from datasette import hookimpl from datasette import hookimpl
from datasette.plugins import pm
class RestrictivePlugin: class RestrictivePlugin:
__name__ = "RestrictivePlugin" __name__ = "RestrictivePlugin"
@ -29,11 +28,12 @@ async def test_multiple_restriction_sources_intersect():
return None return None
plugin = RestrictivePlugin() plugin = RestrictivePlugin()
pm.register(plugin, name="restrictive_plugin")
ds = Datasette()
await ds.invoke_startup()
ds.pm.register(plugin, name="restrictive_plugin")
try: try:
ds = Datasette()
await ds.invoke_startup()
db1 = ds.add_memory_database("db1_multi_intersect") db1 = ds.add_memory_database("db1_multi_intersect")
db2 = ds.add_memory_database("db2_multi_intersect") db2 = ds.add_memory_database("db2_multi_intersect")
await db1.execute_write("CREATE TABLE t1 (id INTEGER)") await db1.execute_write("CREATE TABLE t1 (id INTEGER)")
@ -55,7 +55,7 @@ async def test_multiple_restriction_sources_intersect():
assert ("db1_multi_intersect", "t1") in resources assert ("db1_multi_intersect", "t1") in resources
assert ("db2_multi_intersect", "t1") not in resources assert ("db2_multi_intersect", "t1") not in resources
finally: finally:
pm.unregister(name="restrictive_plugin") ds.pm.unregister(name="restrictive_plugin")
@pytest.mark.asyncio @pytest.mark.asyncio
@ -265,7 +265,6 @@ async def test_permission_resources_sql_multiple_restriction_sources_intersect()
provide restriction_sql - both must pass for access to be granted. provide restriction_sql - both must pass for access to be granted.
""" """
from datasette import hookimpl from datasette import hookimpl
from datasette.plugins import pm
class RestrictivePlugin: class RestrictivePlugin:
__name__ = "RestrictivePlugin" __name__ = "RestrictivePlugin"
@ -281,11 +280,12 @@ async def test_permission_resources_sql_multiple_restriction_sources_intersect()
return None return None
plugin = RestrictivePlugin() plugin = RestrictivePlugin()
pm.register(plugin, name="restrictive_plugin")
ds = Datasette()
await ds.invoke_startup()
ds.pm.register(plugin, name="restrictive_plugin")
try: try:
ds = Datasette()
await ds.invoke_startup()
db1 = ds.add_memory_database("db1_multi_restrictions") db1 = ds.add_memory_database("db1_multi_restrictions")
db2 = ds.add_memory_database("db2_multi_restrictions") db2 = ds.add_memory_database("db2_multi_restrictions")
await db1.execute_write("CREATE TABLE t1 (id INTEGER)") await db1.execute_write("CREATE TABLE t1 (id INTEGER)")
@ -312,4 +312,4 @@ async def test_permission_resources_sql_multiple_restriction_sources_intersect()
assert ("db1_multi_restrictions", "t1") in resources assert ("db1_multi_restrictions", "t1") in resources
assert ("db2_multi_restrictions", "t1") not in resources assert ("db2_multi_restrictions", "t1") not in resources
finally: finally:
pm.unregister(name="restrictive_plugin") ds.pm.unregister(name="restrictive_plugin")