ripgrep

Options:

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

datasette-auth-existing-cookies/test_datasette_auth_existing_cookies.py

37          trust_x_forwarded_proto=False,
38      )
39      async with httpx.AsyncClient(app=auth_app) as client:
40          response = await client.get(
41              "https://demo.example.com{}".format(path), allow_redirects=False
66          trust_x_forwarded_proto=trust_it,
67      )
68      async with httpx.AsyncClient(app=auth_app) as client:
69          url = "http://demo.example.com/"
70          headers = {"x-forwarded-proto": "https"}
98      )
99      auth_app.mock_api_json = {"id": 1, "name": "Simon"}
100     async with httpx.AsyncClient(app=auth_app) as client:
101         response = await client.get("https://demo.example.com/", allow_redirects=False)
102         assert 200 == response.status_code
123     auth_app.mock_api_json = {"forbidden": "Access not allowed"}
124     assert not auth_app.called
125     async with httpx.AsyncClient(app=auth_app) as client:
126         response = await client.get("https://demo.example.com/", allow_redirects=False)
127         assert 403 == response.status_code
148     }
149     assert not auth_app.called
150     async with httpx.AsyncClient(app=auth_app) as client:
151         response = await client.get("https://demo.example.com/", allow_redirects=False)
152         assert 200 == response.status_code

datasette-ripgrep/README.md

14  Some example searches:
15  
16  - [with.\*AsyncClient](https://ripgrep.datasette.io/-/ripgrep?pattern=with.*AsyncClient) - regular expression search for `with.*AsyncClient`
17  - [.plugin_config, literal=on](https://ripgrep.datasette.io/-/ripgrep?pattern=.plugin_config\(&literal=on) - a non-regular expression search for `.plugin_config(`
18  - [with.\*AsyncClient glob=datasette/\*\*](https://ripgrep.datasette.io/-/ripgrep?pattern=with.*AsyncClient&glob=datasette%2F%2A%2A) - search for that pattern only within the `datasette/` top folder
19  - ["sqlite-utils\[">\] glob=setup.py](https://ripgrep.datasette.io/-/ripgrep?pattern=%22sqlite-utils%5B%22%3E%5D&glob=setup.py) - a regular expression search for packages that depend on either `sqlite-utils` or `sqlite-utils>=some-version`
20  - [test glob=!\*.html](https://ripgrep.datasette.io/-/ripgrep?pattern=test&glob=%21*.html) - search for the string `test` but exclude results in HTML files

datasette/datasette/app.py

1373
1374    async def get(self, path, **kwargs):
1375        async with httpx.AsyncClient(app=self.app) as client:
1376            return await client.get(self._fix(path), **kwargs)
1377
1378    async def options(self, path, **kwargs):
1379        async with httpx.AsyncClient(app=self.app) as client:
1380            return await client.options(self._fix(path), **kwargs)
1381
1382    async def head(self, path, **kwargs):
1383        async with httpx.AsyncClient(app=self.app) as client:
1384            return await client.head(self._fix(path), **kwargs)
1385
1386    async def post(self, path, **kwargs):
1387        async with httpx.AsyncClient(app=self.app) as client:
1388            return await client.post(self._fix(path), **kwargs)
1389
1390    async def put(self, path, **kwargs):
1391        async with httpx.AsyncClient(app=self.app) as client:
1392            return await client.put(self._fix(path), **kwargs)
1393
1394    async def patch(self, path, **kwargs):
1395        async with httpx.AsyncClient(app=self.app) as client:
1396            return await client.patch(self._fix(path), **kwargs)
1397
1398    async def delete(self, path, **kwargs):
1399        async with httpx.AsyncClient(app=self.app) as client:
1400            return await client.delete(self._fix(path), **kwargs)
1401
1402    async def request(self, method, path, **kwargs):
1403        avoid_path_rewrites = kwargs.pop("avoid_path_rewrites", None)
1404        async with httpx.AsyncClient(app=self.app) as client:
1405            return await client.request(
1406                method, self._fix(path, avoid_path_rewrites), **kwargs

datasette-allow-permissions-debug/tests/test_allow_permissions_debug.py

7   async def test_plugin_is_installed():
8       app = Datasette([], memory=True).app()
9       async with httpx.AsyncClient(app=app) as client:
10          response = await client.get("http://localhost/-/plugins.json")
11          assert 200 == response.status_code
17  async def test_allows_access():
18      app = Datasette([], memory=True).app()
19      async with httpx.AsyncClient(app=app) as client:
20          response = await client.get("http://localhost/-/permissions")
21          assert 200 == response.status_code

datasette-atom/tests/test_atom.py

101 async def test_incorrect_sql_returns_400():
102     app = Datasette([], immutables=[], memory=True).app()
103     async with httpx.AsyncClient(app=app) as client:
104         response = await client.get(
105             "http://localhost/:memory:.atom?sql=select+sqlite_version()"
128     """
129     app = Datasette([], immutables=[], memory=True).app()
130     async with httpx.AsyncClient(app=app) as client:
131         response = await client.get(
132             "http://localhost/:memory:.atom?" + urllib.parse.urlencode({"sql": sql})
151     """
152     app = Datasette([], immutables=[], memory=True).app()
153     async with httpx.AsyncClient(app=app) as client:
154         response = await client.get(
155             "http://localhost/:memory:.atom?" + urllib.parse.urlencode({"sql": sql})
174     """
175     app = Datasette([], immutables=[], memory=True).app()
176     async with httpx.AsyncClient(app=app) as client:
177         response = await client.get(
178             "http://localhost/:memory:.atom?" + urllib.parse.urlencode({"sql": sql})
197     """
198     app = Datasette([], immutables=[], memory=True).app()
199     async with httpx.AsyncClient(app=app) as client:
200         response = await client.get(
201             "http://localhost/:memory:?" + urllib.parse.urlencode({"sql": sql})
205     assert b'<a href="/:memory:.atom' in response.content
206     # But with a different query that link is not shown:
207     async with httpx.AsyncClient(app=app) as client:
208         response = await client.get(
209             "http://localhost/:memory:?"
234         },
235     ).app()
236     async with httpx.AsyncClient(app=app) as client:
237         response = await client.get("http://localhost/:memory:/feed.atom")
238     assert 200 == response.status_code
273         metadata=metadata,
274     ).app()
275     async with httpx.AsyncClient(app=app) as client:
276         response = await client.get("http://localhost/:memory:/latest.atom")
277     assert 200 == response.status_code

datasette-auth-github/datasette_auth_github/views.py

45  
46      # Exchange that code for a token
47      async with httpx.AsyncClient() as client:
48          github_response = await client.post(
49              "https://github.com/login/oauth/access_token",
67      profile_url = "https://api.github.com/user"
68      try:
69          async with httpx.AsyncClient() as client:
70              profile = (
71                  await client.get(

datasette-auth-github/datasette_auth_github/utils.py

13                  org, profile["login"]
14              )
15              async with httpx.AsyncClient() as client:
16                  response = await client.get(
17                      url, headers={"Authorization": "token {}".format(access_token)}
30                  org_slug, team_slug
31              )
32              async with httpx.AsyncClient() as client:
33                  response = await client.get(
34                      lookup_url,
45                  )
46              )
47              async with httpx.AsyncClient() as client:
48                  response = await client.get(
49                      team_membership_url,

datasette-auth-passwords/tests/test_auth_passwords.py

21  async def test_plugin_is_installed():
22      app = Datasette([], memory=True).app()
23      async with httpx.AsyncClient(app=app) as client:
24          response = await client.get("http://localhost/-/plugins.json")
25          assert 200 == response.status_code
49  async def test_password_tool():
50      app = Datasette([], memory=True).app()
51      async with httpx.AsyncClient(app=app) as client:
52          response1 = await client.get("http://localhost/-/password-tool")
53          csrftoken = response1.cookies["ds_csrftoken"]
68      app = Datasette([], memory=True).app()
69      message = "This instance does not have any configured accounts"
70      async with httpx.AsyncClient(app=app) as client:
71          response = await client.get("http://localhost/-/login")
72          assert message in response.text
73      app2 = Datasette([], memory=True, metadata=TEST_METADATA).app()
74      async with httpx.AsyncClient(app=app2) as client2:
75          response2 = await client2.get("http://localhost/-/login")
76          assert message not in response2.text
90  async def test_login(username, password, should_login, expected_username):
91      ds = Datasette([], memory=True, metadata=TEST_METADATA)
92      async with httpx.AsyncClient(app=ds.app()) as client:
93          # Get csrftoken
94          csrftoken = (await client.get("http://localhost/-/login")).cookies[

datasette-auth-tokens/tests/test_auth_tokens.py

66  @pytest.mark.asyncio
67  async def test_token(ds, token, path, expected_status):
68      async with httpx.AsyncClient(app=ds.app()) as client:
69          response = await client.get(
70              "http://localhost{}".format(path),
89  @pytest.mark.asyncio
90  async def test_query(ds, token, path, expected_status):
91      async with httpx.AsyncClient(app=ds.app()) as client:
92          response = await client.get(
93              "http://localhost{}".format(path),
108 @pytest.mark.asyncio
109 async def test_actor(ds, token, expected_actor):
110     async with httpx.AsyncClient(app=ds.app()) as client:
111         response = await client.get(
112             "http://localhost/-/actor.json",
121 @pytest.mark.asyncio
122 async def test_tokens_table_not_visible(ds, path):
123     async with httpx.AsyncClient(app=ds.app()) as client:
124         response = await client.get("http://localhost{}".format(path))
125         assert 403 == response.status_code

datasette-backup/tests/test_backup.py

22  async def test_plugin_is_installed():
23      app = Datasette([], memory=True).app()
24      async with httpx.AsyncClient(app=app) as client:
25          response = await client.get("http://localhost/-/plugins.json")
26          assert 200 == response.status_code
31  @pytest.mark.asyncio
32  async def test_backup_sql(ds):
33      async with httpx.AsyncClient(app=ds.app()) as client:
34          assert (
35              await client.get("http://localhost/-/backup/nope.sql")
65      db["dogs"].enable_fts(["name"])
66      ds = Datasette([db_path])
67      async with httpx.AsyncClient(app=ds.app()) as client:
68          response = await client.get("http://localhost/-/backup/fts.sql")
69      assert response.status_code == 200

datasette-block/tests/test_block.py

16      app = datasette.app()
17      async with LifespanManager(app):
18          async with httpx.AsyncClient(app=app) as client:
19              response = await client.get("http://localhost" + path)
20              assert response.status_code == expected_status_code

datasette-column-inspect/tests/test_column_inspect.py

22  async def test_table_page_has_script_on_it(db_path):
23      app = Datasette([db_path]).app()
24      async with httpx.AsyncClient(app=app) as client:
25          response = await client.get("http://localhost/data/creatures")
26      assert 200 == response.status_code

datasette-configure-fts/tests/test_configure_fts.py

36  async def test_initial_db_is_not_searchable(db_path):
37      app = Datasette([db_path]).app()
38      async with httpx.AsyncClient(app=app) as client:
39          response = await client.get("http://localhost/data.json")
40      assert 200 == response.status_code
48      ds = Datasette([db_path])
49      app = ds.app()
50      async with httpx.AsyncClient(app=app) as client:
51          response = await client.get("http://localhost{}".format(path))
52          assert 403 == response.status_code
53      # Now try with a root actor
54      async with httpx.AsyncClient(app=app) as client2:
55          response2 = await client2.get(
56              "http://localhost{}".format(path),
65      ds = Datasette([db_path])
66      app = ds.app()
67      async with httpx.AsyncClient(app=app) as client:
68          response = await client.get(
69              "http://localhost/-/configure-fts",
78  async def test_database_page_sets_cookie(db_path):
79      ds = Datasette([db_path])
80      async with httpx.AsyncClient(app=ds.app()) as client:
81          response = await client.get(
82              "http://localhost/-/configure-fts/data",
89  async def test_lists_databases_if_more_than_one(db_path, db_path2):
90      ds = Datasette([db_path, db_path2])
91      async with httpx.AsyncClient(app=ds.app()) as client:
92          response = await client.get(
93              "http://localhost/-/configure-fts",
103 async def test_lists_tables_in_database(db_path2):
104     ds = Datasette([db_path2])
105     async with httpx.AsyncClient(app=ds.app()) as client:
106         response = await client.get(
107             "http://localhost/-/configure-fts/data2",
114     assert b"<h2>mammals</h2>" in response.content
115     # If we select just two tables, only those two
116     async with httpx.AsyncClient(app=ds.app()) as client:
117         response2 = await client.get(
118             "http://localhost/-/configure-fts/data2?table=dogs&table=mammals",
136         }
137     )
138     async with httpx.AsyncClient(app=ds.app()) as client:
139         response = await client.get(
140             "http://localhost/-/configure-fts/data?table=mixed_types",
153 async def test_make_table_searchable(db_path):
154     ds = Datasette([db_path])
155     async with httpx.AsyncClient(app=ds.app()) as client:
156         response1 = await client.get(
157             "http://localhost/-/configure-fts/data",
197     db = sqlite_utils.Database(db_path)
198     db["creatures"].enable_fts(["name"])
199     async with httpx.AsyncClient(app=ds.app()) as client:
200         response1 = await client.get(
201             "http://localhost/-/configure-fts/data",
222 async def test_table_actions(db_path, authenticate):
223     ds = Datasette([db_path])
224     async with httpx.AsyncClient(app=ds.app()) as client:
225         cookies = None
226         if authenticate:

datasette-copyable/tests/test_copyable.py

24  @pytest.mark.asyncio
25  async def test_plugin_is_installed(ds):
26      async with httpx.AsyncClient(app=ds.app()) as client:
27          response = await client.get("http://localhost/-/plugins.json")
28          assert 200 == response.status_code
33  @pytest.mark.asyncio
34  async def test_plugin_adds_copyable_extension(ds):
35      async with httpx.AsyncClient(app=ds.app()) as client:
36          response = await client.get("http://localhost/test/dogs")
37          assert 200 == response.status_code
41  @pytest.mark.asyncio
42  async def test_copyable_page_tsv(ds):
43      async with httpx.AsyncClient(app=ds.app()) as client:
44          response = await client.get("http://localhost/test/dogs.copyable")
45          assert 200 == response.status_code
55  @pytest.mark.asyncio
56  async def test_raw_page_tsv(ds):
57      async with httpx.AsyncClient(app=ds.app()) as client:
58          response = await client.get("http://localhost/test/dogs.copyable?_raw=1")
59          assert 200 == response.status_code
67  @pytest.mark.asyncio
68  async def test_copyable_page_github(ds):
69      async with httpx.AsyncClient(app=ds.app()) as client:
70          response = await client.get(
71              "http://localhost/test/dogs.copyable?_table_format=github"
84  @pytest.mark.asyncio
85  async def test_raw_page_github(ds):
86      async with httpx.AsyncClient(app=ds.app()) as client:
87          response = await client.get(
88              "http://localhost/test/dogs.copyable?_table_format=github&_raw=1"
99  @pytest.mark.asyncio
100 async def test_raw_page_tsv_with_labels(ds):
101     async with httpx.AsyncClient(app=ds.app()) as client:
102         response = await client.get(
103             "http://localhost/test/dogs.copyable?_labels=on&_raw=1"
113 @pytest.mark.asyncio
114 async def test_copyable_page_github_with_labels(ds):
115     async with httpx.AsyncClient(app=ds.app()) as client:
116         response = await client.get(
117             "http://localhost/test/dogs.copyable?_table_format=github&_labels=on"

datasette-dateutil/tests/test_dateutil.py

7   async def test_plugin_is_installed():
8       app = Datasette([], memory=True).app()
9       async with httpx.AsyncClient(app=app) as client:
10          response = await client.get("http://localhost/-/plugins.json")
11          assert 200 == response.status_code
63  async def test_dateutil_sql_functions(sql, expected):
64      app = Datasette([], memory=True).app()
65      async with httpx.AsyncClient(app=app) as client:
66          response = await client.get(
67              "http://localhost/:memory:.json",
79  async def test_dateutil_unbounded_rrule_error():
80      app = Datasette([], memory=True).app()
81      async with httpx.AsyncClient(app=app) as client:
82          response = await client.get(
83              "http://localhost/:memory:.json",

datasette-dns/tests/test_dns.py

10  async def test_plugin_is_installed():
11      app = Datasette([], memory=True).app()
12      async with httpx.AsyncClient(app=app) as client:
13          response = await client.get("http://localhost/-/plugins.json")
14          assert 200 == response.status_code
22      m.side_effect = dns.resolver.NoAnswer
23      app = Datasette([], memory=True).app()
24      async with httpx.AsyncClient(app=app) as client:
25          response = await client.get(
26              "http://localhost/:memory:.json?"

datasette-edit-schema/tests/test_edit_schema.py

27  async def test_csrf_required(db_path):
28      ds = Datasette([db_path])
29      async with httpx.AsyncClient(app=ds.app()) as client:
30          response = await client.post(
31              "http://localhost/-/edit-schema/data/creatures",
41  async def test_table_actions(db_path, authenticate):
42      ds = Datasette([db_path])
43      async with httpx.AsyncClient(app=ds.app()) as client:
44          cookies = None
45          if authenticate:
61      ds = Datasette([db_path])
62      cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
63      async with httpx.AsyncClient(app=ds.app()) as client:
64          # Get a csrftoken
65          csrftoken = (
84      db = sqlite_utils.Database(db_path)
85      assert "creatures" in db.table_names()
86      async with httpx.AsyncClient(app=ds.app()) as client:
87          # Get a csrftoken
88          csrftoken = (
112     table = db["creatures"]
113     assert {"name": str, "description": str} == table.columns_dict
114     async with httpx.AsyncClient(app=ds.app()) as client:
115         # Get a csrftoken
116         csrftoken = (
157     ds = Datasette([db_path])
158     cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
159     async with httpx.AsyncClient(app=ds.app()) as client:
160         csrftoken = (
161             await client.get(
228     table = db["creatures"]
229     assert table.columns_dict == {"name": str, "description": str}
230     async with httpx.AsyncClient(app=ds.app()) as client:
231         csrftoken = (
232             await client.get(
250 async def test_static_assets(db_path):
251     ds = Datasette([db_path])
252     async with httpx.AsyncClient(app=ds.app()) as client:
253         for path in (
254             "/-/static-plugins/datasette-edit-schema/draggable.1.0.0-beta.11.bundle.min.js",
268     someuser_cookies = {"ds_actor": ds.sign({"a": {"id": "someuser"}}, "actor")}
269     root_cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
270     async with httpx.AsyncClient(app=ds.app()) as client:
271         response = await client.get(
272             "http://localhost" + path,
297     ds = Datasette([db_path])
298     cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
299     async with httpx.AsyncClient(app=ds.app()) as client:
300         csrftoken = (
301             await client.get(

datasette-edit-templates/tests/test_edit_templates.py

7   async def test_plugin_is_installed():
8       app = Datasette([], memory=True).app()
9       async with httpx.AsyncClient(app=app) as client:
10          response = await client.get("http://localhost/-/plugins.json")
11          assert 200 == response.status_code

datasette-glitch/tests/test_glitch.py

8   async def test_plugin_is_installed():
9       app = Datasette([], memory=True).app()
10      async with httpx.AsyncClient(app=app) as client:
11          response = await client.get("http://localhost/-/plugins.json")
12          assert 200 == response.status_code

datasette-graphql/tests/test_schema_caching.py

30      assert not mock_schema_for_database.called
31      ds = Datasette([str(db_path)])
32      async with httpx.AsyncClient(app=ds.app()) as client:
33          response = await client.get("http://localhost/graphql/schema.graphql")
34          assert response.status_code == 200
43      # The secod hit should NOT call it
44      assert not mock_schema_for_database.called
45      async with httpx.AsyncClient(app=ds.app()) as client:
46          response = await client.get("http://localhost/graphql/schema.graphql")
47          assert response.status_code == 200
57      db["new_table"].insert({"new_column": 1})
58  
59      async with httpx.AsyncClient(app=ds.app()) as client:
60          response = await client.get("http://localhost/graphql/schema.graphql")
61          assert response.status_code == 200

datasette-graphql/tests/test_graphql.py

18  async def test_plugin_is_installed():
19      app = Datasette([], memory=True).app()
20      async with httpx.AsyncClient(app=app) as client:
21          response = await client.get("http://localhost/-/plugins.json")
22          assert 200 == response.status_code
29      ds = Datasette([], memory=True)
30      app = ds.app()
31      async with httpx.AsyncClient(app=app) as client:
32          response = await client.get("http://localhost/")
33          assert 200 == response.status_code
38  async def test_graphiql():
39      app = Datasette([], memory=True).app()
40      async with httpx.AsyncClient(app=app) as client:
41          response = await client.get(
42              "http://localhost/graphql", headers={"Accept": "text/html"}
48      paths.extend(re.findall(r'<link href="(.*?)"', response.text))
49      paths.extend(re.findall(r'<script src="(.*?)"', response.text))
50      async with httpx.AsyncClient(app=app) as client:
51          for path in paths:
52              assert path.startswith("/-/static-plugins/datasette_graphql/")
68      }
69      """
70      async with httpx.AsyncClient(app=ds.app()) as client:
71          response = await client.post("http://localhost/graphql", json={"query": query})
72          assert response.status_code == 200
126 )
127 async def test_graphql_errors(ds, query, expected_errors):
128     async with httpx.AsyncClient(app=ds.app()) as client:
129         response = await client.post("http://localhost/graphql", json={"query": query})
130         assert response.status_code == 500
144         variables = "{}"
145     expected = json.loads(json_re.search(content)[1])
146     async with httpx.AsyncClient(app=ds.app()) as client:
147         response = await client.post(
148             "http://localhost/graphql",
161 @pytest.mark.asyncio
162 async def test_graphql_error(ds):
163     async with httpx.AsyncClient(app=ds.app()) as client:
164         response = await client.post(
165             "http://localhost/graphql",
215         "TABLE", "testTable" if on else "test_table"
216     )
217     async with httpx.AsyncClient(app=ds.app()) as client:
218         response = await client.post("http://localhost/graphql", json={"query": query})
219         assert response.status_code == 200
257             "ARGS", ", ".join(args)
258         )
259         async with httpx.AsyncClient(app=ds.app()) as client:
260             response = await client.post(
261                 "http://localhost/graphql", json={"query": query}
288     }
289     """
290     async with httpx.AsyncClient(app=ds.app()) as client:
291         response = await client.post(
292             "http://localhost/graphql/test2", json={"query": query}
325     }
326     """
327     async with httpx.AsyncClient(app=ds.app()) as client:
328         response = await client.post("http://localhost/graphql", json={"query": query})
329         assert response.status_code == 200, response.json()
346 @pytest.mark.asyncio
347 async def test_graphql_output_schema(ds):
348     async with httpx.AsyncClient(app=ds.app()) as client:
349         response = await client.options("http://localhost/graphql/test.graphql")
350         assert response.status_code == 200
371         cors=cors_enabled,
372     )
373     async with httpx.AsyncClient(app=ds.app()) as client:
374         response = await client.options("http://localhost/graphql")
375         assert response.status_code == 200
418     }
419     """
420     async with httpx.AsyncClient(app=ds.app()) as client:
421         response = await client.post(
422             "http://localhost/graphql",
477 )
478 async def test_graphql_http_get(ds, query, extra_query_string, expected_data):
479     async with httpx.AsyncClient(app=ds.app()) as client:
480         params = dict(extra_query_string)
481         params["query"] = query
510     }
511     """
512     async with httpx.AsyncClient(app=ds.app()) as client:
513         response = await client.post("http://localhost/graphql", json={"query": query})
514         assert response.status_code == 200
539     }
540     """
541     async with httpx.AsyncClient(app=ds.app()) as client:
542         response = await client.post("http://localhost/graphql", json={"query": query})
543         assert response.status_code == 500
572     }
573     """
574     async with httpx.AsyncClient(app=ds.app()) as client:
575         response = await client.post("http://localhost/graphql", json={"query": query})
576         assert response.status_code == 500
625     }
626     """
627     async with httpx.AsyncClient(app=ds.app()) as client:
628         response = await client.post("http://localhost/graphql", json={"query": query})
629         assert response.status_code == 200
714 async def test_permissions(db_path, db_path2, metadata, expected):
715     ds = Datasette([db_path, db_path2], metadata=metadata)
716     async with httpx.AsyncClient(app=ds.app()) as client:
717         for authenticated, path, expected_status in expected:
718             ds._permission_checks.clear()

datasette-graphql/tests/test_template_tag.py

58  
59      ds = Datasette([str(db_path)], template_dir=template_dir)
60      async with httpx.AsyncClient(app=ds.app()) as client:
61          response = await client.get("http://localhost/about")
62          assert response.status_code == 200

datasette-ics/tests/test_ics.py

9   async def test_incorrect_sql_returns_400():
10      app = Datasette([], immutables=[], memory=True).app()
11      async with httpx.AsyncClient(app=app) as client:
12          response = await client.get(
13              "http://localhost/:memory:.ics?sql=select+sqlite_version()"
30      """
31      app = Datasette([], immutables=[], memory=True).app()
32      async with httpx.AsyncClient(app=app) as client:
33          response = await client.get(
34              "http://localhost/:memory:.ics?" + urllib.parse.urlencode({"sql": sql})
67      """
68      app = Datasette([], immutables=[], memory=True).app()
69      async with httpx.AsyncClient(app=app) as client:
70          response = await client.get(
71              "http://localhost/:memory:.ics?"
106     """
107     app = Datasette([], immutables=[], memory=True).app()
108     async with httpx.AsyncClient(app=app) as client:
109         response = await client.get(
110             "http://localhost/:memory:?" + urllib.parse.urlencode({"sql": sql})
114     assert b'<a href="/:memory:.ics' in response.content
115     # But with a different query that link is not shown:
116     async with httpx.AsyncClient(app=app) as client:
117         response = await client.get(
118             "http://localhost/:memory:?"
144         },
145     ).app()
146     async with httpx.AsyncClient(app=app) as client:
147         response = await client.get("http://localhost/:memory:/calendar.ics")
148     assert 200 == response.status_code

datasette-import-table/datasette_import_table/__init__.py

65  
66                      while url:
67                          async with httpx.AsyncClient() as client:
68                              response = await client.get(url)
69                              data = response.json()
103 async def load_first_page(url):
104     url = url + ".json?_shape=objects&_size=max"
105     async with httpx.AsyncClient() as client:
106         response = await client.get(url)
107         if response.status_code != 200:

datasette-indieauth/datasette_indieauth/__init__.py

125         "code_verifier": code_verifier,
126     }
127     async with httpx.AsyncClient() as client:
128         response = await client.post(authorization_endpoint, data=data)
129 

datasette-indieauth/datasette_indieauth/utils.py

131     canonical_url = None
132     chunk = None
133     async with httpx.AsyncClient(max_redirects=5) as client:
134         try:
135             response = await client.get(url)

datasette-import-table/tests/test_import_table.py

15  async def test_plugin_is_installed():
16      app = Datasette([], memory=True).app()
17      async with httpx.AsyncClient(app=app) as client:
18          response = await client.get("http://localhost/-/plugins.json")
19          assert 200 == response.status_code
39      datasette = Datasette([db_path])
40      cookies = {"ds_actor": datasette.sign({"a": {"id": "root"}}, "actor")}
41      async with httpx.AsyncClient(app=datasette.app()) as client:
42          response = await client.get("http://localhost/-/import-table", cookies=cookies)
43          assert 200 == response.status_code
62      ds = Datasette([path])
63      app = ds.app()
64      async with httpx.AsyncClient(app=app) as client:
65          response = await client.get("http://localhost/-/import-table")
66          assert 403 == response.status_code
67      # Now try with a root actor
68      async with httpx.AsyncClient(app=app) as client2:
69          response2 = await client2.get(
70              "http://localhost/-/import-table",

datasette-indieauth/tests/test_indieauth.py

15  async def test_plugin_is_installed():
16      app = Datasette([], memory=True).app()
17      async with httpx.AsyncClient(app=app) as client:
18          response = await client.get("http://localhost/-/plugins.json")
19          assert 200 == response.status_code
35      app = datasette.app()
36      paths = ("/-/actor.json", "/", "/:memory:", "/-/metadata")
37      async with httpx.AsyncClient(app=app) as client:
38          # All pages should 403 and show login form
39          for path in paths:
168     datasette = Datasette([], memory=True)
169     app = datasette.app()
170     async with httpx.AsyncClient(app=app) as client:
171         # Get CSRF token
172         csrftoken = await _get_csrftoken(client)
241     datasette = Datasette([], memory=True)
242     app = datasette.app()
243     async with httpx.AsyncClient(app=app) as client:
244         response = await client.get("http://localhost/-/indieauth/done")
245         assert response.status_code == 400
274     datasette = Datasette([], memory=True)
275     app = datasette.app()
276     async with httpx.AsyncClient(app=app) as client:
277         csrftoken = await _get_csrftoken(client)
278         # Submit the form
300     else:
301         ds_indieauth = bad_cookie
302     async with httpx.AsyncClient(app=app) as client:
303         response = await client.get(
304             "http://localhost/-/indieauth/done",
322     datasette = Datasette([], memory=True)
323     app = datasette.app()
324     async with httpx.AsyncClient(app=app) as client:
325         csrftoken = await _get_csrftoken(client)
326         # Submit the form
352     datasette = Datasette([], memory=True)
353     app = datasette.app()
354     async with httpx.AsyncClient(app=app) as client:
355         csrftoken = await _get_csrftoken(client)
356         # Submit the form

datasette-init/tests/test_init.py

35      )
36      await ds.invoke_startup()
37      async with httpx.AsyncClient(app=ds.app()) as client:
38          response = await client.get("http://localhost/test/dogs.json")
39          assert 200 == response.status_code
67      )
68      await ds.invoke_startup()
69      async with httpx.AsyncClient(app=ds.app()) as client:
70          response = await client.get("http://localhost/test/dogs.json")
71          assert 200 == response.status_code
79      ds = build_datasette(tmp_path_factory, {"views": {"two": "select 1 + 1"}})
80      await ds.invoke_startup()
81      async with httpx.AsyncClient(app=ds.app()) as client:
82          response = await client.get("http://localhost/test/two.json?_shape=array")
83          assert 200 == response.status_code
92          db_init=lambda db: db.create_view("two", "select 1 + 3"),
93      )
94      async with httpx.AsyncClient(app=ds.app()) as client:
95          response = await client.get("http://localhost/test/two.json?_shape=array")
96          assert [{"1 + 3": 4}] == response.json()
104     ds = Datasette([], memory=True)
105     await ds.invoke_startup()
106     async with httpx.AsyncClient(app=ds.app()) as client:
107         assert 200 == (await client.get("http://localhost/")).status_code

datasette-insert/tests/test_insert_api.py

54  async def test_plugin_is_installed(ds):
55      app = ds.app()
56      async with httpx.AsyncClient(app=app) as client:
57          response = await client.get("http://localhost/-/plugins.json")
58          assert 200 == response.status_code
66  async def test_permission_denied_by_default(ds):
67      app = ds.app()
68      async with httpx.AsyncClient(app=app) as client:
69          response = await client.post(
70              "http://localhost/-/insert/data/newtable", json=[{"foo": "bar"}],
110 async def test_insert_creates_table(ds, unsafe, input, pk, expected):
111     app = ds.app()
112     async with httpx.AsyncClient(app=app) as client:
113         response = await client.post(
114             "http://localhost/-/insert/data/newtable{}".format(
129 @pytest.mark.asyncio
130 async def test_insert_alter(ds, unsafe):
131     async with httpx.AsyncClient(app=ds.app()) as client:
132         response = await client.post(
133             "http://localhost/-/insert/data/dogs?pk=id",
157 @pytest.mark.asyncio
158 async def test_permission_denied_by_allow_block(ds_root_only):
159     async with httpx.AsyncClient(app=ds_root_only.app()) as client:
160         response = await client.post(
161             "http://localhost/-/insert/data/dogs?pk=id",
167 @pytest.mark.asyncio
168 async def test_permission_allowed_by_allow_block(ds_root_only):
169     async with httpx.AsyncClient(app=ds_root_only.app()) as client:
170         response = await client.post(
171             "http://localhost/-/insert/data/dogs?pk=id",
240     pm.register(TestPlugin(), name="undo")
241     try:
242         async with httpx.AsyncClient(app=ds_root_only.app()) as client:
243             # First create the table (if we aren't testing create-table) using
244             # the root authenticated API token

datasette-insert-unsafe/tests/test_insert_unsafe.py

18  async def test_plugin_is_installed():
19      app = Datasette([], memory=True).app()
20      async with httpx.AsyncClient(app=app) as client:
21          response = await client.get("http://localhost/-/plugins.json")
22          assert 200 == response.status_code
28  async def test_insert_allowed(db_path):
29      app = Datasette([str(db_path)]).app()
30      async with httpx.AsyncClient(app=app) as client:
31          response = await client.post(
32              "http://localhost/-/insert/data/newtable", json=[{"foo": "bar"}],
40      plugin = pm.unregister(name="insert_unsafe")
41      try:
42          async with httpx.AsyncClient(app=app) as client:
43              response = await client.post(
44                  "http://localhost/-/insert/data/newtable", json=[{"foo": "bar"}],

datasette-media/datasette_media/__init__.py

73      if should_transform:
74          if content is None and content_url:
75              async with httpx.AsyncClient() as client:
76                  response = await client.get(row["content_url"])
77                  content = response.content

datasette-media/tests/test_media.py

24          },
25      ).app()
26      async with httpx.AsyncClient(app=app) as client:
27          response = await client.get("http://localhost/-/media/photos/key")
28      assert 200 == response.status_code
51          },
52      ).app()
53      async with httpx.AsyncClient(app=app) as client:
54          response = await client.get("http://localhost/-/media/text/key")
55      assert 200 == response.status_code
136         },
137     ).app()
138     async with httpx.AsyncClient(app=app) as client:
139         response = await client.get("http://localhost/-/media/photos/1")
140     assert 200 == response.status_code
166         },
167     ).app()
168     async with httpx.AsyncClient(app=app) as client:
169         response = await client.get("http://localhost/-/media/photos/1")
170     assert 200 == response.status_code
193         },
194     ).app()
195     async with httpx.AsyncClient(app=app) as client:
196         response = await client.get("http://localhost/-/media/photos/1")
197     assert 200 == response.status_code
220         },
221     ).app()
222     async with httpx.AsyncClient(app=app) as client:
223         response = await client.get("http://localhost/-/media/photos/1")
224     assert 200 == response.status_code
260         },
261     ).app()
262     async with httpx.AsyncClient(app=app) as client:
263         response = await client.get("http://localhost/-/media/photos/1", params=args)
264     assert 200 == response.status_code

datasette-mask-columns/tests/test_mask_columns.py

12      datasette = Datasette([path], memory=True)
13      # Without the plugin:
14      async with httpx.AsyncClient(app=datasette.app()) as client:
15          response = await client.get("http://localhost/foo/users.json?_shape=array")
16          assert 200 == response.status_code
29          },
30      )
31      async with httpx.AsyncClient(app=datasette2.app()) as client:
32          response = await client.get("http://localhost/foo/users.json?_shape=array")
33          assert 200 == response.status_code

datasette-permissions-sql/tests/test_permissions_sql.py

90  @pytest.mark.asyncio
91  async def test_permissions_sql(ds, actor, table, expected_status):
92      async with httpx.AsyncClient(app=ds.app()) as client:
93          cookies = {}
94          if actor:
105 @pytest.mark.asyncio
106 async def test_fallback(ds, actor, expected_status):
107     async with httpx.AsyncClient(app=ds.app()) as client:
108         cookies = {}
109         if actor:

datasette-plugin-demos/tests/test_plugin_demos.py

7   async def test_plugin_is_installed():
8       app = Datasette([], memory=True).app()
9       async with httpx.AsyncClient(app=app) as client:
10          response = await client.get("http://localhost/-/plugins.json")
11          assert 200 == response.status_code

datasette-psutil/tests/test_psutil.py

7   async def test_datasette_psutil():
8       ds = Datasette([], memory=True)
9       async with httpx.AsyncClient(app=ds.app()) as client:
10          response = await client.get("http://localhost/-/psutil")
11          assert 200 == response.status_code

datasette-saved-queries/tests/test_saved_queries.py

22  @pytest.mark.asyncio
23  async def test_plugin_is_installed(ds):
24      async with httpx.AsyncClient(app=ds.app()) as client:
25          response = await client.get("http://localhost/-/plugins.json")
26          assert 200 == response.status_code
31  @pytest.mark.asyncio
32  async def test_table_created(ds):
33      async with httpx.AsyncClient(app=ds.app()) as client:
34          response = await client.get("http://localhost/data/saved_queries.json")
35          assert 200 == response.status_code
39  @pytest.mark.asyncio
40  async def test_save_query(ds):
41      async with httpx.AsyncClient(app=ds.app()) as client:
42          # Get the csrftoken cookie
43          response1 = await client.get("http://localhost/data/save_query")
69  @pytest.mark.asyncio
70  async def test_save_query_authenticated_actor(ds):
71      async with httpx.AsyncClient(app=ds.app()) as client:
72          response1 = await client.get("http://localhost/data/save_query")
73          response2 = await client.post(

datasette-schema-versions/tests/test_schema_versions.py

8   async def test_plugin_is_installed():
9       app = Datasette([], memory=True).app()
10      async with httpx.AsyncClient(app=app) as client:
11          response = await client.get("http://localhost/-/plugins.json")
12          assert 200 == response.status_code
24      db2["foo"].insert({"bar": 1})
25      app = Datasette([path1, path2]).app()
26      async with httpx.AsyncClient(app=app) as client:
27          response = await client.get("http://localhost/-/schema-versions")
28          assert 200 == response.status_code

datasette-seaborn/tests/test_seaborn.py

24  async def test_requires_seaborn(ds):
25      app = ds.app()
26      async with httpx.AsyncClient(app=app) as client:
27          response = await client.get("http://localhost/penguins/penguins.seaborn")
28          assert response.status_code == 500
36  
37      async def get_dims(path):
38          async with httpx.AsyncClient(app=app) as client:
39              response = await client.get("http://localhost{}".format(path))
40              assert response.status_code == 200

datasette-template-sql/tests/test_template_sql.py

28  @pytest.mark.asyncio
29  async def test_sql_against_named_database(app):
30      async with httpx.AsyncClient(app=app) as client:
31          response = await client.get("http://localhost/news")
32          assert 200 == response.status_code
40  @pytest.mark.asyncio
41  async def test_sql_against_default_database(app):
42      async with httpx.AsyncClient(app=app) as client:
43          response = await client.get("http://localhost/")
44          assert 200 == response.status_code
49  @pytest.mark.asyncio
50  async def test_sql_with_arguments(app):
51      async with httpx.AsyncClient(app=app) as client:
52          response = await client.get("http://localhost/news")
53          assert 200 == response.status_code

datasette-upload-csvs/tests/test_datasette_upload_csvs.py

13      app = ds.app()
14      async with LifespanManager(app):
15          async with httpx.AsyncClient(app=app) as client:
16              response = await client.get("http://localhost/")
17              assert 200 == response.status_code
21  async def test_redirect():
22      datasette = Datasette([], memory=True)
23      async with httpx.AsyncClient(app=datasette.app()) as client:
24          response = await client.get(
25              "http://localhost/-/upload-csv", allow_redirects=False
35      app = ds.app()
36      async with LifespanManager(app):
37          async with httpx.AsyncClient(app=app) as client:
38              cookies = {}
39              if auth:
59  
60      # First test the upload page exists
61      async with httpx.AsyncClient(app=datasette.app()) as client:
62          response = await client.get("http://localhost/-/upload-csvs", cookies=cookies)
63          assert 200 == response.status_code
100     ds = Datasette([path])
101     app = ds.app()
102     async with httpx.AsyncClient(app=app) as client:
103         response = await client.get("http://localhost/-/upload-csvs")
104         assert 403 == response.status_code
105     # Now try with a root actor
106     async with httpx.AsyncClient(app=app) as client2:
107         response2 = await client2.get(
108             "http://localhost/-/upload-csvs",

datasette-vega/tests/test_vega.py

7   async def test_plugin_is_installed():
8       app = Datasette([], memory=True).app()
9       async with httpx.AsyncClient(app=app) as client:
10          response = await client.get("http://localhost/-/plugins.json")
11          assert response.status_code == 200

datasette-write/tests/test_write.py

22  @pytest.mark.asyncio
23  async def test_permission_denied(ds):
24      async with httpx.AsyncClient(app=ds.app()) as client:
25          response = await client.get("http://localhost/-/write")
26          assert 403 == response.status_code
29  @pytest.mark.asyncio
30  async def test_permission_granted_to_root(ds):
31      async with httpx.AsyncClient(app=ds.app()) as client:
32          response = await client.get(
33              "http://localhost/-/write",
47  @pytest.mark.asyncio
48  async def test_execute_write(ds, sql, expected_message):
49      async with httpx.AsyncClient(
50          app=ds.app(), cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
51      ) as client:

datasette-yaml/tests/test_yaml.py

9   async def test_plugin_is_installed():
10      app = Datasette([], memory=True).app()
11      async with httpx.AsyncClient(app=app) as client:
12          response = await client.get("http://localhost/-/plugins.json")
13          assert 200 == response.status_code
29      )
30      app = Datasette([str(db_path)]).app()
31      async with httpx.AsyncClient(app=app) as client:
32          response = await client.get("http://localhost/test/dogs.yaml")
33          assert response.status_code == 200

dogsheep-beta/tests/test_plugin.py

12  @pytest.mark.asyncio
13  async def test_search(ds):
14      async with httpx.AsyncClient(app=ds.app()) as client:
15          response = await client.get("http://localhost/-/beta")
16          assert 200 == response.status_code
141 )
142 async def test_advanced_search(ds, q, expected):
143     async with httpx.AsyncClient(app=ds.app()) as client:
144         response = await client.get(
145             "http://localhost/-/beta?" + urllib.parse.urlencode({"q": q})
173 )
174 async def test_search_order(ds, sort, expected):
175     async with httpx.AsyncClient(app=ds.app()) as client:
176         q = "email"
177         response = await client.get(
213 )
214 async def test_search_order_for_timeline(ds, sort, expected):
215     async with httpx.AsyncClient(app=ds.app()) as client:
216         url = "http://localhost/-/beta"
217         if sort:
234 @pytest.mark.asyncio
235 async def test_fixture(ds):
236     async with httpx.AsyncClient(app=ds.app()) as client:
237         response = await client.get("http://localhost/-/databases.json")
238         assert 200 == response.status_code
243 async def test_plugin_is_installed():
244     app = Datasette([], memory=True).app()
245     async with httpx.AsyncClient(app=app) as client:
246         response = await client.get("http://localhost/-/plugins.json")
247         assert 200 == response.status_code

til/asgi/lifespan-test-httpx.md

23      app = ds.app()
24      async with LifespanManager(app):
25          async with httpx.AsyncClient(app=app) as client:
26              response = await client.get("http://localhost/-/asgi-scope")
27              assert 200 == response.status_code

til/pytest/registering-plugins-in-tests.md

53  @pytest.mark.asyncio
54  async def test_insert_alter(ds, unsafe):
55      async with httpx.AsyncClient(app=ds.app()) as client:
56          response = await client.post(
57              "http://localhost/-/insert/data/dogs?pk=id",