home

Menu
  • ripgrep search

ripgrep

Options:

For example *.py or **/templates/**/*.html or datasette/** or !setup.py

apple-notes-to-sqlite/README.md

128 To run the tests:
129 
130     pytest

apple-notes-to-sqlite/setup.py

33      """,
34      install_requires=["click", "sqlite-utils"],
35      extras_require={"test": ["pytest", "pytest-subprocess", "cogapp"]},
36      python_requires=">=3.7",
37  )

asgi-log-to-sqlite/setup.py

24      py_modules=["asgi_log_to_sqlite"],
25      install_requires=["sqlite_utils~=2.3.1"],
26      extras_require={"test": ["pytest", "pytest-asyncio", "asgiref==3.1.2"]},
27  )

asgi-log-to-sqlite/test_asgi_log_to_sqlite.py

2   from asgi_log_to_sqlite import AsgiLogToSqlite
3   import sqlite_utils
4   import pytest
5   
6   
17  
18  
19  @pytest.mark.asyncio
20  async def test_log_to_sqlite(tmpdir):
21      logfile = str(tmpdir / "log.db")
44  
45  
46  @pytest.mark.asyncio
47  async def test_log_to_sqlite_with_more_fields(tmpdir):
48      logfile = str(tmpdir / "log2.db")

csvs-to-sqlite/CHANGELOG.md

133   Which is itself an updated version of the pattern described in http://dan-blanchard.roughdraft.io/7045057-quicker-travis-builds-that-rely-on-numpy-and-scipy-using-miniconda
134 
135   I had to switch to running `pytest` directly, because `python setup.py test` was still trying to install a pandas package that involved compiling everything from scratch (which is why Travis CI builds were taking around 15 minutes).
136 - Don't include an `index` column - rely on SQLite rowid instead.
137 

csvs-to-sqlite/setup.cfg

1   [aliases]
2   test=pytest
3   
4   [bdist_wheel]

csvs-to-sqlite/setup.py

30          "six",
31      ],
32      extras_require={"test": ["pytest", "cogapp"]},
33      entry_points="""
34          [console_scripts]

csvs-to-sqlite/tests/test_utils.py

1   from csvs_to_sqlite import utils
2   import pytest
3   import sqlite3
4   import pandas as pd
12  
13  
14  @pytest.mark.parametrize("table,expected", [("foo", True), ("bar", False)])
15  def test_table_exists(table, expected):
16      conn = sqlite3.connect(":memory:")

datasette/datasette/cli.py

628     asyncio.get_event_loop().run_until_complete(ds.invoke_startup())
629 
630     # Run async soundness checks - but only if we're not under pytest
631     asyncio.get_event_loop().run_until_complete(check_databases(ds))
632 

datasette/docs/changelog.rst

408 - Columns with the name "Link" are no longer incorrectly displayed in bold. (:issue:`1308`)
409 - Fixed error caused by tables with a single quote in their names. (:issue:`1257`)
410 - Updated dependencies: ``pytest-asyncio``, ``Black``, ``jinja2``, ``aiofiles``, ``click``, and ``itsdangerous``.
411 - The official Datasette Docker image now supports ``apt-get install``. (:issue:`1320`)
412 - The Heroku runtime used by ``datasette publish heroku`` is now ``python-3.8.10``.
511 - Better error message for disallowed ``PRAGMA`` clauses in SQL queries. (:issue:`1185`)
512 - ``datasette publish heroku`` now deploys using ``python-3.8.7``.
513 - New plugin testing documentation on :ref:`testing_plugins_pytest_httpx`. (:issue:`1198`)
514 - All ``?_*`` query string parameters passed to the table page are now persisted in hidden form fields, so parameters such as ``?_size=10`` will be correctly passed to the next page when query filters are changed. (:issue:`1194`)
515 - Fixed a bug loading a database file called ``test-database (1).sqlite``. (:issue:`1181`)
833 - :ref:`writing_plugins` describes how to author plugins, from  one-off single file plugins to packaged plugins that can be published to PyPI. It also describes how to start a plugin using the new `datasette-plugin <https://github.com/simonw/datasette-plugin>`__ cookiecutter template.
834 - :ref:`plugin_hooks` is a full list of detailed documentation for every Datasette plugin hook.
835 - :ref:`testing_plugins` describes how to write tests for Datasette plugins, using `pytest <https://docs.pytest.org/>`__ and `HTTPX <https://www.python-httpx.org/>`__.
836 
837 New plugin hooks
1443- New project guideline: master should stay shippable at all times! (`31f36e1 <https://github.com/simonw/datasette/commit/31f36e1b97ccc3f4387c80698d018a69798b6228>`__)
1444- Fixed a bug where ``sqlite_timelimit()`` occasionally failed to clean up after itself (`bac4e01 <https://github.com/simonw/datasette/commit/bac4e01f40ae7bd19d1eab1fb9349452c18de8f5>`__)
1445- We no longer load additional plugins when executing pytest (:issue:`438`)
1446- Homepage now links to database views if there are less than five tables in a database (:issue:`373`)
1447- The ``--cors`` option is now respected by error pages (:issue:`453`)
1496- Added documentation on :ref:`how to build the documentation <contributing_documentation>`
1497- Added documentation covering :ref:`our release process <contributing_release>`
1498- Upgraded to pytest 4.0.2
1499
1500.. _v0_25_1:

datasette/docs/contributing.rst

49  -----------------
50  
51  Once you have done this, you can run the Datasette unit tests from inside your ``datasette/`` directory using `pytest <https://docs.pytest.org/>`__ like so::
52  
53      pytest
54  
55  You can run the tests faster using multiple CPU cores with `pytest-xdist <https://pypi.org/project/pytest-xdist/>`__ like this::
56  
57      pytest -n auto -m "not serial"
58  
59  ``-n auto`` detects the number of available cores automatically. The ``-m "not serial"`` skips tests that don't work well in a parallel test environment. You can run those tests separately like so::
60  
61      pytest -m "serial"
62  
63  .. _contributing_using_fixtures:

datasette/docs/plugin_hooks.rst

911    .. code-block:: python
912 
913         @pytest.mark.asyncio
914         async def test_my_plugin():
915             ds = Datasette()

datasette/docs/testing_plugins.rst

4   ===============
5   
6   We recommend using `pytest <https://docs.pytest.org/>`__ to write automated tests for your plugins.
7   
8   If you use the template described in :ref:`writing_plugins_cookiecutter` your plugin will start with a single test in your ``tests/`` directory that looks like this:
11  
12      from datasette.app import Datasette
13      import pytest
14  
15  
16      @pytest.mark.asyncio
17      async def test_plugin_is_installed():
18          datasette = Datasette(memory=True)
28  This test uses the :ref:`internals_datasette_client` object to exercise a test instance of Datasette. ``datasette.client`` is a wrapper around the `HTTPX <https://www.python-httpx.org/>`__ Python library which can imitate HTTP requests using ASGI. This is the recommended way to write tests against a Datasette instance.
29  
30  This test also uses the `pytest-asyncio <https://pypi.org/project/pytest-asyncio/>`__ package to add support for ``async def`` test functions running under pytest.
31  
32  You can install these packages like so::
33  
34      pip install pytest pytest-asyncio
35  
36  If you are building an installable package you can add them as test dependencies to your ``setup.py`` module like this:
41          name="datasette-my-plugin",
42          # ...
43          extras_require={"test": ["pytest", "pytest-asyncio"]},
44          tests_require=["datasette-my-plugin[test]"],
45      )
49      pip install -e '.[test]'
50  
51  Then run the tests using pytest like so::
52  
53      pytest
54  
55  .. _testing_plugins_datasette_test_instance:
63  
64      from datasette.app import Datasette
65      import pytest
66  
67  
68      @pytest.mark.asyncio
69      async def test_plugin_is_installed():
70          datasette = Datasette(memory=True)
100         response = await ds.client.get("/")
101 
102 If you use this pattern you will need to run ``pytest`` with the ``-s`` option to avoid capturing stdin/stdout in order to interact with the debugger prompt.
103 
104 .. _testing_plugins_fixtures:
105 
106 Using pytest fixtures
107 ---------------------
108 
109 `Pytest fixtures <https://docs.pytest.org/en/stable/fixture.html>`__ can be used to create initial testable objects which can then be used by multiple tests.
110 
111 A common pattern for Datasette plugins is to create a fixture which sets up a temporary test database and wraps it in a Datasette instance.
116 
117     from datasette.app import Datasette
118     import pytest
119     import sqlite_utils
120 
121 
122     @pytest.fixture(scope="session")
123     def datasette(tmp_path_factory):
124         db_directory = tmp_path_factory.mktemp("dbs")
147 
148 
149     @pytest.mark.asyncio
150     async def test_example_table_json(datasette):
151         response = await datasette.client.get(
159 
160 
161     @pytest.mark.asyncio
162     async def test_example_table_html(datasette):
163         response = await datasette.client.get("/test/dogs")
164         assert ">Some dogs</h1>" in response.text
165 
166 Here the ``datasette()`` function defines the fixture, which is than automatically passed to the two test functions based on pytest automatically matching their ``datasette`` function parameters.
167 
168 The ``@pytest.fixture(scope="session")`` line here ensures the fixture is reused for the full ``pytest`` execution session. This means that the temporary database file will be created once and reused for each test.
169 
170 If you want to create that test database repeatedly for every individual test function, write the fixture function like this instead. You may want to do this if your plugin modifies the database contents in some way:
172 .. code-block:: python
173 
174     @pytest.fixture
175     def datasette(tmp_path_factory):
176         # This fixture will be executed repeatedly for every test
177         ...
178 
179 .. _testing_plugins_pytest_httpx:
180 
181 Testing outbound HTTP calls with pytest-httpx
182 ---------------------------------------------
183 
184 If your plugin makes outbound HTTP calls - for example datasette-auth-github or datasette-import-table - you may need to mock those HTTP requests in your tests.
185 
186 The `pytest-httpx <https://pypi.org/project/pytest-httpx/>`__ package is a useful library for mocking calls. It can be tricky to use with Datasette though since it mocks all HTTPX requests, and Datasette's own testing mechanism uses HTTPX internally.
187 
188 To avoid breaking your tests, you can return ``["localhost"]`` from the ``non_mocked_hosts()`` fixture.
224 
225     from datasette.app import Datasette
226     import pytest
227 
228 
229     @pytest.fixture
230     def non_mocked_hosts():
231         # This ensures httpx-mock will not affect Datasette's own
263     from datasette.app import Datasette
264     from datasette.plugins import pm
265     import pytest
266 
267 
268     @pytest.mark.asyncio
269     async def test_using_test_plugin():
270         class TestPlugin:

datasette/pytest.ini

1   [pytest]
2   filterwarnings=
3       # https://github.com/pallets/jinja/issues/927
8       ignore:.*current_task.*:PendingDeprecationWarning
9   markers =
10      serial: tests to avoid using with pytest-xdist
11  asyncio_mode = strict

datasette/setup.cfg

1   [aliases]
2   test=pytest
3   
4   [flake8]

datasette/setup.py

65          datasette=datasette.cli:cli
66      """,
67      setup_requires=["pytest-runner"],
68      extras_require={
69          "docs": [
76          ],
77          "test": [
78              "pytest>=5.2.2",
79              "pytest-xdist>=2.2.1",
80              "pytest-asyncio>=0.17",
81              "beautifulsoup4>=4.8.1",
82              "black==23.3.0",
83              "blacken-docs==1.13.0",
84              "pytest-timeout>=1.4.2",
85              "trustme>=0.7",
86              "cogapp>=3.3.0",

datasette/tests/conftest.py

3   import os
4   import pathlib
5   import pytest
6   import pytest_asyncio
7   import re
8   import subprocess
39  
40  
41  @pytest_asyncio.fixture
42  async def ds_client():
43      from datasette.app import Datasette
78  
79  
80  def pytest_report_header(config):
81      return "SQLite: {}".format(
82          sqlite3.connect(":memory:").execute("select sqlite_version()").fetchone()[0]
84  
85  
86  def pytest_configure(config):
87      import sys
88  
90  
91  
92  def pytest_unconfigure(config):
93      import sys
94  
96  
97  
98  def pytest_collection_modifyitems(items):
99      # Ensure test_cli.py and test_black.py and test_inspect.py run first before any asyncio code kicks in
100     move_to_front(items, "test_cli")
115 
116 
117 @pytest.fixture
118 def restore_working_directory(tmpdir, request):
119     try:
131 
132 
133 @pytest.fixture(scope="session", autouse=True)
134 def check_permission_actions_are_documented():
135     from datasette.plugins import pm
164 
165 
166 @pytest.fixture(scope="session")
167 def ds_localhost_http_server():
168     ds_proc = subprocess.Popen(
177     assert not ds_proc.poll(), ds_proc.stdout.read().decode("utf-8")
178     yield ds_proc
179     # Shut it down at the end of the pytest session
180     ds_proc.terminate()
181 
182 
183 @pytest.fixture(scope="session")
184 def ds_unix_domain_socket_server(tmp_path_factory):
185     # This used to use tmp_path_factory.mktemp("uds") but that turned out to
201     assert not ds_proc.poll(), ds_proc.stdout.read().decode("utf-8")
202     yield ds_proc, uds
203     # Shut it down at the end of the pytest session
204     ds_proc.terminate()

datasette/tests/fixtures.py

8   import os
9   import pathlib
10  import pytest
11  import random
12  import string
174 
175 
176 @pytest.fixture(scope="session")
177 def app_client():
178     with make_app_client() as client:
180 
181 
182 @pytest.fixture(scope="session")
183 def app_client_no_files():
184     ds = Datasette([])
186 
187 
188 @pytest.fixture(scope="session")
189 def app_client_base_url_prefix():
190     with make_app_client(settings={"base_url": "/prefix/"}) as client:
192 
193 
194 @pytest.fixture(scope="session")
195 def app_client_two_attached_databases():
196     with make_app_client(
200 
201 
202 @pytest.fixture(scope="session")
203 def app_client_two_attached_databases_crossdb_enabled():
204     with make_app_client(
209 
210 
211 @pytest.fixture(scope="session")
212 def app_client_conflicting_database_names():
213     with make_app_client(
217 
218 
219 @pytest.fixture(scope="session")
220 def app_client_two_attached_databases_one_immutable():
221     with make_app_client(
225 
226 
227 @pytest.fixture(scope="session")
228 def app_client_with_trace():
229     with make_app_client(settings={"trace_debug": True}, is_immutable=True) as client:
231 
232 
233 @pytest.fixture(scope="session")
234 def app_client_shorter_time_limit():
235     with make_app_client(20) as client:
237 
238 
239 @pytest.fixture(scope="session")
240 def app_client_returned_rows_matches_page_size():
241     with make_app_client(max_returned_rows=50) as client:
243 
244 
245 @pytest.fixture(scope="session")
246 def app_client_larger_cache_size():
247     with make_app_client(settings={"cache_size_kb": 2500}) as client:
249 
250 
251 @pytest.fixture(scope="session")
252 def app_client_csv_max_mb_one():
253     with make_app_client(settings={"max_csv_mb": 1}) as client:
255 
256 
257 @pytest.fixture(scope="session")
258 def app_client_with_dot():
259     with make_app_client(filename="fixtures.dot.db") as client:
261 
262 
263 @pytest.fixture(scope="session")
264 def app_client_with_cors():
265     with make_app_client(is_immutable=True, cors=True) as client:
267 
268 
269 @pytest.fixture(scope="session")
270 def app_client_immutable_and_inspect_file():
271     inspect_data = {"fixtures": {"tables": {"sortable": {"count": 100}}}}

datasette/tests/test_api.py

19  )
20  import pathlib
21  import pytest
22  import sys
23  import urllib
24  
25  
26  @pytest.mark.asyncio
27  async def test_homepage(ds_client):
28      response = await ds_client.get("/.json")
43  
44  
45  @pytest.mark.asyncio
46  async def test_homepage_sort_by_relationships(ds_client):
47      response = await ds_client.get("/.json?_sort=relationships")
59  
60  
61  @pytest.mark.asyncio
62  async def test_database_page(ds_client):
63      response = await ds_client.get("/fixtures.json")
617 
618 
619 @pytest.mark.parametrize(
620     "path,expected_redirect",
621     (
638 
639 
640 @pytest.mark.asyncio
641 async def test_custom_sql(ds_client):
642     response = await ds_client.get(
679 
680 
681 @pytest.mark.asyncio
682 async def test_custom_sql_time_limit(ds_client):
683     response = await ds_client.get("/fixtures.json?sql=select+sleep(0.01)")
688 
689 
690 @pytest.mark.asyncio
691 async def test_invalid_custom_sql(ds_client):
692     response = await ds_client.get("/fixtures.json?sql=.schema")
696 
697 
698 @pytest.mark.asyncio
699 async def test_row(ds_client):
700     response = await ds_client.get("/fixtures/simple_primary_key/1.json?_shape=objects")
703 
704 
705 @pytest.mark.asyncio
706 async def test_row_strange_table_name(ds_client):
707     response = await ds_client.get(
712 
713 
714 @pytest.mark.asyncio
715 async def test_row_foreign_key_tables(ds_client):
716     response = await ds_client.get(
773 
774 
775 @pytest.mark.asyncio
776 async def test_metadata_json(ds_client):
777     response = await ds_client.get("/-/metadata.json")
779 
780 
781 @pytest.mark.asyncio
782 async def test_threads_json(ds_client):
783     response = await ds_client.get("/-/threads.json")
788 
789 
790 @pytest.mark.asyncio
791 async def test_plugins_json(ds_client):
792     response = await ds_client.get("/-/plugins.json")
799 
800 
801 @pytest.mark.asyncio
802 async def test_versions_json(ds_client):
803     response = await ds_client.get("/-/versions.json")
816 
817 
818 @pytest.mark.asyncio
819 async def test_settings_json(ds_client):
820     response = await ds_client.get("/-/settings.json")
846 
847 
848 @pytest.mark.asyncio
849 @pytest.mark.parametrize(
850     "path,expected_redirect",
851     (
865 
866 
867 @pytest.mark.asyncio
868 @pytest.mark.parametrize(
869     "extra_args,expected",
870     [
911 
912 
913 @pytest.mark.parametrize(
914     "path,status_code",
915     [
954 
955 
956 @pytest.mark.parametrize(
957     "path",
958     (
990 
991 
992 @pytest.mark.asyncio
993 async def test_http_options_request(ds_client):
994     response = await ds_client.options("/fixtures")
997 
998 
999 @pytest.mark.asyncio
1000async def test_db_path(app_client):
1001    # Needs app_client because needs file based database
1011
1012
1013@pytest.mark.asyncio
1014async def test_hidden_sqlite_stat1_table():
1015    ds = Datasette()
1023
1024
1025@pytest.mark.asyncio
1026@pytest.mark.parametrize("db_name", ("foo", r"fo%o", "f~/c.d"))
1027async def test_tilde_encoded_database_names(db_name):
1028    ds = Datasette()

datasette/tests/test_api_write.py

1   from datasette.app import Datasette
2   from datasette.utils import sqlite3
3   import pytest
4   import time
5   
6   
7   @pytest.fixture
8   def ds_write(tmp_path_factory):
9       db_directory = tmp_path_factory.mktemp("dbs")
36  
37  
38  @pytest.mark.asyncio
39  async def test_insert_row(ds_write):
40      token = write_token(ds_write)
52  
53  
54  @pytest.mark.asyncio
55  @pytest.mark.parametrize("return_rows", (True, False))
56  async def test_insert_rows(ds_write, return_rows):
57      token = write_token(ds_write)
85  
86  
87  @pytest.mark.asyncio
88  @pytest.mark.parametrize(
89      "path,input,special_case,expected_status,expected_errors",
90      (
311 
312 
313 @pytest.mark.asyncio
314 @pytest.mark.parametrize(
315     "ignore,replace,expected_rows",
316     (
331     ),
332 )
333 @pytest.mark.parametrize("should_return", (True, False))
334 async def test_insert_ignore_replace(
335     ds_write, ignore, replace, expected_rows, should_return
364 
365 
366 @pytest.mark.asyncio
367 @pytest.mark.parametrize(
368     "initial,input,expected_rows",
369     (
406     ),
407 )
408 @pytest.mark.parametrize("should_return", (False, True))
409 async def test_upsert(ds_write, initial, input, expected_rows, should_return):
410     token = write_token(ds_write)
449 
450 
451 @pytest.mark.asyncio
452 @pytest.mark.parametrize("scenario", ("no_token", "no_perm", "bad_table"))
453 async def test_delete_row_errors(ds_write, scenario):
454     if scenario == "no_token":
478 
479 
480 @pytest.mark.asyncio
481 @pytest.mark.parametrize(
482     "table,row_for_create,pks,delete_path",
483     (
536 
537 
538 @pytest.mark.asyncio
539 @pytest.mark.parametrize("scenario", ("no_token", "no_perm", "bad_table"))
540 async def test_update_row_check_permission(ds_write, scenario):
541     if scenario == "no_token":
566 
567 
568 @pytest.mark.asyncio
569 @pytest.mark.parametrize(
570     "input,expected_errors",
571     (
578     ),
579 )
580 @pytest.mark.parametrize("use_return", (True, False))
581 async def test_update_row(ds_write, input, expected_errors, use_return):
582     token = write_token(ds_write)
620 
621 
622 @pytest.mark.asyncio
623 @pytest.mark.parametrize(
624     "scenario", ("no_token", "no_perm", "bad_table", "has_perm", "immutable")
625 )
678 
679 
680 @pytest.mark.asyncio
681 @pytest.mark.parametrize(
682     "input,expected_status,expected_response",
683     (
1097
1098
1099@pytest.mark.asyncio
1100@pytest.mark.parametrize(
1101    "permissions,body,expected_status,expected_errors",
1102    (
1141
1142
1143@pytest.mark.asyncio
1144@pytest.mark.parametrize(
1145    "input,expected_rows_after",
1146    (
1205
1206
1207@pytest.mark.asyncio
1208async def test_create_table_error_if_pk_changed(ds_write):
1209    token = write_token(ds_write)
1236
1237
1238@pytest.mark.asyncio
1239async def test_create_table_error_rows_twice_with_duplicates(ds_write):
1240    # Error if you don't send ignore: True or replace: True
1263
1264
1265@pytest.mark.asyncio
1266@pytest.mark.parametrize(
1267    "path",
1268    (

datasette/tests/test_auth.py

5   from datasette.utils import baseconv
6   from datasette.cli import cli
7   import pytest
8   import time
9   
10  
11  @pytest.mark.asyncio
12  async def test_auth_token(ds_client):
13      """The /-/auth-token endpoint sets the correct cookie"""
25  
26  
27  @pytest.mark.asyncio
28  async def test_actor_cookie(ds_client):
29      """A valid actor cookie sets request.scope['actor']"""
33  
34  
35  @pytest.mark.asyncio
36  async def test_actor_cookie_invalid(ds_client):
37      cookie = ds_client.actor_cookie({"id": "test"})
45  
46  
47  @pytest.mark.asyncio
48  @pytest.mark.parametrize(
49      "offset,expected",
50      [
94  
95  
96  @pytest.mark.asyncio
97  @pytest.mark.parametrize("path", ["/", "/fixtures", "/fixtures/facetable"])
98  async def test_logout_button_in_navigation(ds_client, path):
99      response = await ds_client.get(
109 
110 
111 @pytest.mark.asyncio
112 @pytest.mark.parametrize("path", ["/", "/fixtures", "/fixtures/facetable"])
113 async def test_no_logout_button_in_navigation_if_no_ds_actor_cookie(ds_client, path):
114     response = await ds_client.get(path + "?_bot=1")
117 
118 
119 @pytest.mark.parametrize(
120     "post_data,errors,expected_duration,expected_r",
121     (
207 
208 
209 @pytest.mark.asyncio
210 async def test_auth_create_token_not_allowed_for_tokens(ds_client):
211     ds_tok = ds_client.ds.sign({"a": "test", "token": "dstok"}, "token")
217 
218 
219 @pytest.mark.asyncio
220 async def test_auth_create_token_not_allowed_if_allow_signed_tokens_off(ds_client):
221     ds_client.ds._settings["allow_signed_tokens"] = False
230 
231 
232 @pytest.mark.asyncio
233 @pytest.mark.parametrize(
234     "scenario,should_work",
235     (
283 
284 
285 @pytest.mark.parametrize("expires", (None, 1000, -1000))
286 def test_cli_create_token(app_client, expires):
287     secret = app_client.ds._secret

datasette/tests/test_black.py

2   from click.testing import CliRunner
3   from pathlib import Path
4   import pytest
5   import sys
6   

datasette/tests/test_canned_queries.py

1   from bs4 import BeautifulSoup as Soup
2   import json
3   import pytest
4   import re
5   from .fixtures import make_app_client, app_client
6   
7   
8   @pytest.fixture
9   def canned_write_client(tmpdir):
10      template_dir = tmpdir / "canned_write_templates"
54  
55  
56  @pytest.fixture
57  def canned_write_immutable_client():
58      with make_app_client(
74  
75  
76  @pytest.mark.asyncio
77  async def test_canned_query_with_named_parameter(ds_client):
78      response = await ds_client.get(
104 
105 
106 @pytest.mark.parametrize(
107     "query_name,expect_csrf_hidden_field",
108     [
202 
203 
204 @pytest.mark.parametrize(
205     "headers,body,querystring",
206     (
274 
275 
276 @pytest.fixture(scope="session")
277 def magic_parameters_client():
278     with make_app_client(
292 
293 
294 @pytest.mark.parametrize(
295     "magic_parameter,expected_re",
296     [
343 
344 
345 @pytest.mark.parametrize("use_csrf", [True, False])
346 @pytest.mark.parametrize("return_json", [True, False])
347 def test_magic_parameters_csrf_json(magic_parameters_client, use_csrf, return_json):
348     magic_parameters_client.ds._metadata["databases"]["data"]["queries"]["runme_post"][

datasette/tests/test_cli.py

16  import json
17  import pathlib
18  import pytest
19  import sys
20  import textwrap
58  
59  
60  @pytest.mark.parametrize(
61      "spatialite_paths,should_suggest_load_extension",
62      (
170 
171 
172 @pytest.mark.parametrize("flag", ["-U", "--upgrade"])
173 @mock.patch("datasette.cli.run_module")
174 def test_install_upgrade(run_module, flag):
210 
211 
212 @pytest.mark.parametrize("invalid_port", ["-1", "0.5", "dog", "65536"])
213 def test_serve_invalid_ports(invalid_port):
214     runner = CliRunner(mix_stderr=False)
234 
235 
236 @pytest.mark.parametrize("default_allow_sql", (True, False))
237 def test_setting_default_allow_sql(default_allow_sql):
238     runner = CliRunner()
318 
319 
320 @pytest.mark.parametrize(
321     "filename", ["test-database (1).sqlite", "database (1).sqlite"]
322 )
347 
348 
349 @pytest.mark.parametrize("setting", ("hash_urls", "default_cache_ttl_hashed"))
350 def test_help_error_on_hash_urls_setting(setting):
351     runner = CliRunner()

datasette/tests/test_cli_serve_server.py

1   import httpx
2   import pytest
3   import socket
4   
5   
6   @pytest.mark.serial
7   def test_serve_localhost_http(ds_localhost_http_server):
8       response = httpx.get("http://localhost:8041/_memory.json")
14  
15  
16  @pytest.mark.serial
17  @pytest.mark.skipif(
18      not hasattr(socket, "AF_UNIX"), reason="Requires socket.AF_UNIX support"
19  )

datasette/tests/test_config_dir.py

1   import json
2   import pathlib
3   import pytest
4   
5   from datasette.app import Datasette
28  
29  
30  @pytest.fixture(scope="session")
31  def config_dir(tmp_path_factory):
32      config_dir = tmp_path_factory.mktemp("config-dir")
87      )
88      try:
89          with pytest.raises(StartupError) as ex:
90              ds = Datasette([], config_dir=config_dir)
91          assert ex.value.args[0] == "Invalid setting 'invalid' in settings.json"
94  
95  
96  @pytest.fixture(scope="session")
97  def config_dir_client(config_dir):
98      ds = Datasette([], config_dir=config_dir)
159 
160 
161 @pytest.mark.parametrize("filename", ("metadata.yml", "metadata.yaml"))
162 def test_metadata_yaml(tmp_path_factory, filename):
163     config_dir = tmp_path_factory.mktemp("yaml-config-dir")

datasette/tests/test_csv.py

1   from bs4 import BeautifulSoup as Soup
2   import pytest
3   from .fixtures import (  # noqa
4       app_client,
55  
56  
57  @pytest.mark.asyncio
58  async def test_table_csv(ds_client):
59      response = await ds_client.get("/fixtures/simple_primary_key.csv?_oh=1")
70  
71  
72  @pytest.mark.asyncio
73  async def test_table_csv_no_header(ds_client):
74      response = await ds_client.get("/fixtures/simple_primary_key.csv?_header=off")
79  
80  
81  @pytest.mark.asyncio
82  async def test_table_csv_with_labels(ds_client):
83      response = await ds_client.get("/fixtures/facetable.csv?_labels=1")
87  
88  
89  @pytest.mark.asyncio
90  async def test_table_csv_with_nullable_labels(ds_client):
91      response = await ds_client.get("/fixtures/foreign_key_references.csv?_labels=1")
95  
96  
97  @pytest.mark.asyncio
98  async def test_table_csv_blob_columns(ds_client):
99      response = await ds_client.get("/fixtures/binary_data.csv")
108 
109 
110 @pytest.mark.asyncio
111 async def test_custom_sql_csv_blob_columns(ds_client):
112     response = await ds_client.get(
123 
124 
125 @pytest.mark.asyncio
126 async def test_custom_sql_csv(ds_client):
127     response = await ds_client.get(
133 
134 
135 @pytest.mark.asyncio
136 async def test_table_csv_download(ds_client):
137     response = await ds_client.get("/fixtures/simple_primary_key.csv?_dl=1")
144 
145 
146 @pytest.mark.asyncio
147 async def test_csv_with_non_ascii_characters(ds_client):
148     response = await ds_client.get(
168 
169 
170 @pytest.mark.asyncio
171 async def test_table_csv_stream(ds_client):
172     # Without _stream should return header + 100 rows:

datasette/tests/test_custom_pages.py

1   import pathlib
2   import pytest
3   from .fixtures import make_app_client
4   
6   
7   
8   @pytest.fixture(scope="session")
9   def custom_pages_client():
10      with make_app_client(template_dir=TEST_TEMPLATE_DIRS) as client:
12  
13  
14  @pytest.fixture(scope="session")
15  def custom_pages_client_with_base_url():
16      with make_app_client(
79  
80  
81  @pytest.mark.parametrize(
82      "path,expected",
83      [

datasette/tests/test_docs.py

7   from datasette.filters import Filters
8   from pathlib import Path
9   import pytest
10  import re
11  
24  
25  
26  @pytest.fixture(scope="session")
27  def settings_headings():
28      return get_headings((docs_path / "settings.rst").read_text(), "~")
29  
30  
31  @pytest.mark.parametrize("setting", app.SETTINGS)
32  def test_settings_are_documented(settings_headings, setting):
33      assert setting.name in settings_headings
34  
35  
36  @pytest.fixture(scope="session")
37  def plugin_hooks_content():
38      return (docs_path / "plugin_hooks.rst").read_text()
39  
40  
41  @pytest.mark.parametrize(
42      "plugin", [name for name in dir(app.pm.hook) if not name.startswith("_")]
43  )
54  
55  
56  @pytest.fixture(scope="session")
57  def documented_views():
58      view_labels = set()
67  
68  
69  @pytest.mark.parametrize("view_class", [v for v in dir(app) if v.endswith("View")])
70  def test_view_classes_are_documented(documented_views, view_class):
71      assert view_class in documented_views
72  
73  
74  @pytest.fixture(scope="session")
75  def documented_table_filters():
76      json_api_rst = (docs_path / "json_api.rst").read_text()
84  
85  
86  @pytest.mark.parametrize("filter", [f.key for f in Filters._filters])
87  def test_table_filters_are_documented(documented_table_filters, filter):
88      assert filter in documented_table_filters
89  
90  
91  @pytest.fixture(scope="session")
92  def documented_fns():
93      internals_rst = (docs_path / "internals.rst").read_text()
100 
101 
102 @pytest.mark.parametrize("fn", utils.functions_marked_as_documented)
103 def test_functions_marked_with_documented_are_documented(documented_fns, fn):
104     assert fn.__name__ in documented_fns

datasette/tests/test_facets.py

6   from .fixtures import make_app_client
7   import json
8   import pytest
9   
10  
11  @pytest.mark.asyncio
12  async def test_column_facet_suggest(ds_client):
13      facet = ColumnFacet(
37  
38  
39  @pytest.mark.asyncio
40  async def test_column_facet_suggest_skip_if_already_selected(ds_client):
41      facet = ColumnFacet(
75  
76  
77  @pytest.mark.asyncio
78  async def test_column_facet_suggest_skip_if_enabled_by_metadata(ds_client):
79      facet = ColumnFacet(
97  
98  
99  @pytest.mark.asyncio
100 async def test_column_facet_results(ds_client):
101     facet = ColumnFacet(
149 
150 
151 @pytest.mark.asyncio
152 async def test_column_facet_results_column_starts_with_underscore(ds_client):
153     facet = ColumnFacet(
271 
272 
273 @pytest.mark.asyncio
274 async def test_column_facet_from_metadata_cannot_be_hidden(ds_client):
275     facet = ColumnFacet(
324 
325 
326 @pytest.mark.asyncio
327 @pytest.mark.skipif(not detect_json1(), reason="Requires the SQLite json1 module")
328 async def test_array_facet_suggest(ds_client):
329     facet = ArrayFacet(
344 
345 
346 @pytest.mark.asyncio
347 @pytest.mark.skipif(not detect_json1(), reason="Requires the SQLite json1 module")
348 async def test_array_facet_suggest_not_if_all_empty_arrays(ds_client):
349     facet = ArrayFacet(
358 
359 
360 @pytest.mark.asyncio
361 @pytest.mark.skipif(not detect_json1(), reason="Requires the SQLite json1 module")
362 async def test_array_facet_results(ds_client):
363     facet = ArrayFacet(
404 
405 
406 @pytest.mark.asyncio
407 @pytest.mark.skipif(not detect_json1(), reason="Requires the SQLite json1 module")
408 async def test_array_facet_handle_duplicate_tags():
409     ds = Datasette([], memory=True)
459 
460 
461 @pytest.mark.asyncio
462 async def test_date_facet_results(ds_client):
463     facet = DateFacet(
511 
512 
513 @pytest.mark.asyncio
514 async def test_json_array_with_blanks_and_nulls():
515     ds = Datasette([], memory=True)
529 
530 
531 @pytest.mark.asyncio
532 async def test_facet_size():
533     ds = Datasette([], memory=True, settings={"max_returned_rows": 50})
632 
633 
634 @pytest.mark.asyncio
635 async def test_conflicting_facet_names_json(ds_client):
636     response = await ds_client.get(

datasette/tests/test_filters.py

1   from datasette.filters import Filters, through_filters, where_filters, search_filters
2   from datasette.utils.asgi import Request
3   import pytest
4   
5   
6   @pytest.mark.parametrize(
7       "args,expected_where,expected_params",
8       [
78  
79  
80  @pytest.mark.asyncio
81  async def test_through_filters_from_request(ds_client):
82      request = Request.fake(
99  
100 
101 @pytest.mark.asyncio
102 async def test_through_filters_from_request(ds_client):
103     request = Request.fake(
120 
121 
122 @pytest.mark.asyncio
123 async def test_where_filters_from_request(ds_client):
124     await ds_client.ds.invoke_startup()
137 
138 
139 @pytest.mark.asyncio
140 async def test_search_filters_from_request(ds_client):
141     request = Request.fake("/?_search=bobcat")

datasette/tests/test_html.py

12  import json
13  import pathlib
14  import pytest
15  import re
16  import urllib.parse
49  
50  
51  @pytest.mark.asyncio
52  async def test_http_head(ds_client):
53      response = await ds_client.head("/")
55  
56  
57  @pytest.mark.asyncio
58  async def test_homepage_options(ds_client):
59      response = await ds_client.options("/")
62  
63  
64  @pytest.mark.asyncio
65  async def test_favicon(ds_client):
66      response = await ds_client.get("/favicon.ico")
71  
72  
73  @pytest.mark.asyncio
74  async def test_static(ds_client):
75      response = await ds_client.get("/-/static/app2.css")
105 
106 
107 @pytest.mark.asyncio
108 async def test_database_page(ds_client):
109     response = await ds_client.get("/fixtures")
151 
152 
153 @pytest.mark.asyncio
154 async def test_invalid_custom_sql(ds_client):
155     response = await ds_client.get("/fixtures?sql=.schema")
158 
159 
160 @pytest.mark.asyncio
161 async def test_disallowed_custom_sql_pragma(ds_client):
162     response = await ds_client.get(
217 
218 
219 @pytest.mark.asyncio
220 @pytest.mark.parametrize(
221     "path,expected_classes",
222     [
249 
250 
251 @pytest.mark.asyncio
252 @pytest.mark.parametrize(
253     "path,expected_considered",
254     [
275 
276 
277 @pytest.mark.asyncio
278 async def test_row_json_export_link(ds_client):
279     response = await ds_client.get("/fixtures/simple_primary_key/1")
282 
283 
284 @pytest.mark.asyncio
285 async def test_query_json_csv_export_links(ds_client):
286     response = await ds_client.get("/fixtures?sql=select+1")
290 
291 
292 @pytest.mark.asyncio
293 async def test_row_html_simple_primary_key(ds_client):
294     response = await ds_client.get("/fixtures/simple_primary_key/1")
304 
305 
306 @pytest.mark.asyncio
307 async def test_row_html_no_primary_key(ds_client):
308     response = await ds_client.get("/fixtures/no_primary_key/1")
326 
327 
328 @pytest.mark.asyncio
329 @pytest.mark.parametrize(
330     "path,expected_text,expected_link",
331     (
357 
358 
359 @pytest.mark.asyncio
360 @pytest.mark.parametrize(
361     "path,expected",
362     (
395 
396 
397 @pytest.mark.asyncio
398 async def test_index_metadata(ds_client):
399     response = await ds_client.get("/")
408 
409 
410 @pytest.mark.asyncio
411 async def test_database_metadata(ds_client):
412     response = await ds_client.get("/fixtures")
423 
424 
425 @pytest.mark.asyncio
426 async def test_database_metadata_with_custom_sql(ds_client):
427     response = await ds_client.get("/fixtures?sql=select+*+from+simple_primary_key")
506 
507 
508 @pytest.mark.asyncio
509 @pytest.mark.parametrize("path", ["/404", "/fixtures/404"])
510 async def test_404(ds_client, path):
511     response = await ds_client.get(path)
517 
518 
519 @pytest.mark.asyncio
520 @pytest.mark.parametrize(
521     "path,expected_redirect",
522     [("/fixtures/", "/fixtures"), ("/fixtures/simple_view/", "/fixtures/simple_view")],
528 
529 
530 @pytest.mark.asyncio
531 async def test_404_content_type(ds_client):
532     response = await ds_client.get("/404")
535 
536 
537 @pytest.mark.asyncio
538 async def test_canned_query_default_title(ds_client):
539     response = await ds_client.get("/fixtures/magic_parameters")
543 
544 
545 @pytest.mark.asyncio
546 async def test_canned_query_with_custom_metadata(ds_client):
547     response = await ds_client.get("/fixtures/neighborhood_search?text=town")
561 
562 
563 @pytest.mark.asyncio
564 async def test_urlify_custom_queries(ds_client):
565     path = "/fixtures?" + urllib.parse.urlencode(
579 
580 
581 @pytest.mark.asyncio
582 async def test_show_hide_sql_query(ds_client):
583     path = "/fixtures?" + urllib.parse.urlencode(
605 
606 
607 @pytest.mark.asyncio
608 async def test_canned_query_with_hide_has_no_hidden_sql(ds_client):
609     # For a canned query the show/hide should NOT have a hidden SQL field
617 
618 
619 @pytest.mark.parametrize(
620     "hide_sql,querystring,expected_hidden,expected_show_hide_link,expected_show_hide_text",
621     (
666 
667 
668 @pytest.mark.asyncio
669 async def test_binary_data_display_in_query(ds_client):
670     response = await ds_client.get("/fixtures?sql=select+*+from+binary_data")
685 
686 
687 @pytest.mark.asyncio
688 @pytest.mark.parametrize(
689     "path,expected_filename",
690     [
708 
709 
710 @pytest.mark.asyncio
711 @pytest.mark.parametrize(
712     "path,expected_message",
713     [
726 
727 
728 @pytest.mark.asyncio
729 async def test_metadata_json_html(ds_client):
730     response = await ds_client.get("/-/metadata")
734 
735 
736 @pytest.mark.asyncio
737 @pytest.mark.parametrize(
738     "path",
739     [
749 
750 
751 @pytest.mark.asyncio
752 async def test_query_error(ds_client):
753     response = await ds_client.get("/fixtures?sql=select+*+from+notatable")
766 
767 
768 @pytest.mark.asyncio
769 async def test_config_template_debug_off(ds_client):
770     response = await ds_client.get("/fixtures/facetable?_context=1")
781 
782 
783 @pytest.mark.parametrize(
784     "path",
785     [
794     ],
795 )
796 @pytest.mark.parametrize("use_prefix", (True, False))
797 def test_base_url_config(app_client_base_url_prefix, path, use_prefix):
798     client = app_client_base_url_prefix
853 
854 
855 @pytest.mark.asyncio
856 @pytest.mark.parametrize(
857     "path,expected",
858     [
884 
885 
886 @pytest.mark.parametrize("permission_allowed", [True, False])
887 def test_edit_sql_link_not_shown_if_user_lacks_permission(permission_allowed):
888     with make_app_client(
899 
900 
901 @pytest.mark.asyncio
902 @pytest.mark.parametrize(
903     "actor_id,should_have_links,should_not_have_links",
904     [
938 
939 
940 @pytest.mark.asyncio
941 async def test_trace_correctly_escaped(ds_client):
942     response = await ds_client.get("/fixtures?sql=select+'<h1>Hello'&_trace=1")
945 
946 
947 @pytest.mark.asyncio
948 @pytest.mark.parametrize(
949     "path,expected",
950     (
999 
1000
1001@pytest.mark.asyncio
1002@pytest.mark.parametrize(
1003    "path",
1004    ("/-/patterns", "/-/messages", "/-/allow-debug", "/fixtures.db"),
1012
1013
1014@pytest.mark.asyncio
1015@pytest.mark.parametrize(
1016    "path,expected",
1017    (
1034
1035
1036@pytest.mark.asyncio
1037@pytest.mark.parametrize(
1038    "path,metadata,expected_links",
1039    (

datasette/tests/test_internal_db.py

1   import pytest
2   
3   
4   @pytest.mark.asyncio
5   async def test_internal_only_available_to_root(ds_client):
6       cookie = ds_client.actor_cookie({"id": "root"})
11  
12  
13  @pytest.mark.asyncio
14  async def test_internal_databases(ds_client):
15      cookie = ds_client.actor_cookie({"id": "root"})
28  
29  
30  @pytest.mark.asyncio
31  async def test_internal_tables(ds_client):
32      cookie = ds_client.actor_cookie({"id": "root"})
41  
42  
43  @pytest.mark.asyncio
44  async def test_internal_indexes(ds_client):
45      cookie = ds_client.actor_cookie({"id": "root"})
62  
63  
64  @pytest.mark.asyncio
65  async def test_internal_foreign_keys(ds_client):
66      cookie = ds_client.actor_cookie({"id": "root"})

datasette/tests/test_internals_database.py

6   from datasette.utils import Column
7   from .fixtures import app_client, app_client_two_attached_databases_crossdb_enabled
8   import pytest
9   import time
10  import uuid
11  
12  
13  @pytest.fixture
14  def db(app_client):
15      return app_client.ds.get_database("fixtures")
16  
17  
18  @pytest.mark.asyncio
19  async def test_execute(db):
20      results = await db.execute("select * from facetable")
23  
24  
25  @pytest.mark.asyncio
26  async def test_results_first(db):
27      assert None is (await db.execute("select * from facetable where pk > 100")).first()
31  
32  
33  @pytest.mark.asyncio
34  @pytest.mark.parametrize("expected", (True, False))
35  async def test_results_bool(db, expected):
36      where = "" if expected else "where pk = 0"
39  
40  
41  @pytest.mark.parametrize(
42      "query,expected",
43      [
47      ],
48  )
49  @pytest.mark.asyncio
50  async def test_results_single_value(db, query, expected):
51      results = await db.execute(query)
53          assert expected == results.single_value()
54      else:
55          with pytest.raises(MultipleValues):
56              results.single_value()
57  
58  
59  @pytest.mark.asyncio
60  async def test_execute_fn(db):
61      def get_1_plus_1(conn):
65  
66  
67  @pytest.mark.parametrize(
68      "tables,exists",
69      (
72      ),
73  )
74  @pytest.mark.asyncio
75  async def test_table_exists(db, tables, exists):
76      for table in tables:
79  
80  
81  @pytest.mark.parametrize(
82      "view,expected",
83      (
86      ),
87  )
88  @pytest.mark.asyncio
89  async def test_view_exists(db, view, expected):
90      actual = await db.view_exists(view)
92  
93  
94  @pytest.mark.parametrize(
95      "table,expected",
96      (
125     ),
126 )
127 @pytest.mark.asyncio
128 async def test_table_columns(db, table, expected):
129     columns = await db.table_columns(table)
131 
132 
133 @pytest.mark.parametrize(
134     "table,expected",
135     (
308     ),
309 )
310 @pytest.mark.asyncio
311 async def test_table_column_details(db, table, expected):
312     columns = await db.table_column_details(table)
322 
323 
324 @pytest.mark.asyncio
325 async def test_get_all_foreign_keys(db):
326     all_foreign_keys = await db.get_all_foreign_keys()
377 
378 
379 @pytest.mark.asyncio
380 async def test_table_names(db):
381     table_names = await db.table_names()
414 
415 
416 @pytest.mark.asyncio
417 async def test_view_names(db):
418     view_names = await db.view_names()
425 
426 
427 @pytest.mark.asyncio
428 async def test_execute_write_block_true(db):
429     await db.execute_write(
434 
435 
436 @pytest.mark.asyncio
437 async def test_execute_write_block_false(db):
438     await db.execute_write(
445 
446 
447 @pytest.mark.asyncio
448 async def test_execute_write_script(db):
449     await db.execute_write_script(
454 
455 
456 @pytest.mark.asyncio
457 async def test_execute_write_many(db):
458     await db.execute_write_script("create table foomany (id integer primary key)")
464 
465 
466 @pytest.mark.asyncio
467 async def test_execute_write_has_correctly_prepared_connection(db):
468     # The sleep() function is only available if ds._prepare_connection() was called
470 
471 
472 @pytest.mark.asyncio
473 async def test_execute_write_fn_block_false(db):
474     def write_fn(conn):
482 
483 
484 @pytest.mark.asyncio
485 async def test_execute_write_fn_block_true(db):
486     def write_fn(conn):
494 
495 
496 @pytest.mark.asyncio
497 async def test_execute_write_fn_exception(db):
498     def write_fn(conn):
499         assert False
500 
501     with pytest.raises(AssertionError):
502         await db.execute_write_fn(write_fn)
503 
504 
505 @pytest.mark.asyncio
506 @pytest.mark.timeout(1)
507 async def test_execute_write_fn_connection_exception(tmpdir, app_client):
508     path = str(tmpdir / "immutable.db")
514         assert False
515 
516     with pytest.raises(AssertionError):
517         await db.execute_write_fn(write_fn)
518 
520 
521 
522 @pytest.mark.asyncio
523 async def test_mtime_ns(db):
524     assert isinstance(db.mtime_ns, int)
537 
538 
539 @pytest.mark.asyncio
540 async def test_attached_databases(app_client_two_attached_databases_crossdb_enabled):
541     database = app_client_two_attached_databases_crossdb_enabled.ds.get_database(
546 
547 
548 @pytest.mark.asyncio
549 async def test_database_memory_name(app_client):
550     ds = app_client.ds
564 
565 
566 @pytest.mark.asyncio
567 async def test_in_memory_databases_forbid_writes(app_client):
568     ds = app_client.ds
569     db = ds.add_database(Database(ds, memory_name="test"))
570     with pytest.raises(sqlite3.OperationalError):
571         await db.execute("create table foo (t text)")
572     assert await db.table_names() == []

datasette/tests/test_internals_datasette.py

5   from datasette.app import Datasette, Database
6   from itsdangerous import BadSignature
7   import pytest
8   
9   
10  @pytest.fixture
11  def datasette(ds_client):
12      return ds_client.ds
16      db = datasette.get_database("fixtures")
17      assert "fixtures" == db.name
18      with pytest.raises(KeyError):
19          datasette.get_database("missing")
20  
26  
27  
28  @pytest.mark.parametrize("value", ["hello", 123, {"key": "value"}])
29  @pytest.mark.parametrize("namespace", [None, "two"])
30  def test_sign_unsign(datasette, value, namespace):
31      extra_args = [namespace] if namespace else []
33      assert value != signed
34      assert value == datasette.unsign(signed, *extra_args)
35      with pytest.raises(BadSignature):
36          datasette.unsign(signed[:-1] + ("!" if signed[-1] != "!" else ":"))
37  
38  
39  @pytest.mark.parametrize(
40      "setting,expected",
41      (
49  
50  
51  @pytest.mark.asyncio
52  async def test_datasette_constructor():
53      ds = Datasette()
66  
67  
68  @pytest.mark.asyncio
69  async def test_num_sql_threads_zero():
70      ds = Datasette([], memory=True, settings={"num_sql_threads": 0})
82  
83  
84  @pytest.mark.asyncio
85  @pytest.mark.parametrize(
86      "actor,metadata,permissions,should_allow,expected_private",
87      (
118     await ds.invoke_startup()
119     if not should_allow:
120         with pytest.raises(Forbidden):
121             await ds.ensure_permissions(actor, permissions)
122     else:
128 
129 
130 @pytest.mark.asyncio
131 async def test_datasette_render_template_no_request():
132     # https://github.com/simonw/datasette/issues/1849
140     # https://github.com/simonw/datasette/issues/1985
141     db_path = str(tmpdir / "data.db")
142     with pytest.raises(ValueError):
143         ds = Datasette(db_path)

datasette/tests/test_internals_datasette_client.py

1   import httpx
2   import pytest
3   import pytest_asyncio
4   
5   
6   @pytest_asyncio.fixture
7   async def datasette(ds_client):
8       await ds_client.ds.invoke_startup()
10  
11  
12  @pytest.mark.asyncio
13  @pytest.mark.parametrize(
14      "method,path,expected_status",
15      [
32  
33  
34  @pytest.mark.asyncio
35  @pytest.mark.parametrize("prefix", [None, "/prefix/"])
36  async def test_client_post(datasette, prefix):
37      original_base_url = datasette._settings["base_url"]
52  
53  
54  @pytest.mark.asyncio
55  @pytest.mark.parametrize(
56      "prefix,expected_path", [(None, "/asgi-scope"), ("/prefix/", "/prefix/asgi-scope")]
57  )

datasette/tests/test_internals_request.py

1   from datasette.utils.asgi import Request
2   import json
3   import pytest
4   
5   
6   @pytest.mark.asyncio
7   async def test_request_post_vars():
8       scope = {
28  
29  
30  @pytest.mark.asyncio
31  async def test_request_post_body():
32      scope = {
72          assert expected[i] == key
73      assert 2 == len(request.args)
74      with pytest.raises(KeyError):
75          request.args["missing"]
76  
108 
109 
110 @pytest.mark.parametrize(
111     "path,query_string,expected_full_path",
112     [("/", "", "/"), ("/", "foo=bar", "/?foo=bar"), ("/foo", "bar", "/foo?bar")],

datasette/tests/test_internals_response.py

1   from datasette.utils.asgi import Response
2   import pytest
3   
4   
30  
31  
32  @pytest.mark.asyncio
33  async def test_response_set_cookie():
34      events = []

datasette/tests/test_internals_urls.py

1   from datasette.app import Datasette
2   from datasette.utils import PrefixedUrlString
3   import pytest
4   
5   
6   @pytest.fixture(scope="module")
7   def ds():
8       return Datasette([], memory=True)
9   
10  
11  @pytest.mark.parametrize(
12      "base_url,path,expected",
13      [
34  
35  
36  @pytest.mark.parametrize(
37      "base_url,expected",
38      [
48  
49  
50  @pytest.mark.parametrize(
51      "base_url,file,expected",
52      [
62  
63  
64  @pytest.mark.parametrize(
65      "base_url,plugin,file,expected",
66      [
86  
87  
88  @pytest.mark.parametrize(
89      "base_url,expected",
90      [
100 
101 
102 @pytest.mark.parametrize(
103     "base_url,format,expected",
104     [
115 
116 
117 @pytest.mark.parametrize(
118     "base_url,name,format,expected",
119     [
134 
135 
136 @pytest.mark.parametrize(
137     "base_url,format,expected",
138     [

datasette/tests/test_load_extensions.py

1   from datasette.app import Datasette
2   import pytest
3   from pathlib import Path
4   
19  
20  
21  @pytest.mark.asyncio
22  @pytest.mark.skipif(not has_compiled_ext(), reason="Requires compiled ext.c")
23  async def test_load_extension_default_entrypoint():
24      # The default entrypoint only loads a() and NOT b() or c(), so those
39  
40  
41  @pytest.mark.asyncio
42  @pytest.mark.skipif(not has_compiled_ext(), reason="Requires compiled ext.c")
43  async def test_load_extension_multiple_entrypoints():
44      # Load in the default entrypoint and the other 2 custom entrypoints, now

datasette/tests/test_messages.py

1   from .utils import cookie_was_deleted
2   import pytest
3   
4   
5   @pytest.mark.asyncio
6   @pytest.mark.parametrize(
7       "qs,expected",
8       [
19  
20  
21  @pytest.mark.asyncio
22  async def test_messages_are_displayed_and_cleared(ds_client):
23      # First set the message cookie

datasette/tests/test_package.py

4   import os
5   import pathlib
6   import pytest
7   
8   
26  
27  
28  @pytest.mark.serial
29  @mock.patch("shutil.which")
30  @mock.patch("datasette.cli.call")

datasette/tests/test_permissions.py

8   import json
9   from pprint import pprint
10  import pytest_asyncio
11  import pytest
12  import re
13  import time
15  
16  
17  @pytest.fixture(scope="module")
18  def padlock_client():
19      with make_app_client(
29  
30  
31  @pytest_asyncio.fixture
32  async def perms_ds():
33      ds = Datasette()
41  
42  
43  @pytest.mark.parametrize(
44      "allow,expected_anon,expected_auth",
45      [
49      ],
50  )
51  @pytest.mark.parametrize(
52      "path",
53      (
78  
79  
80  @pytest.mark.parametrize(
81      "allow,expected_anon,expected_auth",
82      [
162 
163 
164 @pytest.mark.parametrize(
165     "allow,expected_anon,expected_auth",
166     [
222 
223 
224 @pytest.mark.parametrize(
225     "allow,expected_anon,expected_auth",
226     [
251 
252 
253 @pytest.mark.parametrize(
254     "metadata",
255     [
312 
313 
314 @pytest.mark.parametrize(
315     "path,permissions",
316     [
363 
364 
365 @pytest.mark.asyncio
366 async def test_permissions_debug(ds_client):
367     ds_client.ds._permission_checks.clear()
395 
396 
397 @pytest.mark.asyncio
398 @pytest.mark.parametrize(
399     "actor,allow,expected_fragment",
400     [
413 
414 
415 @pytest.mark.parametrize(
416     "allow,expected",
417     [
425 
426 
427 @pytest.fixture(scope="session")
428 def view_instance_client():
429     with make_app_client(metadata={"allow": {}}) as client:
431 
432 
433 @pytest.mark.parametrize(
434     "path",
435     [
454 
455 
456 @pytest.fixture(scope="session")
457 def cascade_app_client():
458     with make_app_client(is_immutable=True) as client:
460 
461 
462 @pytest.mark.parametrize(
463     "path,permissions,expected_status",
464     [
568 
569 
570 @pytest.mark.asyncio
571 @pytest.mark.parametrize(
572     "actor,permission,resource_1,resource_2,expected_result",
573     (
672 
673 
674 @pytest.mark.asyncio
675 @pytest.mark.parametrize(
676     "metadata,actor,action,resource,expected_result",
677     (
848 
849 
850 @pytest.mark.asyncio
851 async def test_actor_endpoint_allows_any_token():
852     ds = Datasette()
871 
872 
873 @pytest.mark.parametrize(
874     "options,expected",
875     (

datasette/tests/test_plugins.py

24  import re
25  import textwrap
26  import pytest
27  import urllib
28  
30  
31  
32  @pytest.mark.parametrize(
33      "plugin_hook", [name for name in dir(pm.hook) if not name.startswith("_")]
34  )
43  
44  
45  @pytest.mark.asyncio
46  async def test_hook_plugins_dir_plugin_prepare_connection(ds_client):
47      response = await ds_client.get(
48          "/fixtures.json?_shape=arrayfirst&sql=select+convert_units(100%2C+'m'%2C+'ft')"
49      )
50      assert response.json()[0] == pytest.approx(328.0839)
51  
52  
53  @pytest.mark.asyncio
54  async def test_hook_plugin_prepare_connection_arguments(ds_client):
55      response = await ds_client.get(
61  
62  
63  @pytest.mark.asyncio
64  @pytest.mark.parametrize(
65      "path,expected_decoded_object",
66      [
125 
126 
127 @pytest.mark.asyncio
128 async def test_hook_extra_js_urls(ds_client):
129     response = await ds_client.get("/")
144 
145 
146 @pytest.mark.asyncio
147 async def test_plugins_with_duplicate_js_urls(ds_client):
148     # If two plugins both require jQuery, jQuery should be loaded only once
171 
172 
173 @pytest.mark.asyncio
174 async def test_hook_render_cell_link_from_json(ds_client):
175     sql = """
186 
187 
188 @pytest.mark.asyncio
189 async def test_hook_render_cell_demo(ds_client):
190     response = await ds_client.get(
203 
204 
205 @pytest.mark.asyncio
206 @pytest.mark.parametrize(
207     "path", ("/fixtures?sql=select+'RENDER_CELL_ASYNC'", "/fixtures/simple_primary_key")
208 )
212 
213 
214 @pytest.mark.asyncio
215 async def test_plugin_config(ds_client):
216     assert {"depth": "table"} == ds_client.ds.plugin_config(
230 
231 
232 @pytest.mark.asyncio
233 async def test_plugin_config_env(ds_client):
234     os.environ["FOO_ENV"] = "FROM_ENVIRONMENT"
240 
241 
242 @pytest.mark.asyncio
243 async def test_plugin_config_env_from_list(ds_client):
244     os.environ["FOO_ENV"] = "FROM_ENVIRONMENT"
254 
255 
256 @pytest.mark.asyncio
257 async def test_plugin_config_file(ds_client):
258     with open(TEMP_PLUGIN_SECRET_FILE, "w") as fp:
267 
268 
269 @pytest.mark.parametrize(
270     "path,expected_extra_body_script",
271     [
326 
327 
328 @pytest.mark.asyncio
329 async def test_hook_asgi_wrapper(ds_client):
330     response = await ds_client.get("/fixtures")
385 
386 
387 @pytest.fixture(scope="session")
388 def view_names_client(tmp_path_factory):
389     tmpdir = tmp_path_factory.mktemp("test-view-names")
420 
421 
422 @pytest.mark.parametrize(
423     "path,view_name",
424     (
437 
438 
439 @pytest.mark.asyncio
440 async def test_hook_register_output_renderer_no_parameters(ds_client):
441     response = await ds_client.get("/fixtures/facetable.testnone")
444 
445 
446 @pytest.mark.asyncio
447 async def test_hook_register_output_renderer_all_parameters(ds_client):
448     response = await ds_client.get("/fixtures/facetable.testall")
496 
497 
498 @pytest.mark.asyncio
499 async def test_hook_register_output_renderer_custom_status_code(ds_client):
500     response = await ds_client.get(
504 
505 
506 @pytest.mark.asyncio
507 async def test_hook_register_output_renderer_custom_content_type(ds_client):
508     response = await ds_client.get(
512 
Powered by Datasette