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

Run black formatter

Browse source
This commit is contained in:
Simon Willison 2025-10-25 08:45:10 -07:00
commit 60a38cee85
14 changed files with 222 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

20
tests/plugins/my_plugin.py
View file

@ -495,7 +495,7 @@ def register_actions(datasette):
takes_parent=True,
takes_child=False,
resource_class=DatabaseResource,
)
),
]
# Support old-style config for backwards compatibility
@ -509,7 +509,11 @@ def register_actions(datasette):
description=p["description"],
takes_parent=p.get("takes_database", False),
takes_child=p.get("takes_resource", False),
resource_class=DatabaseResource if p.get("takes_database") else InstanceResource,
resource_class=(
DatabaseResource
if p.get("takes_database")
else InstanceResource
),
)
)
@ -521,7 +525,9 @@ def register_actions(datasette):
"InstanceResource": InstanceResource,
"DatabaseResource": DatabaseResource,
}
resource_class = resource_class_map.get(a.get("resource_class", "InstanceResource"), InstanceResource)
resource_class = resource_class_map.get(
a.get("resource_class", "InstanceResource"), InstanceResource
)
actions.append(
Action(
@ -561,7 +567,13 @@ def permission_resources_sql(datasette, actor, action):
else:
return None # No opinion
return PermissionSQL(source="my_plugin", sql=sql, params={})
elif action in ("insert-row", "create-table", "drop-table", "delete-row", "update-row"):
elif action in (
"insert-row",
"create-table",
"drop-table",
"delete-row",
"update-row",
):
# Special permissions for latest.datasette.io demos
actor_id = actor.get("id") if actor else None
if actor_id == "todomvc":

19
tests/test_auth.py
View file

@ -358,18 +358,12 @@ async def test_root_with_root_enabled_gets_all_permissions(ds_client):
# Test instance-level permissions (no resource)
assert (
await ds_client.ds.allowed(action="permissions-debug", actor=root_actor)
is True
await ds_client.ds.allowed(action="permissions-debug", actor=root_actor) is True
)
assert await ds_client.ds.allowed(action="debug-menu", actor=root_actor) is True
# Test view permissions using the new ds.allowed() method
assert (
await ds_client.ds.allowed(
action="view-instance", actor=root_actor
)
is True
)
assert await ds_client.ds.allowed(action="view-instance", actor=root_actor) is True
assert (
await ds_client.ds.allowed(
@ -462,10 +456,7 @@ async def test_root_without_root_enabled_no_special_permissions(ds_client):
# View permissions should still work (default=True)
assert (
await ds_client.ds.allowed(
action="view-instance", actor=root_actor
)
is True
await ds_client.ds.allowed(action="view-instance", actor=root_actor) is True
) # Default permission
assert (
@ -479,9 +470,7 @@ async def test_root_without_root_enabled_no_special_permissions(ds_client):
# But restricted permissions should NOT automatically be granted
# Test with instance-level permission (no resource class)
result = await ds_client.ds.allowed(
action="permissions-debug", actor=root_actor
)
result = await ds_client.ds.allowed(action="permissions-debug", actor=root_actor)
assert (
result is not True
), "Root without root_enabled should not automatically get permissions-debug"

4
tests/test_canned_queries.py
View file

@ -5,7 +5,9 @@ import re
from .fixtures import make_app_client, app_client
# Mark entire module as xfail since view-query permission not yet migrated, refs #2534
pytestmark = pytest.mark.xfail(reason="view-query permission not yet migrated to new permission system, refs #2534")
pytestmark = pytest.mark.xfail(
reason="view-query permission not yet migrated to new permission system, refs #2534"
)
@pytest.fixture

74
tests/test_config_permission_rules.py
View file

@ -19,8 +19,16 @@ async def test_root_permissions_allow():
config = {"permissions": {"execute-sql": {"id": "alice"}}}
ds = await setup_datasette(config=config, databases=["content"])
assert await ds.allowed(action="execute-sql", resource=DatabaseResource(database="content"), actor={"id": "alice"})
assert not await ds.allowed(action="execute-sql", resource=DatabaseResource(database="content"), actor={"id": "bob"})
assert await ds.allowed(
action="execute-sql",
resource=DatabaseResource(database="content"),
actor={"id": "alice"},
)
assert not await ds.allowed(
action="execute-sql",
resource=DatabaseResource(database="content"),
actor={"id": "bob"},
)
@pytest.mark.asyncio
@ -37,10 +45,14 @@ async def test_database_permission():
ds = await setup_datasette(config=config, databases=["content"])
assert await ds.allowed(
action="insert-row", resource=TableResource(database="content", table="repos"), actor={"id": "alice"}
action="insert-row",
resource=TableResource(database="content", table="repos"),
actor={"id": "alice"},
)
assert not await ds.allowed(
action="insert-row", resource=TableResource(database="content", table="repos"), actor={"id": "bob"}
action="insert-row",
resource=TableResource(database="content", table="repos"),
actor={"id": "bob"},
)
@ -56,10 +68,14 @@ async def test_table_permission():
ds = await setup_datasette(config=config, databases=["content"])
assert await ds.allowed(
action="delete-row", resource=TableResource(database="content", table="repos"), actor={"id": "alice"}
action="delete-row",
resource=TableResource(database="content", table="repos"),
actor={"id": "alice"},
)
assert not await ds.allowed(
action="delete-row", resource=TableResource(database="content", table="repos"), actor={"id": "bob"}
action="delete-row",
resource=TableResource(database="content", table="repos"),
actor={"id": "bob"},
)
@ -71,13 +87,19 @@ async def test_view_table_allow_block():
ds = await setup_datasette(config=config, databases=["content"])
assert await ds.allowed(
action="view-table", resource=TableResource(database="content", table="repos"), actor={"id": "alice"}
action="view-table",
resource=TableResource(database="content", table="repos"),
actor={"id": "alice"},
)
assert not await ds.allowed(
action="view-table", resource=TableResource(database="content", table="repos"), actor={"id": "bob"}
action="view-table",
resource=TableResource(database="content", table="repos"),
actor={"id": "bob"},
)
assert await ds.allowed(
action="view-table", resource=TableResource(database="content", table="other"), actor={"id": "bob"}
action="view-table",
resource=TableResource(database="content", table="other"),
actor={"id": "bob"},
)
@ -87,7 +109,9 @@ async def test_view_table_allow_false_blocks():
ds = await setup_datasette(config=config, databases=["content"])
assert not await ds.allowed(
action="view-table", resource=TableResource(database="content", table="repos"), actor={"id": "alice"}
action="view-table",
resource=TableResource(database="content", table="repos"),
actor={"id": "alice"},
)
@ -96,18 +120,38 @@ async def test_allow_sql_blocks():
config = {"allow_sql": {"id": "alice"}}
ds = await setup_datasette(config=config, databases=["content"])
assert await ds.allowed(action="execute-sql", resource=DatabaseResource(database="content"), actor={"id": "alice"})
assert not await ds.allowed(action="execute-sql", resource=DatabaseResource(database="content"), actor={"id": "bob"})
assert await ds.allowed(
action="execute-sql",
resource=DatabaseResource(database="content"),
actor={"id": "alice"},
)
assert not await ds.allowed(
action="execute-sql",
resource=DatabaseResource(database="content"),
actor={"id": "bob"},
)
config = {"databases": {"content": {"allow_sql": {"id": "bob"}}}}
ds = await setup_datasette(config=config, databases=["content"])
assert await ds.allowed(action="execute-sql", resource=DatabaseResource(database="content"), actor={"id": "bob"})
assert not await ds.allowed(action="execute-sql", resource=DatabaseResource(database="content"), actor={"id": "alice"})
assert await ds.allowed(
action="execute-sql",
resource=DatabaseResource(database="content"),
actor={"id": "bob"},
)
assert not await ds.allowed(
action="execute-sql",
resource=DatabaseResource(database="content"),
actor={"id": "alice"},
)
config = {"allow_sql": False}
ds = await setup_datasette(config=config, databases=["content"])
assert not await ds.allowed(action="execute-sql", resource=DatabaseResource(database="content"), actor={"id": "alice"})
assert not await ds.allowed(
action="execute-sql",
resource=DatabaseResource(database="content"),
actor={"id": "alice"},
)
@pytest.mark.asyncio

28
tests/test_html.py
View file

@ -135,7 +135,9 @@ def test_not_allowed_methods():
@pytest.mark.asyncio
@pytest.mark.xfail(reason="Canned queries not displayed due to view-query permission, refs #2510")
@pytest.mark.xfail(
reason="Canned queries not displayed due to view-query permission, refs #2510"
)
async def test_database_page(ds_client):
response = await ds_client.get("/fixtures")
soup = Soup(response.text, "html.parser")
@ -264,7 +266,7 @@ def test_query_page_truncates():
pytest.param(
"/fixtures/neighborhood_search",
["query", "db-fixtures", "query-neighborhood_search"],
marks=pytest.mark.xfail(reason="Canned queries not accessible, refs #2510")
marks=pytest.mark.xfail(reason="Canned queries not accessible, refs #2510"),
),
(
"/fixtures/table~2Fwith~2Fslashes~2Ecsv",
@ -597,7 +599,9 @@ async def test_404_content_type(ds_client):
@pytest.mark.asyncio
@pytest.mark.xfail(reason="Canned queries not yet migrated to new permission system, refs #2510")
@pytest.mark.xfail(
reason="Canned queries not yet migrated to new permission system, refs #2510"
)
async def test_canned_query_default_title(ds_client):
response = await ds_client.get("/fixtures/magic_parameters")
assert response.status_code == 200
@ -606,7 +610,9 @@ async def test_canned_query_default_title(ds_client):
@pytest.mark.asyncio
@pytest.mark.xfail(reason="Canned queries not yet migrated to new permission system, refs #2510")
@pytest.mark.xfail(
reason="Canned queries not yet migrated to new permission system, refs #2510"
)
async def test_canned_query_with_custom_metadata(ds_client):
response = await ds_client.get("/fixtures/neighborhood_search?text=town")
assert response.status_code == 200
@ -669,7 +675,9 @@ async def test_show_hide_sql_query(ds_client):
@pytest.mark.asyncio
@pytest.mark.xfail(reason="Canned queries not yet migrated to new permission system, refs #2510")
@pytest.mark.xfail(
reason="Canned queries not yet migrated to new permission system, refs #2510"
)
async def test_canned_query_with_hide_has_no_hidden_sql(ds_client):
# For a canned query the show/hide should NOT have a hidden SQL field
# https://github.com/simonw/datasette/issues/1411
@ -681,7 +689,9 @@ async def test_canned_query_with_hide_has_no_hidden_sql(ds_client):
] == [(hidden["name"], hidden["value"]) for hidden in hiddens]
@pytest.mark.xfail(reason="Canned queries not yet migrated to new permission system, refs #2510")
@pytest.mark.xfail(
reason="Canned queries not yet migrated to new permission system, refs #2510"
)
@pytest.mark.parametrize(
"hide_sql,querystring,expected_hidden,expected_show_hide_link,expected_show_hide_text",
(
@ -931,7 +941,9 @@ def test_base_url_affects_metadata_extra_css_urls(app_client_base_url_prefix):
("/fixtures/magic_parameters", None),
],
)
@pytest.mark.xfail(reason="Canned queries not yet migrated to new permission system, refs #2510")
@pytest.mark.xfail(
reason="Canned queries not yet migrated to new permission system, refs #2510"
)
async def test_edit_sql_link_on_canned_queries(ds_client, path, expected):
response = await ds_client.get(path)
assert response.status_code == 200
@ -1037,7 +1049,7 @@ async def test_trace_correctly_escaped(ds_client):
pytest.param(
"/fixtures/neighborhood_search?text=town",
"http://localhost/fixtures/neighborhood_search.json?text=town",
marks=pytest.mark.xfail(reason="Canned queries not accessible, refs #2510")
marks=pytest.mark.xfail(reason="Canned queries not accessible, refs #2510"),
),
# /-/ pages
(

4
tests/test_internals_datasette.py
View file

@ -121,7 +121,9 @@ async def test_datasette_check_visibility(
):
ds = Datasette([], memory=True, config=config)
await ds.invoke_startup()
visible, private = await ds.check_visibility(actor, action=action, resource=resource)
visible, private = await ds.check_visibility(
actor, action=action, resource=resource
)
assert visible == should_allow
assert private == expected_private

39
tests/test_permissions.py
View file

@ -59,7 +59,12 @@ async def perms_ds():
"/-/api",
"/fixtures/compound_three_primary_keys",
"/fixtures/compound_three_primary_keys/a,a,a",
pytest.param("/fixtures/two", marks=pytest.mark.xfail(reason="view-query not yet migrated to new permission system")), # Query
pytest.param(
"/fixtures/two",
marks=pytest.mark.xfail(
reason="view-query not yet migrated to new permission system"
),
), # Query
),
)
def test_view_padlock(allow, expected_anon, expected_auth, path, padlock_client):
@ -904,6 +909,7 @@ async def test_permissions_in_config(
try:
# Convert old-style resource to Resource object
from datasette.resources import DatabaseResource, TableResource
resource_obj = None
if resource:
if isinstance(resource, str):
@ -911,7 +917,9 @@ async def test_permissions_in_config(
elif isinstance(resource, tuple) and len(resource) == 2:
resource_obj = TableResource(database=resource[0], table=resource[1])
result = await perms_ds.allowed(action=action, resource=resource_obj, actor=actor)
result = await perms_ds.allowed(
action=action, resource=resource_obj, actor=actor
)
if result != expected_result:
pprint(perms_ds._permission_checks)
assert result == expected_result
@ -1123,7 +1131,16 @@ async def test_view_table_token_can_access_table(perms_ds):
({"a": ["vi"]}, "get", "/perms_ds_one/t1/1.json", None, 403),
({"a": ["vi"]}, "get", "/perms_ds_one/v1.json", None, 403),
# Restricted to just view-database
pytest.param({"a": ["vd"]}, "get", "/.json", None, 200, marks=pytest.mark.xfail(reason="Actor restrictions need additional work, refs #2534")), # Can see instance too
pytest.param(
{"a": ["vd"]},
"get",
"/.json",
None,
200,
marks=pytest.mark.xfail(
reason="Actor restrictions need additional work, refs #2534"
),
), # Can see instance too
({"a": ["vd"]}, "get", "/perms_ds_one.json", None, 200),
({"a": ["vd"]}, "get", "/perms_ds_one/t1.json", None, 403),
({"a": ["vd"]}, "get", "/perms_ds_one/t1/1.json", None, 403),
@ -1135,7 +1152,9 @@ async def test_view_table_token_can_access_table(perms_ds):
"/.json",
None,
200,
marks=pytest.mark.xfail(reason="Actor restrictions need additional work, refs #2534")
marks=pytest.mark.xfail(
reason="Actor restrictions need additional work, refs #2534"
),
), # Can see instance
pytest.param(
{"d": {"perms_ds_one": ["vt"]}},
@ -1143,7 +1162,9 @@ async def test_view_table_token_can_access_table(perms_ds):
"/perms_ds_one.json",
None,
200,
marks=pytest.mark.xfail(reason="Actor restrictions need additional work, refs #2534")
marks=pytest.mark.xfail(
reason="Actor restrictions need additional work, refs #2534"
),
), # and this database
(
{"d": {"perms_ds_one": ["vt"]}},
@ -1175,7 +1196,9 @@ async def test_view_table_token_can_access_table(perms_ds):
"/.json",
None,
200,
marks=pytest.mark.xfail(reason="Actor restrictions need additional work, refs #2534")
marks=pytest.mark.xfail(
reason="Actor restrictions need additional work, refs #2534"
),
),
pytest.param(
{"r": {"perms_ds_one": {"t1": ["vt"]}}},
@ -1183,7 +1206,9 @@ async def test_view_table_token_can_access_table(perms_ds):
"/perms_ds_one.json",
None,
200,
marks=pytest.mark.xfail(reason="Actor restrictions need additional work, refs #2534")
marks=pytest.mark.xfail(
reason="Actor restrictions need additional work, refs #2534"
),
),
(
{"r": {"perms_ds_one": {"t1": ["vt"]}}},

20
tests/test_plugins.py
View file

@ -848,7 +848,9 @@ async def test_hook_startup(ds_client):
@pytest.mark.asyncio
@pytest.mark.xfail(reason="Canned queries not yet migrated to new permission system, refs #2510")
@pytest.mark.xfail(
reason="Canned queries not yet migrated to new permission system, refs #2510"
)
async def test_hook_canned_queries(ds_client):
queries = (await ds_client.get("/fixtures.json")).json()["queries"]
queries_by_name = {q["name"]: q for q in queries}
@ -865,21 +867,27 @@ async def test_hook_canned_queries(ds_client):
@pytest.mark.asyncio
@pytest.mark.xfail(reason="Canned queries not yet migrated to new permission system, refs #2510")
@pytest.mark.xfail(
reason="Canned queries not yet migrated to new permission system, refs #2510"
)
async def test_hook_canned_queries_non_async(ds_client):
response = await ds_client.get("/fixtures/from_hook.json?_shape=array")
assert [{"1": 1, "actor_id": "null"}] == response.json()
@pytest.mark.asyncio
@pytest.mark.xfail(reason="Canned queries not yet migrated to new permission system, refs #2510")
@pytest.mark.xfail(
reason="Canned queries not yet migrated to new permission system, refs #2510"
)
async def test_hook_canned_queries_async(ds_client):
response = await ds_client.get("/fixtures/from_async_hook.json?_shape=array")
assert [{"2": 2}] == response.json()
@pytest.mark.asyncio
@pytest.mark.xfail(reason="Canned queries not yet migrated to new permission system, refs #2510")
@pytest.mark.xfail(
reason="Canned queries not yet migrated to new permission system, refs #2510"
)
async def test_hook_canned_queries_actor(ds_client):
assert (
await ds_client.get("/fixtures/from_hook.json?_bot=1&_shape=array")
@ -1541,7 +1549,9 @@ async def test_hook_top_query(ds_client):
@pytest.mark.asyncio
@pytest.mark.xfail(reason="Canned queries not yet migrated to new permission system, refs #2510")
@pytest.mark.xfail(
reason="Canned queries not yet migrated to new permission system, refs #2510"
)
async def test_hook_top_canned_query(ds_client):
try:
pm.register(SlotPlugin(), name="SlotPlugin")