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"}
94       )
95       auth_app.mock_api_json = {"id": 1, "name": "Simon"}
96       async with httpx.AsyncClient(app=auth_app) as client:
97           response = await client.get("https://demo.example.com/", allow_redirects=False)
98           assert 200 == response.status_code
119      auth_app.mock_api_json = {"forbidden": "Access not allowed"}
120      assert not auth_app.called
121      async with httpx.AsyncClient(app=auth_app) as client:
122          response = await client.get("https://demo.example.com/", allow_redirects=False)
123          assert 403 == response.status_code
144      }
145      assert not auth_app.called
146      async with httpx.AsyncClient(app=auth_app) as client:
147          response = await client.get("https://demo.example.com/", allow_redirects=False)
148          assert 200 == response.status_code

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/datasette/app.py

1367 
1368     async def get(self, path, **kwargs):
1369         async with httpx.AsyncClient(app=self.app) as client:
1370             return await client.get(self._fix(path), **kwargs)
1371 
1372     async def options(self, path, **kwargs):
1373         async with httpx.AsyncClient(app=self.app) as client:
1374             return await client.options(self._fix(path), **kwargs)
1375 
1376     async def head(self, path, **kwargs):
1377         async with httpx.AsyncClient(app=self.app) as client:
1378             return await client.head(self._fix(path), **kwargs)
1379 
1380     async def post(self, path, **kwargs):
1381         async with httpx.AsyncClient(app=self.app) as client:
1382             return await client.post(self._fix(path), **kwargs)
1383 
1384     async def put(self, path, **kwargs):
1385         async with httpx.AsyncClient(app=self.app) as client:
1386             return await client.put(self._fix(path), **kwargs)
1387 
1388     async def patch(self, path, **kwargs):
1389         async with httpx.AsyncClient(app=self.app) as client:
1390             return await client.patch(self._fix(path), **kwargs)
1391 
1392     async def delete(self, path, **kwargs):
1393         async with httpx.AsyncClient(app=self.app) as client:
1394             return await client.delete(self._fix(path), **kwargs)
1395 
1396     async def request(self, method, path, **kwargs):
1397         async with httpx.AsyncClient(app=self.app) as client:
1398             return await client.request(method, self._fix(path), **kwargs)

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-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-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-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-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_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",
213          "TABLE", "testTable" if on else "test_table"
214      )
215      async with httpx.AsyncClient(app=ds.app()) as client:
216          response = await client.post("http://localhost/graphql", json={"query": query})
217          assert response.status_code == 200
255              "ARGS", ", ".join(args)
256          )
257          async with httpx.AsyncClient(app=ds.app()) as client:
258              response = await client.post(
259                  "http://localhost/graphql", json={"query": query}
286      }
287      """
288      async with httpx.AsyncClient(app=ds.app()) as client:
289          response = await client.post(
290              "http://localhost/graphql/test2", json={"query": query}
323      }
324      """
325      async with httpx.AsyncClient(app=ds.app()) as client:
326          response = await client.post("http://localhost/graphql", json={"query": query})
327          assert response.status_code == 200, response.json()
344  @pytest.mark.asyncio
345  async def test_graphql_output_schema(ds):
346      async with httpx.AsyncClient(app=ds.app()) as client:
347          response = await client.options("http://localhost/graphql/test.graphql")
348          assert response.status_code == 200
369          cors=cors_enabled,
370      )
371      async with httpx.AsyncClient(app=ds.app()) as client:
372          response = await client.options("http://localhost/graphql")
373          assert response.status_code == 200
416      }
417      """
418      async with httpx.AsyncClient(app=ds.app()) as client:
419          response = await client.post(
420              "http://localhost/graphql",
475  )
476  async def test_graphql_http_get(ds, query, extra_query_string, expected_data):
477      async with httpx.AsyncClient(app=ds.app()) as client:
478          params = dict(extra_query_string)
479          params["query"] = query
508      }
509      """
510      async with httpx.AsyncClient(app=ds.app()) as client:
511          response = await client.post("http://localhost/graphql", json={"query": query})
512          assert response.status_code == 200
537      }
538      """
539      async with httpx.AsyncClient(app=ds.app()) as client:
540          response = await client.post("http://localhost/graphql", json={"query": query})
541          assert response.status_code == 500
570      }
571      """
572      async with httpx.AsyncClient(app=ds.app()) as client:
573          response = await client.post("http://localhost/graphql", json={"query": query})
574          assert response.status_code == 500
623      }
624      """
625      async with httpx.AsyncClient(app=ds.app()) as client:
626          response = await client.post("http://localhost/graphql", json={"query": query})
627          assert response.status_code == 200
712  async def test_permissions(db_path, db_path2, metadata, expected):
713      ds = Datasette([db_path, db_path2], metadata=metadata)
714      async with httpx.AsyncClient(app=ds.app()) as client:
715          for authenticated, path, expected_status in expected:
716              ds._permission_checks.clear()

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_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-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-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/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/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-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-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-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-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-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-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
36       app = ds.app()
37       async with LifespanManager(app):
38           async with httpx.AsyncClient(app=app) as client:
39               cookies = {}
40               if auth:
60   
61       # First test the upload page exists
62       async with httpx.AsyncClient(app=datasette.app()) as client:
63           response = await client.get("http://localhost/-/upload-csvs", cookies=cookies)
64           assert 200 == response.status_code
101      ds = Datasette([path])
102      app = ds.app()
103      async with httpx.AsyncClient(app=app) as client:
104          response = await client.get("http://localhost/-/upload-csvs")
105          assert 403 == response.status_code
106      # Now try with a root actor
107      async with httpx.AsyncClient(app=app) as client2:
108          response2 = await client2.get(
109              "http://localhost/-/upload-csvs",

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

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:

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
131  )
132  async def test_advanced_search(ds, q, expected):
133      async with httpx.AsyncClient(app=ds.app()) as client:
134          response = await client.get(
135              "http://localhost/-/beta?" + urllib.parse.urlencode({"q": q})
163  )
164  async def test_search_order(ds, sort, expected):
165      async with httpx.AsyncClient(app=ds.app()) as client:
166          q = "email"
167          response = await client.get(
203  )
204  async def test_search_order_for_timeline(ds, sort, expected):
205      async with httpx.AsyncClient(app=ds.app()) as client:
206          url = "http://localhost/-/beta"
207          if sort:
224  @pytest.mark.asyncio
225  async def test_fixture(ds):
226      async with httpx.AsyncClient(app=ds.app()) as client:
227          response = await client.get("http://localhost/-/databases.json")
228          assert 200 == response.status_code
233  async def test_plugin_is_installed():
234      app = Datasette([], memory=True).app()
235      async with httpx.AsyncClient(app=app) as client:
236          response = await client.get("http://localhost/-/plugins.json")
237          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",