SQL query took too long. The time limit is controlled by the\n" 'sql_time_limit_ms\n' "configuration option.
\n" '\n' "" ), "status": 400, "title": "SQL Interrupted", } @pytest.mark.asyncio async def test_custom_sql_time_limit(ds_client): response = await ds_client.get( "/fixtures/-/query.json?sql=select+sleep(0.01)", ) assert response.status_code == 200 response = await ds_client.get( "/fixtures/-/query.json?sql=select+sleep(0.01)&_timelimit=5", ) assert response.status_code == 400 assert response.json()["title"] == "SQL Interrupted" @pytest.mark.asyncio async def test_invalid_custom_sql(ds_client): response = await ds_client.get( "/fixtures/-/query.json?sql=.schema", ) assert response.status_code == 400 assert response.json()["ok"] is False assert "Statement must be a SELECT" == response.json()["error"] @pytest.mark.asyncio async def test_row(ds_client): response = await ds_client.get("/fixtures/simple_primary_key/1.json?_shape=objects") assert response.status_code == 200 assert response.json()["ok"] is True assert response.json()["rows"] == [{"id": 1, "content": "hello"}] @pytest.mark.asyncio async def test_row_strange_table_name(ds_client): response = await ds_client.get( "/fixtures/table~2Fwith~2Fslashes~2Ecsv/3.json?_shape=objects" ) assert response.status_code == 200 assert response.json()["rows"] == [{"pk": "3", "content": "hey"}] @pytest.mark.asyncio async def test_row_foreign_key_tables(ds_client): response = await ds_client.get( "/fixtures/simple_primary_key/1.json?_extras=foreign_key_tables" ) assert response.status_code == 200 # Foreign keys are sorted by (other_table, column, other_column) assert response.json()["foreign_key_tables"] == [ { "other_table": "complex_foreign_keys", "column": "id", "other_column": "f1", "count": 1, "link": "/fixtures/complex_foreign_keys?f1=1", }, { "other_table": "complex_foreign_keys", "column": "id", "other_column": "f2", "count": 0, "link": "/fixtures/complex_foreign_keys?f2=1", }, { "other_table": "complex_foreign_keys", "column": "id", "other_column": "f3", "count": 1, "link": "/fixtures/complex_foreign_keys?f3=1", }, { "other_table": "foreign_key_references", "column": "id", "other_column": "foreign_key_with_blank_label", "count": 0, "link": "/fixtures/foreign_key_references?foreign_key_with_blank_label=1", }, { "other_table": "foreign_key_references", "column": "id", "other_column": "foreign_key_with_label", "count": 1, "link": "/fixtures/foreign_key_references?foreign_key_with_label=1", }, ] @pytest.mark.asyncio async def test_row_extra_render_cell(): """Test that _extra=render_cell returns rendered HTML from render_cell plugin hook on row pages""" from datasette import hookimpl from datasette.app import Datasette class TestRenderCellPlugin: __name__ = "TestRenderCellPlugin" @hookimpl def render_cell(self, value, column, table, database): # Only modify cells in our test table if table == "test_render" and column == "name": return f"{value}" return None ds = Datasette(memory=True) await ds.invoke_startup() db = ds.add_memory_database("test_row_render") await db.execute_write( "create table test_render (id integer primary key, name text)" ) await db.execute_write("insert into test_render values (1, 'Alice')") # Register our test plugin ds.pm.register(TestRenderCellPlugin(), name="TestRenderCellPlugin") try: # Request row with _extra=render_cell response = await ds.client.get( "/test_row_render/test_render/1.json?_extra=render_cell" ) assert response.status_code == 200 data = response.json() # Verify the response structure assert "render_cell" in data assert "rows" in data # render_cell should be a list with one row (since this is a row page) # Only columns modified by plugins are included (sparse output) render_cell = data["render_cell"] assert len(render_cell) == 1 # The row: id=1, name='Alice' # The 'name' column should be rendered by our plugin as Alice assert render_cell[0]["name"] == "Alice" # The 'id' column is not included since no plugin modified it assert "id" not in render_cell[0] # The regular rows should still contain raw values assert data["rows"] == [{"id": 1, "name": "Alice"}] finally: ds.pm.unregister(name="TestRenderCellPlugin") def test_databases_json(app_client_two_attached_databases_one_immutable): response = app_client_two_attached_databases_one_immutable.get("/-/databases.json") databases = response.json assert 2 == len(databases) extra_database, fixtures_database = databases assert "extra database" == extra_database["name"] assert extra_database["hash"] is None assert extra_database["is_mutable"] is True assert extra_database["is_memory"] is False assert "fixtures" == fixtures_database["name"] assert fixtures_database["hash"] is not None assert fixtures_database["is_mutable"] is False assert fixtures_database["is_memory"] is False @pytest.mark.asyncio async def test_threads_json(ds_client): response = await ds_client.get("/-/threads.json") expected_keys = {"threads", "num_threads"} if sys.version_info >= (3, 7, 0): expected_keys.update({"tasks", "num_tasks"}) data = response.json() assert set(data.keys()) == expected_keys # Should be at least one _execute_writes thread for __INTERNAL__ thread_names = [thread["name"] for thread in data["threads"]] assert "_execute_writes for database __INTERNAL__" in thread_names @pytest.mark.asyncio async def test_plugins_json(ds_client): response = await ds_client.get("/-/plugins.json") # Filter out TrackEventPlugin actual_plugins = sorted( [p for p in response.json() if p["name"] != "TrackEventPlugin"], key=lambda p: p["name"], ) assert EXPECTED_PLUGINS == actual_plugins # Try with ?all=1 response = await ds_client.get("/-/plugins.json?all=1") names = {p["name"] for p in response.json()} assert names.issuperset(p["name"] for p in EXPECTED_PLUGINS) assert names.issuperset(DEFAULT_PLUGINS) @pytest.mark.asyncio async def test_versions_json(ds_client): response = await ds_client.get("/-/versions.json") data = response.json() assert "python" in data assert "3.0" == data.get("asgi") assert "version" in data["python"] assert "full" in data["python"] assert "datasette" in data assert "version" in data["datasette"] assert data["datasette"]["version"] == __version__ assert "sqlite" in data assert "version" in data["sqlite"] assert "fts_versions" in data["sqlite"] assert "compile_options" in data["sqlite"] # By default, the json1 extension is enabled in the SQLite # provided by the `ubuntu-latest` github actions runner, and # all versions of SQLite from 3.38.0 onwards assert data["sqlite"]["extensions"]["json1"] @pytest.mark.asyncio async def test_actions_json(ds_client): original_root_enabled = ds_client.ds.root_enabled try: ds_client.ds.root_enabled = True cookies = {"ds_actor": ds_client.actor_cookie({"id": "root"})} response = await ds_client.get("/-/actions.json", cookies=cookies) data = response.json() finally: ds_client.ds.root_enabled = original_root_enabled assert isinstance(data, list) assert len(data) > 0 # Check structure of first action action = data[0] for key in ( "name", "abbr", "description", "takes_parent", "takes_child", "resource_class", "also_requires", ): assert key in action # Check that some expected actions exist action_names = {a["name"] for a in data} for expected_action in ( "view-instance", "view-database", "view-table", "execute-sql", ): assert expected_action in action_names @pytest.mark.asyncio async def test_settings_json(ds_client): response = await ds_client.get("/-/settings.json") assert response.json() == { "default_page_size": 50, "default_facet_size": 30, "default_allow_sql": True, "facet_suggest_time_limit_ms": 200, "facet_time_limit_ms": 200, "max_returned_rows": 100, "max_insert_rows": 100, "sql_time_limit_ms": 200, "allow_download": True, "allow_signed_tokens": True, "max_signed_tokens_ttl": 0, "allow_facet": True, "suggest_facets": True, "default_cache_ttl": 5, "num_sql_threads": 1, "cache_size_kb": 0, "allow_csv_stream": True, "max_csv_mb": 100, "truncate_cells_html": 2048, "force_https_urls": False, "template_debug": False, "trace_debug": False, "base_url": "/", } test_json_columns_default_expected = [ {"intval": 1, "strval": "s", "floatval": 0.5, "jsonval": '{"foo": "bar"}'} ] @pytest.mark.asyncio @pytest.mark.parametrize( "extra_args,expected", [ ("", test_json_columns_default_expected), ("&_json=intval", test_json_columns_default_expected), ("&_json=strval", test_json_columns_default_expected), ("&_json=floatval", test_json_columns_default_expected), ( "&_json=jsonval", [{"intval": 1, "strval": "s", "floatval": 0.5, "jsonval": {"foo": "bar"}}], ), ], ) async def test_json_columns(ds_client, extra_args, expected): sql = """ select 1 as intval, "s" as strval, 0.5 as floatval, '{"foo": "bar"}' as jsonval """ path = "/fixtures/-/query.json?" + urllib.parse.urlencode( {"sql": sql, "_shape": "array"} ) path += extra_args response = await ds_client.get( path, ) assert response.json() == expected def test_config_cache_size(app_client_larger_cache_size): response = app_client_larger_cache_size.get("/fixtures/pragma_cache_size.json") assert response.json["rows"] == [{"cache_size": -2500}] def test_config_force_https_urls(): with make_app_client(settings={"force_https_urls": True}) as client: response = client.get( "/fixtures/facetable.json?_size=3&_facet=state&_extra=next_url,suggested_facets" ) assert response.json["next_url"].startswith("https://") assert response.json["facet_results"]["results"]["state"]["results"][0][ "toggle_url" ].startswith("https://") assert response.json["suggested_facets"][0]["toggle_url"].startswith("https://") # Also confirm that request.url and request.scheme are set correctly response = client.get("/") assert client.ds._last_request.url.startswith("https://") assert client.ds._last_request.scheme == "https" @pytest.mark.parametrize( "path,status_code", [ ("/fixtures.db", 200), ("/fixtures.json", 200), ("/fixtures/no_primary_key.json", 200), # A 400 invalid SQL query should still have the header: ("/fixtures/-/query.json?sql=select+blah", 400), # Write APIs ("/fixtures/-/create", 405), ("/fixtures/facetable/-/insert", 405), ("/fixtures/facetable/-/drop", 405), ], ) def test_cors( app_client_with_cors, app_client_two_attached_databases_one_immutable, path, status_code, ): response = app_client_with_cors.get( path, ) assert response.status == status_code assert response.headers["Access-Control-Allow-Origin"] == "*" assert ( response.headers["Access-Control-Allow-Headers"] == "Authorization, Content-Type" ) assert response.headers["Access-Control-Expose-Headers"] == "Link" assert ( response.headers["Access-Control-Allow-Methods"] == "GET, POST, HEAD, OPTIONS" ) assert response.headers["Access-Control-Max-Age"] == "3600" # Same request to app_client_two_attached_databases_one_immutable # should not have those headers - I'm using that fixture because # regular app_client doesn't have immutable fixtures.db which means # the test for /fixtures.db returns a 403 error response = app_client_two_attached_databases_one_immutable.get( path, ) assert response.status == status_code assert "Access-Control-Allow-Origin" not in response.headers assert "Access-Control-Allow-Headers" not in response.headers assert "Access-Control-Expose-Headers" not in response.headers assert "Access-Control-Allow-Methods" not in response.headers assert "Access-Control-Max-Age" not in response.headers @pytest.mark.parametrize( "path", ( "/", ".json", "/searchable", "/searchable.json", "/searchable_view", "/searchable_view.json", ), ) def test_database_with_space_in_name(app_client_two_attached_databases, path): response = app_client_two_attached_databases.get( "/extra~20database" + path, follow_redirects=True ) assert response.status == 200 def test_common_prefix_database_names(app_client_conflicting_database_names): # https://github.com/simonw/datasette/issues/597 assert ["foo-bar", "foo", "fixtures"] == [ d["name"] for d in app_client_conflicting_database_names.get("/-/databases.json").json ] for db_name, path in (("foo", "/foo.json"), ("foo-bar", "/foo-bar.json")): data = app_client_conflicting_database_names.get(path).json assert db_name == data["database"] def test_inspect_file_used_for_count(app_client_immutable_and_inspect_file): response = app_client_immutable_and_inspect_file.get( "/fixtures/sortable.json?_extra=count" ) assert response.json["count"] == 100 @pytest.mark.asyncio async def test_http_options_request(ds_client): response = await ds_client.options("/fixtures") assert response.status_code == 200 assert response.text == "ok" @pytest.mark.asyncio async def test_db_path(app_client): # Needs app_client because needs file based database db = app_client.ds.get_database() path = pathlib.Path(db.path) assert path.exists() datasette = Datasette([path]) # Previously this broke if path was a pathlib.Path: await datasette.refresh_schemas() @pytest.mark.asyncio async def test_hidden_sqlite_stat1_table(): ds = Datasette() db = ds.add_memory_database("db") await db.execute_write("create table normal (id integer primary key, name text)") await db.execute_write("create index idx on normal (name)") await db.execute_write("analyze") data = (await ds.client.get("/db.json?_show_hidden=1")).json() tables = [(t["name"], t["hidden"]) for t in data["tables"]] assert tables in ( [("normal", False), ("sqlite_stat1", True)], [("normal", False), ("sqlite_stat1", True), ("sqlite_stat4", True)], ) @pytest.mark.asyncio async def test_hide_tables_starting_with_underscore(): ds = Datasette() db = ds.add_memory_database("test_hide_tables_starting_with_underscore") await db.execute_write("create table normal (id integer primary key, name text)") await db.execute_write("create table _hidden (id integer primary key, name text)") data = ( await ds.client.get( "/test_hide_tables_starting_with_underscore.json?_show_hidden=1" ) ).json() tables = [(t["name"], t["hidden"]) for t in data["tables"]] assert tables == [("normal", False), ("_hidden", True)] @pytest.mark.asyncio @pytest.mark.parametrize("db_name", ("foo", r"fo%o", "f~/c.d")) async def test_tilde_encoded_database_names(db_name): ds = Datasette() ds.add_memory_database(db_name) response = await ds.client.get("/.json") assert db_name in response.json()["databases"].keys() path = response.json()["databases"][db_name]["path"] # And the JSON for that database response2 = await ds.client.get(path + ".json") assert response2.status_code == 200 @pytest.mark.asyncio @pytest.mark.parametrize( "config,expected", ( ({}, {}), ({"plugins": {"datasette-foo": "bar"}}, {"plugins": {"datasette-foo": "bar"}}), # Test redaction ( { "plugins": { "datasette-auth": {"secret_key": "key"}, "datasette-foo": "bar", "datasette-auth2": {"password": "password"}, "datasette-sentry": { "dsn": "sentry:///foo", }, } }, { "plugins": { "datasette-auth": {"secret_key": "***"}, "datasette-foo": "bar", "datasette-auth2": {"password": "***"}, "datasette-sentry": {"dsn": "***"}, } }, ), ), ) async def test_config_json(config, expected): "/-/config.json should return redacted configuration" ds = Datasette(config=config) response = await ds.client.get("/-/config.json") assert response.json() == expected @pytest.mark.asyncio @pytest.mark.skip(reason="rm?") @pytest.mark.parametrize( "metadata,expected_config,expected_metadata", ( ({}, {}, {}), ( # Metadata input { "title": "Datasette Fixtures", "databases": { "fixtures": { "tables": { "sortable": { "sortable_columns": [ "sortable", "sortable_with_nulls", "sortable_with_nulls_2", "text", ], }, "no_primary_key": {"sortable_columns": [], "hidden": True}, "primary_key_multiple_columns_explicit_label": { "label_column": "content2" }, "simple_view": {"sortable_columns": ["content"]}, "searchable_view_configured_by_metadata": { "fts_table": "searchable_fts", "fts_pk": "pk", }, "roadside_attractions": { "columns": { "name": "The name of the attraction", "address": "The street address for the attraction", } }, "attraction_characteristic": {"sort_desc": "pk"}, "facet_cities": {"sort": "name"}, "paginated_view": {"size": 25}, }, } }, }, # Should produce a config with just the table configuration keys { "databases": { "fixtures": { "tables": { "sortable": { "sortable_columns": [ "sortable", "sortable_with_nulls", "sortable_with_nulls_2", "text", ] }, # These one get redacted: "no_primary_key": "***", "primary_key_multiple_columns_explicit_label": "***", "simple_view": {"sortable_columns": ["content"]}, "searchable_view_configured_by_metadata": { "fts_table": "searchable_fts", "fts_pk": "pk", }, "attraction_characteristic": {"sort_desc": "pk"}, "facet_cities": {"sort": "name"}, "paginated_view": {"size": 25}, } } } }, # And metadata with everything else { "title": "Datasette Fixtures", "databases": { "fixtures": { "tables": { "roadside_attractions": { "columns": { "name": "The name of the attraction", "address": "The street address for the attraction", } }, } } }, }, ), ), ) async def test_upgrade_metadata(metadata, expected_config, expected_metadata): ds = Datasette(metadata=metadata) response = await ds.client.get("/-/config.json") assert response.json() == expected_config response2 = await ds.client.get("/-/metadata.json") assert response2.json() == expected_metadata class Either: def __init__(self, a, b): self.a = a self.b = b def __eq__(self, other): return other == self.a or other == self.b