ripgrep
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",