diff --git a/tests/test_auth.py b/tests/test_auth.py index 571389a3..e351b8fd 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -357,10 +357,10 @@ async def test_root_with_root_enabled_gets_all_permissions(ds_client): # Test instance-level permissions (no resource) assert ( - await ds_client.ds.permission_allowed(root_actor, "permissions-debug", None) + await ds_client.ds.allowed(action="permissions-debug", actor=root_actor) is True ) - assert await ds_client.ds.permission_allowed(root_actor, "debug-menu", None) 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 ( @@ -478,8 +478,8 @@ 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.permission_allowed( - root_actor, "permissions-debug", None + result = await ds_client.ds.allowed( + action="permissions-debug", actor=root_actor ) assert ( result is not True diff --git a/tests/test_permissions.py b/tests/test_permissions.py index 25dce2c9..dbdad368 100644 --- a/tests/test_permissions.py +++ b/tests/test_permissions.py @@ -900,7 +900,16 @@ async def test_permissions_in_config( updated_config.update(config) perms_ds.config = updated_config try: - result = await perms_ds.permission_allowed(actor, action, resource) + # Convert old-style resource to Resource object + from datasette.resources import DatabaseResource, TableResource + resource_obj = None + if resource: + if isinstance(resource, str): + resource_obj = DatabaseResource(database=resource) + 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) if result != expected_result: pprint(perms_ds._permission_checks) assert result == expected_result diff --git a/tests/test_plugins.py b/tests/test_plugins.py index 9c74b432..546b0e9e 100644 --- a/tests/test_plugins.py +++ b/tests/test_plugins.py @@ -696,7 +696,7 @@ async def test_hook_permission_allowed(action, expected): try: ds = Datasette(plugins_dir=PLUGINS_DIR) await ds.invoke_startup() - actual = await ds.permission_allowed({"id": "actor"}, action) + actual = await ds.allowed(action=action, actor={"id": "actor"}) assert expected == actual finally: pm.unregister(name="undo_register_extras") diff --git a/tests/test_special.py b/tests/test_special.py new file mode 100644 index 00000000..5a4f4ca3 --- /dev/null +++ b/tests/test_special.py @@ -0,0 +1,196 @@ +""" +Tests for special endpoints in datasette/views/special.py +""" + +import pytest +import pytest_asyncio +from datasette.app import Datasette + + +@pytest_asyncio.fixture +async def ds_with_tables(): + """Create a Datasette instance with some tables for searching.""" + ds = Datasette( + config={ + "databases": { + "content": { + "allow": {"id": "*"}, # Allow all authenticated users + "tables": { + "articles": { + "allow": {"id": "editor"}, # Only editor can view + }, + "comments": { + "allow": True, # Everyone can view + }, + }, + }, + "private": { + "allow": False, # Deny everyone + }, + } + } + ) + await ds.invoke_startup() + + # Add content database with some tables + content_db = ds.add_memory_database("content") + await content_db.execute_write( + "CREATE TABLE IF NOT EXISTS articles (id INTEGER PRIMARY KEY, title TEXT)" + ) + await content_db.execute_write( + "CREATE TABLE IF NOT EXISTS comments (id INTEGER PRIMARY KEY, body TEXT)" + ) + await content_db.execute_write( + "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)" + ) + + # Add private database with a table + private_db = ds.add_memory_database("private") + await private_db.execute_write( + "CREATE TABLE IF NOT EXISTS secrets (id INTEGER PRIMARY KEY, data TEXT)" + ) + + # Add another public database + public_db = ds.add_memory_database("public") + await public_db.execute_write( + "CREATE TABLE IF NOT EXISTS articles (id INTEGER PRIMARY KEY, content TEXT)" + ) + + return ds + + +# /-/tables.json tests +@pytest.mark.asyncio +async def test_tables_empty_query(ds_with_tables): + """Test that empty query returns empty matches.""" + response = await ds_with_tables.client.get("/-/tables.json") + assert response.status_code == 200 + data = response.json() + assert data == {"matches": []} + + +@pytest.mark.asyncio +async def test_tables_no_query_param(ds_with_tables): + """Test that missing q parameter returns empty matches.""" + response = await ds_with_tables.client.get("/-/tables.json?q=") + assert response.status_code == 200 + data = response.json() + assert data == {"matches": []} + + +@pytest.mark.asyncio +async def test_tables_basic_search(ds_with_tables): + """Test basic table search functionality.""" + # Search for "articles" - should find it in both content and public databases + # but only return public.articles for anonymous user (content.articles requires auth) + response = await ds_with_tables.client.get("/-/tables.json?q=articles") + assert response.status_code == 200 + data = response.json() + + # Should only see public.articles (content.articles restricted to authenticated users) + assert "matches" in data + assert len(data["matches"]) == 1 + + match = data["matches"][0] + assert "url" in match + assert "name" in match + assert match["name"] == "public: articles" + assert "/public/articles" in match["url"] + + +@pytest.mark.asyncio +async def test_tables_search_with_auth(ds_with_tables): + """Test that authenticated users see more tables.""" + # Editor user should see content.articles + response = await ds_with_tables.client.get( + "/-/tables.json?q=articles", + cookies={"ds_actor": ds_with_tables.client.actor_cookie({"id": "editor"})}, + ) + assert response.status_code == 200 + data = response.json() + + # Should see both content.articles and public.articles + assert len(data["matches"]) == 2 + + names = {match["name"] for match in data["matches"]} + assert names == {"content: articles", "public: articles"} + + +@pytest.mark.asyncio +async def test_tables_search_partial_match(ds_with_tables): + """Test that search matches partial table names.""" + # Search for "com" should match "comments" + response = await ds_with_tables.client.get( + "/-/tables.json?q=com", + cookies={"ds_actor": ds_with_tables.client.actor_cookie({"id": "user"})}, + ) + assert response.status_code == 200 + data = response.json() + + assert len(data["matches"]) == 1 + assert data["matches"][0]["name"] == "content: comments" + + +@pytest.mark.asyncio +async def test_tables_search_respects_database_permissions(ds_with_tables): + """Test that tables from denied databases are not shown.""" + # Search for "secrets" which is in the private database + # Even authenticated users shouldn't see it because database is denied + response = await ds_with_tables.client.get( + "/-/tables.json?q=secrets", + cookies={"ds_actor": ds_with_tables.client.actor_cookie({"id": "user"})}, + ) + assert response.status_code == 200 + data = response.json() + + # Should not see secrets table from private database + assert len(data["matches"]) == 0 + + +@pytest.mark.asyncio +async def test_tables_search_respects_table_permissions(ds_with_tables): + """Test that tables with specific permissions are filtered correctly.""" + # Regular authenticated user searching for "users" + response = await ds_with_tables.client.get( + "/-/tables.json?q=users", + cookies={"ds_actor": ds_with_tables.client.actor_cookie({"id": "regular"})}, + ) + assert response.status_code == 200 + data = response.json() + + # Should see content.users (authenticated users can view content database) + assert len(data["matches"]) == 1 + assert data["matches"][0]["name"] == "content: users" + + +@pytest.mark.asyncio +async def test_tables_search_no_matches(ds_with_tables): + """Test search with no matching tables.""" + response = await ds_with_tables.client.get("/-/tables.json?q=nonexistent") + assert response.status_code == 200 + data = response.json() + + assert data == {"matches": []} + + +@pytest.mark.asyncio +async def test_tables_search_response_structure(ds_with_tables): + """Test that response has correct structure.""" + response = await ds_with_tables.client.get( + "/-/tables.json?q=users", + cookies={"ds_actor": ds_with_tables.client.actor_cookie({"id": "user"})}, + ) + assert response.status_code == 200 + data = response.json() + + assert "matches" in data + assert isinstance(data["matches"], list) + + if data["matches"]: + match = data["matches"][0] + assert "url" in match + assert "name" in match + assert isinstance(match["url"], str) + assert isinstance(match["name"], str) + # Name should be in format "database: table" + assert ": " in match["name"] diff --git a/tests/vec.db b/tests/vec.db new file mode 100644 index 00000000..195ef213 Binary files /dev/null and b/tests/vec.db differ