diff --git a/CHANGES/3957.misc b/CHANGES/3957.misc new file mode 100644 index 00000000000..b4f9f58edb9 --- /dev/null +++ b/CHANGES/3957.misc @@ -0,0 +1 @@ +Improve test suite handling of paths and temp files to consistently use pathlib and pytest fixtures. diff --git a/tests/test_client_request.py b/tests/test_client_request.py index f8107ffad88..c54e1828e34 100644 --- a/tests/test_client_request.py +++ b/tests/test_client_request.py @@ -1,7 +1,7 @@ import asyncio import hashlib import io -import os.path +import pathlib import urllib.parse import zlib from http.cookies import BaseCookie, Morsel, SimpleCookie @@ -921,12 +921,11 @@ async def test_chunked_transfer_encoding(loop, conn) -> None: async def test_file_upload_not_chunked(loop) -> None: - here = os.path.dirname(__file__) - fname = os.path.join(here, "aiohttp.png") - with open(fname, "rb") as f: + file_path = pathlib.Path(__file__).parent / "aiohttp.png" + with file_path.open("rb") as f: req = ClientRequest("post", URL("http://python.org/"), data=f, loop=loop) assert not req.chunked - assert req.headers["CONTENT-LENGTH"] == str(os.path.getsize(fname)) + assert req.headers["CONTENT-LENGTH"] == str(file_path.stat().st_size) await req.close() @@ -947,19 +946,17 @@ async def test_precompressed_data_stays_intact(loop) -> None: async def test_file_upload_not_chunked_seek(loop) -> None: - here = os.path.dirname(__file__) - fname = os.path.join(here, "aiohttp.png") - with open(fname, "rb") as f: + file_path = pathlib.Path(__file__).parent / "aiohttp.png" + with file_path.open("rb") as f: f.seek(100) req = ClientRequest("post", URL("http://python.org/"), data=f, loop=loop) - assert req.headers["CONTENT-LENGTH"] == str(os.path.getsize(fname) - 100) + assert req.headers["CONTENT-LENGTH"] == str(file_path.stat().st_size - 100) await req.close() async def test_file_upload_force_chunked(loop) -> None: - here = os.path.dirname(__file__) - fname = os.path.join(here, "aiohttp.png") - with open(fname, "rb") as f: + file_path = pathlib.Path(__file__).parent / "aiohttp.png" + with file_path.open("rb") as f: req = ClientRequest( "post", URL("http://python.org/"), data=f, chunked=True, loop=loop ) diff --git a/tests/test_cookiejar.py b/tests/test_cookiejar.py index cffca3a4b59..9c608959c39 100644 --- a/tests/test_cookiejar.py +++ b/tests/test_cookiejar.py @@ -1,9 +1,8 @@ import asyncio import datetime import itertools -import os +import pathlib import pickle -import tempfile import unittest from http.cookies import BaseCookie, Morsel, SimpleCookie from unittest import mock @@ -200,8 +199,8 @@ async def test_constructor_with_expired( assert jar._loop is loop -async def test_save_load(loop, cookies_to_send, cookies_to_receive) -> None: - file_path = tempfile.mkdtemp() + "/aiohttp.test.cookie" +async def test_save_load(tmp_path, loop, cookies_to_send, cookies_to_receive) -> None: + file_path = pathlib.Path(str(tmp_path)) / "aiohttp.test.cookie" # export cookie jar jar_save = CookieJar(loop=loop) @@ -215,7 +214,6 @@ async def test_save_load(loop, cookies_to_send, cookies_to_receive) -> None: for cookie in jar_load: jar_test[cookie.key] = cookie - os.unlink(file_path) assert jar_test == cookies_to_receive diff --git a/tests/test_helpers.py b/tests/test_helpers.py index 390d2390065..b59528d3468 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -4,7 +4,6 @@ import gc import platform import sys -import tempfile import weakref from math import ceil, modf from pathlib import Path @@ -73,11 +72,21 @@ def test_parse_mimetype(mimetype, expected) -> None: # ------------------- guess_filename ---------------------------------- -def test_guess_filename_with_tempfile() -> None: - with tempfile.TemporaryFile() as fp: +def test_guess_filename_with_file_object(tmp_path) -> None: + file_path = tmp_path / "test_guess_filename" + with file_path.open("w+b") as fp: assert helpers.guess_filename(fp, "no-throw") is not None +def test_guess_filename_with_path(tmp_path) -> None: + file_path = tmp_path / "test_guess_filename" + assert helpers.guess_filename(file_path, "no-throw") is not None + + +def test_guess_filename_with_default() -> None: + assert helpers.guess_filename(None, "no-throw") == "no-throw" + + # ------------------- BasicAuth ----------------------------------- diff --git a/tests/test_multipart.py b/tests/test_multipart.py index c68ba2dd6ff..f9d130e7949 100644 --- a/tests/test_multipart.py +++ b/tests/test_multipart.py @@ -1,6 +1,7 @@ import asyncio import io import json +import pathlib import zlib from unittest import mock @@ -1270,7 +1271,7 @@ async def test_write_preserves_content_disposition(self, buf, stream) -> None: async def test_preserve_content_disposition_header(self, buf, stream): # https://github.com/aio-libs/aiohttp/pull/3475#issuecomment-451072381 - with open(__file__, "rb") as fobj: + with pathlib.Path(__file__).open("rb") as fobj: with aiohttp.MultipartWriter("form-data", boundary=":") as writer: part = writer.append( fobj, @@ -1297,7 +1298,7 @@ async def test_preserve_content_disposition_header(self, buf, stream): async def test_set_content_disposition_override(self, buf, stream): # https://github.com/aio-libs/aiohttp/pull/3475#issuecomment-451072381 - with open(__file__, "rb") as fobj: + with pathlib.Path(__file__).open("rb") as fobj: with aiohttp.MultipartWriter("form-data", boundary=":") as writer: part = writer.append( fobj, @@ -1324,7 +1325,7 @@ async def test_set_content_disposition_override(self, buf, stream): async def test_reset_content_disposition_header(self, buf, stream): # https://github.com/aio-libs/aiohttp/pull/3475#issuecomment-451072381 - with open(__file__, "rb") as fobj: + with pathlib.Path(__file__).open("rb") as fobj: with aiohttp.MultipartWriter("form-data", boundary=":") as writer: part = writer.append( fobj, diff --git a/tests/test_proxy_functional.py b/tests/test_proxy_functional.py index f199404f159..099922ac77f 100644 --- a/tests/test_proxy_functional.py +++ b/tests/test_proxy_functional.py @@ -731,7 +731,7 @@ async def test_proxy_from_env_http_with_auth_from_netrc( auth.login, auth.password, ) - with open(str(netrc_file), "w") as f: + with netrc_file.open("w") as f: f.write(netrc_file_data) mocker.patch.dict( os.environ, {"http_proxy": str(proxy.url), "NETRC": str(netrc_file)} @@ -757,7 +757,7 @@ async def test_proxy_from_env_http_without_auth_from_netrc( auth.login, auth.password, ) - with open(str(netrc_file), "w") as f: + with netrc_file.open("w") as f: f.write(netrc_file_data) mocker.patch.dict( os.environ, {"http_proxy": str(proxy.url), "NETRC": str(netrc_file)} @@ -780,7 +780,7 @@ async def test_proxy_from_env_http_without_auth_from_wrong_netrc( auth = aiohttp.BasicAuth("user", "pass") netrc_file = tmp_path / "test_netrc" invalid_data = f"machine 127.0.0.1 {auth.login} pass {auth.password}" - with open(str(netrc_file), "w") as f: + with netrc_file.open("w") as f: f.write(invalid_data) mocker.patch.dict( diff --git a/tests/test_urldispatch.py b/tests/test_urldispatch.py index 588daed8d40..7c4941f9b3c 100644 --- a/tests/test_urldispatch.py +++ b/tests/test_urldispatch.py @@ -1,4 +1,3 @@ -import os import pathlib import re from collections.abc import Container, Iterable, Mapping, MutableMapping, Sized @@ -49,7 +48,7 @@ def fill_routes(router): def go(): route1 = router.add_route("GET", "/plain", make_handler()) route2 = router.add_route("GET", "/variable/{name}", make_handler()) - resource = router.add_static("/static", os.path.dirname(aiohttp.__file__)) + resource = router.add_static("/static", pathlib.Path(aiohttp.__file__).parent) return [route1, route2] + list(resource) return go @@ -342,7 +341,7 @@ def test_route_dynamic(router) -> None: def test_add_static(router) -> None: resource = router.add_static( - "/st", os.path.dirname(aiohttp.__file__), name="static" + "/st", pathlib.Path(aiohttp.__file__).parent, name="static" ) assert router["static"] is resource url = resource.url_for(filename="/dir/a.txt") @@ -351,7 +350,7 @@ def test_add_static(router) -> None: def test_add_static_append_version(router) -> None: - resource = router.add_static("/st", os.path.dirname(__file__), name="static") + resource = router.add_static("/st", pathlib.Path(__file__).parent, name="static") url = resource.url_for(filename="/data.unknown_mime_type", append_version=True) expect_url = ( "/st/data.unknown_mime_type?" "v=aUsn8CHEhhszc81d28QmlcBW0KQpfS2F4trgQKhOYd8%3D" @@ -361,7 +360,7 @@ def test_add_static_append_version(router) -> None: def test_add_static_append_version_set_from_constructor(router) -> None: resource = router.add_static( - "/st", os.path.dirname(__file__), append_version=True, name="static" + "/st", pathlib.Path(__file__).parent, append_version=True, name="static" ) url = resource.url_for(filename="/data.unknown_mime_type") expect_url = ( @@ -372,7 +371,7 @@ def test_add_static_append_version_set_from_constructor(router) -> None: def test_add_static_append_version_override_constructor(router) -> None: resource = router.add_static( - "/st", os.path.dirname(__file__), append_version=True, name="static" + "/st", pathlib.Path(__file__).parent, append_version=True, name="static" ) url = resource.url_for(filename="/data.unknown_mime_type", append_version=False) expect_url = "/st/data.unknown_mime_type" @@ -380,7 +379,7 @@ def test_add_static_append_version_override_constructor(router) -> None: def test_add_static_append_version_filename_without_slash(router) -> None: - resource = router.add_static("/st", os.path.dirname(__file__), name="static") + resource = router.add_static("/st", pathlib.Path(__file__).parent, name="static") url = resource.url_for(filename="data.unknown_mime_type", append_version=True) expect_url = ( "/st/data.unknown_mime_type?" "v=aUsn8CHEhhszc81d28QmlcBW0KQpfS2F4trgQKhOYd8%3D" @@ -389,27 +388,26 @@ def test_add_static_append_version_filename_without_slash(router) -> None: def test_add_static_append_version_non_exists_file(router) -> None: - resource = router.add_static("/st", os.path.dirname(__file__), name="static") + resource = router.add_static("/st", pathlib.Path(__file__).parent, name="static") url = resource.url_for(filename="/non_exists_file", append_version=True) assert "/st/non_exists_file" == str(url) def test_add_static_append_version_non_exists_file_without_slash(router) -> None: - resource = router.add_static("/st", os.path.dirname(__file__), name="static") + resource = router.add_static("/st", pathlib.Path(__file__).parent, name="static") url = resource.url_for(filename="non_exists_file", append_version=True) assert "/st/non_exists_file" == str(url) -def test_add_static_append_version_follow_symlink(router, tmpdir) -> None: +def test_add_static_append_version_follow_symlink(router, tmp_path) -> None: # Tests the access to a symlink, in static folder with apeend_version - tmp_dir_path = str(tmpdir) - symlink_path = os.path.join(tmp_dir_path, "append_version_symlink") - symlink_target_path = os.path.dirname(__file__) - os.symlink(symlink_target_path, symlink_path, True) + symlink_path = tmp_path / "append_version_symlink" + symlink_target_path = pathlib.Path(__file__).parent + pathlib.Path(str(symlink_path)).symlink_to(str(symlink_target_path), True) # Register global static route: resource = router.add_static( - "/st", tmp_dir_path, follow_symlinks=True, append_version=True + "/st", str(tmp_path), follow_symlinks=True, append_version=True ) url = resource.url_for(filename="/append_version_symlink/data.unknown_mime_type") @@ -421,16 +419,16 @@ def test_add_static_append_version_follow_symlink(router, tmpdir) -> None: assert expect_url == str(url) -def test_add_static_append_version_not_follow_symlink(router, tmpdir) -> None: +def test_add_static_append_version_not_follow_symlink(router, tmp_path) -> None: # Tests the access to a symlink, in static folder with apeend_version - tmp_dir_path = str(tmpdir) - symlink_path = os.path.join(tmp_dir_path, "append_version_symlink") - symlink_target_path = os.path.dirname(__file__) - os.symlink(symlink_target_path, symlink_path, True) + symlink_path = tmp_path / "append_version_symlink" + symlink_target_path = pathlib.Path(__file__).parent + + pathlib.Path(str(symlink_path)).symlink_to(str(symlink_target_path), True) # Register global static route: resource = router.add_static( - "/st", tmp_dir_path, follow_symlinks=False, append_version=True + "/st", str(tmp_path), follow_symlinks=False, append_version=True ) filename = "/append_version_symlink/data.unknown_mime_type" @@ -467,7 +465,7 @@ def test_dynamic_not_match(router) -> None: async def test_static_not_match(router) -> None: - router.add_static("/pre", os.path.dirname(aiohttp.__file__), name="name") + router.add_static("/pre", pathlib.Path(aiohttp.__file__).parent, name="name") resource = router["name"] ret = await resource.resolve(make_mocked_request("GET", "/another/path")) assert (None, set()) == ret @@ -503,17 +501,17 @@ def test_contains(router) -> None: def test_static_repr(router) -> None: - router.add_static("/get", os.path.dirname(aiohttp.__file__), name="name") + router.add_static("/get", pathlib.Path(aiohttp.__file__).parent, name="name") assert Matches(r" None: - route = router.add_static("/prefix", os.path.dirname(aiohttp.__file__)) + route = router.add_static("/prefix", pathlib.Path(aiohttp.__file__).parent) assert "/prefix" == route._prefix def test_static_remove_trailing_slash(router) -> None: - route = router.add_static("/prefix/", os.path.dirname(aiohttp.__file__)) + route = router.add_static("/prefix/", pathlib.Path(aiohttp.__file__).parent) assert "/prefix" == route._prefix @@ -778,7 +776,7 @@ def test_named_resources(router) -> None: route1 = router.add_route("GET", "/plain", make_handler(), name="route1") route2 = router.add_route("GET", "/variable/{name}", make_handler(), name="route2") route3 = router.add_static( - "/static", os.path.dirname(aiohttp.__file__), name="route3" + "/static", pathlib.Path(aiohttp.__file__).parent, name="route3" ) names = {route1.name, route2.name, route3.name} @@ -943,11 +941,11 @@ def test_resources_abc(router) -> None: def test_static_route_user_home(router) -> None: here = pathlib.Path(aiohttp.__file__).parent - home = pathlib.Path(os.path.expanduser("~")) - if not str(here).startswith(str(home)): # pragma: no cover + try: + static_dir = pathlib.Path("~") / here.relative_to(pathlib.Path.home()) + except ValueError: # pragma: no cover pytest.skip("aiohttp folder is not placed in user's HOME") - static_dir = "~/" + str(here.relative_to(home)) - route = router.add_static("/st", static_dir) + route = router.add_static("/st", str(static_dir)) assert here == route.get_info()["directory"] @@ -958,13 +956,13 @@ def test_static_route_points_to_file(router) -> None: async def test_404_for_static_resource(router) -> None: - resource = router.add_static("/st", os.path.dirname(aiohttp.__file__)) + resource = router.add_static("/st", pathlib.Path(aiohttp.__file__).parent) ret = await resource.resolve(make_mocked_request("GET", "/unknown/path")) assert (None, set()) == ret async def test_405_for_resource_adapter(router) -> None: - resource = router.add_static("/st", os.path.dirname(aiohttp.__file__)) + resource = router.add_static("/st", pathlib.Path(aiohttp.__file__).parent) ret = await resource.resolve(make_mocked_request("POST", "/st/abc.py")) assert (None, {"HEAD", "GET"}) == ret @@ -979,12 +977,12 @@ async def test_check_allowed_method_for_found_resource(router) -> None: def test_url_for_in_static_resource(router) -> None: - resource = router.add_static("/static", os.path.dirname(aiohttp.__file__)) + resource = router.add_static("/static", pathlib.Path(aiohttp.__file__).parent) assert URL("/static/file.txt") == resource.url_for(filename="file.txt") def test_url_for_in_static_resource_pathlib(router) -> None: - resource = router.add_static("/static", os.path.dirname(aiohttp.__file__)) + resource = router.add_static("/static", pathlib.Path(aiohttp.__file__).parent) assert URL("/static/file.txt") == resource.url_for( filename=pathlib.Path("file.txt") ) @@ -1163,7 +1161,7 @@ def test_frozen_app_on_subapp(app) -> None: def test_set_options_route(router) -> None: - resource = router.add_static("/static", os.path.dirname(aiohttp.__file__)) + resource = router.add_static("/static", pathlib.Path(aiohttp.__file__).parent) options = None for route in resource: if route.method == "OPTIONS": @@ -1233,7 +1231,7 @@ def test_dynamic_resource_canonical() -> None: def test_static_resource_canonical() -> None: prefix = "/prefix" - directory = str(os.path.dirname(aiohttp.__file__)) + directory = str(pathlib.Path(aiohttp.__file__).parent) canonical = prefix res = StaticResource(prefix=prefix, directory=directory) assert res.canonical == canonical diff --git a/tests/test_web_sendfile_functional.py b/tests/test_web_sendfile_functional.py index b044f29bc81..d67d67743ba 100644 --- a/tests/test_web_sendfile_functional.py +++ b/tests/test_web_sendfile_functional.py @@ -1,5 +1,4 @@ import asyncio -import os import pathlib import socket import zlib @@ -503,7 +502,7 @@ async def test_static_file_ssl( aiohttp_client, client_ssl_ctx, ) -> None: - dirname = os.path.dirname(__file__) + dirname = pathlib.Path(__file__).parent filename = "data.unknown_mime_type" app = web.Application() app.router.add_static("/static", dirname) @@ -524,9 +523,10 @@ async def test_static_file_ssl( async def test_static_file_directory_traversal_attack(aiohttp_client) -> None: - dirname = os.path.dirname(__file__) + dirname = pathlib.Path(__file__).parent relpath = "../README.rst" - assert os.path.isfile(os.path.join(dirname, relpath)) + full_path = dirname / relpath + assert full_path.is_file() app = web.Application() app.router.add_static("/static", dirname) @@ -541,7 +541,7 @@ async def test_static_file_directory_traversal_attack(aiohttp_client) -> None: assert 404 == resp.status await resp.release() - url_abspath = "/static/" + os.path.abspath(os.path.join(dirname, relpath)) + url_abspath = "/static/" + str(full_path.resolve()) resp = await client.get(url_abspath) assert 403 == resp.status await resp.release() @@ -550,36 +550,36 @@ async def test_static_file_directory_traversal_attack(aiohttp_client) -> None: def test_static_route_path_existence_check() -> None: - directory = os.path.dirname(__file__) + directory = pathlib.Path(__file__).parent web.StaticResource("/", directory) - nodirectory = os.path.join(directory, "nonexistent-uPNiOEAg5d") + nodirectory = directory / "nonexistent-uPNiOEAg5d" with pytest.raises(ValueError): web.StaticResource("/", nodirectory) async def test_static_file_huge(aiohttp_client, tmp_path) -> None: - filename = "huge_data.unknown_mime_type" + file_path = tmp_path / "huge_data.unknown_mime_type" # fill 20MB file - with (tmp_path / filename).open("wb") as f: + with file_path.open("wb") as f: for i in range(1024 * 20): f.write((chr(i % 64 + 0x20) * 1024).encode()) - file_st = os.stat(str(tmp_path / filename)) + file_st = file_path.stat() app = web.Application() app.router.add_static("/static", str(tmp_path)) client = await aiohttp_client(app) - resp = await client.get("/static/" + filename) + resp = await client.get("/static/" + file_path.name) assert 200 == resp.status ct = resp.headers["CONTENT-TYPE"] assert "application/octet-stream" == ct assert resp.headers.get("CONTENT-ENCODING") is None assert int(resp.headers.get("CONTENT-LENGTH")) == file_st.st_size - f = (tmp_path / filename).open("rb") + f = file_path.open("rb") off = 0 cnt = 0 while off < file_st.st_size: @@ -989,10 +989,10 @@ async def handler(request): async def test_static_file_huge_cancel(aiohttp_client, tmp_path) -> None: - filename = "huge_data.unknown_mime_type" + file_path = tmp_path / "huge_data.unknown_mime_type" # fill 100MB file - with (tmp_path / filename).open("wb") as f: + with file_path.open("wb") as f: for i in range(1024 * 20): f.write((chr(i % 64 + 0x20) * 1024).encode()) @@ -1005,7 +1005,7 @@ async def handler(request): tr = request.transport sock = tr.get_extra_info("socket") sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024) - ret = web.FileResponse(pathlib.Path(str(tmp_path / filename))) + ret = web.FileResponse(file_path) return ret app = web.Application() @@ -1030,10 +1030,10 @@ async def handler(request): async def test_static_file_huge_error(aiohttp_client, tmp_path) -> None: - filename = "huge_data.unknown_mime_type" + file_path = tmp_path / "huge_data.unknown_mime_type" # fill 20MB file - with (tmp_path / filename).open("wb") as f: + with file_path.open("wb") as f: f.seek(20 * 1024 * 1024) f.write(b"1") @@ -1042,7 +1042,7 @@ async def handler(request): tr = request.transport sock = tr.get_extra_info("socket") sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024) - ret = web.FileResponse(pathlib.Path(str(tmp_path / filename))) + ret = web.FileResponse(file_path) return ret app = web.Application() diff --git a/tests/test_web_urldispatcher.py b/tests/test_web_urldispatcher.py index aee3ecd5c24..76e533e473a 100644 --- a/tests/test_web_urldispatcher.py +++ b/tests/test_web_urldispatcher.py @@ -1,9 +1,6 @@ import asyncio import functools -import os import pathlib -import shutil -import tempfile from typing import Optional from unittest import mock from unittest.mock import MagicMock @@ -16,24 +13,6 @@ from aiohttp.web_urldispatcher import SystemRoute -@pytest.fixture(scope="function") -def tmp_dir_path(request): - """ - Give a path for a temporary directory - - The directory is destroyed at the end of the test. - """ - # Temporary directory. - tmp_dir = tempfile.mkdtemp() - - def teardown(): - # Delete the whole directory: - shutil.rmtree(tmp_dir) - - request.addfinalizer(teardown) - return tmp_dir - - @pytest.mark.parametrize( "show_index,status,prefix,data", [ @@ -63,7 +42,7 @@ def teardown(): ], ) async def test_access_root_of_static_handler( - tmp_dir_path: pathlib.Path, + tmp_path: pathlib.Path, aiohttp_client: AiohttpClient, show_index: bool, status: int, @@ -74,22 +53,22 @@ async def test_access_root_of_static_handler( # Try to access the root of static file server, and make # sure that correct HTTP statuses are returned depending if we directory # index should be shown or not. - # Put a file inside tmp_dir_path: - my_file_path = os.path.join(tmp_dir_path, "my_file") - with open(my_file_path, "w") as fw: - fw.write("hello") + # Put a file inside tmp_path: + my_file = tmp_path / "my_file" + my_dir = tmp_path / "my_dir" + my_dir.mkdir() + my_file_in_dir = my_dir / "my_file_in_dir" - my_dir_path = os.path.join(tmp_dir_path, "my_dir") - os.mkdir(my_dir_path) + with my_file.open("w") as fw: + fw.write("hello") - my_file_path = os.path.join(my_dir_path, "my_file_in_dir") - with open(my_file_path, "w") as fw: + with my_file_in_dir.open("w") as fw: fw.write("world") app = web.Application() # Register global static route: - app.router.add_static(prefix, tmp_dir_path, show_index=show_index) + app.router.add_static(prefix, str(tmp_path), show_index=show_index) client = await aiohttp_client(app) # Request the root of the static directory. @@ -103,25 +82,25 @@ async def test_access_root_of_static_handler( async def test_follow_symlink( - tmp_dir_path: pathlib.Path, aiohttp_client: AiohttpClient + tmp_path: pathlib.Path, aiohttp_client: AiohttpClient ) -> None: # Tests the access to a symlink, in static folder data = "hello world" - my_dir_path = os.path.join(tmp_dir_path, "my_dir") - os.mkdir(my_dir_path) + my_dir_path = tmp_path / "my_dir" + my_dir_path.mkdir() - my_file_path = os.path.join(my_dir_path, "my_file_in_dir") - with open(my_file_path, "w") as fw: + my_file_path = my_dir_path / "my_file_in_dir" + with my_file_path.open("w") as fw: fw.write(data) - my_symlink_path = os.path.join(tmp_dir_path, "my_symlink") - os.symlink(my_dir_path, my_symlink_path) + my_symlink_path = tmp_path / "my_symlink" + pathlib.Path(str(my_symlink_path)).symlink_to(str(my_dir_path), True) app = web.Application() # Register global static route: - app.router.add_static("/", tmp_dir_path, follow_symlinks=True) + app.router.add_static("/", str(tmp_path), follow_symlinks=True) client = await aiohttp_client(app) # Request the root of the static directory. @@ -229,7 +208,7 @@ async def test_follow_symlink_directory_traversal_after_normalization( ], ) async def test_access_to_the_file_with_spaces( - tmp_dir_path: pathlib.Path, + tmp_path: pathlib.Path, aiohttp_client: AiohttpClient, dir_name: str, filename: str, @@ -237,21 +216,19 @@ async def test_access_to_the_file_with_spaces( ) -> None: # Checks operation of static files with spaces - my_dir_path = os.path.join(tmp_dir_path, dir_name) + my_dir_path = tmp_path / dir_name + if my_dir_path != tmp_path: + my_dir_path.mkdir() - if dir_name: - os.mkdir(my_dir_path) - - my_file_path = os.path.join(my_dir_path, filename) - - with open(my_file_path, "w") as fw: + my_file_path = my_dir_path / filename + with my_file_path.open("w") as fw: fw.write(data) app = web.Application() - url = os.path.join("/", dir_name, filename) + url = "/" + str(pathlib.Path(dir_name, filename)) - app.router.add_static("/", tmp_dir_path) + app.router.add_static("/", str(tmp_path)) client = await aiohttp_client(app) r = await client.get(url) @@ -261,7 +238,7 @@ async def test_access_to_the_file_with_spaces( async def test_access_non_existing_resource( - tmp_dir_path: pathlib.Path, aiohttp_client: AiohttpClient + tmp_path: pathlib.Path, aiohttp_client: AiohttpClient ) -> None: # Tests accessing non-existing resource # Try to access a non-exiting resource and make sure that 404 HTTP status @@ -269,7 +246,7 @@ async def test_access_non_existing_resource( app = web.Application() # Register global static route: - app.router.add_static("/", tmp_dir_path, show_index=True) + app.router.add_static("/", str(tmp_path), show_index=True) client = await aiohttp_client(app) # Request the root of the static directory. @@ -323,13 +300,13 @@ def sync_handler(request): async def test_unauthorized_folder_access( - tmp_dir_path: pathlib.Path, aiohttp_client: AiohttpClient + tmp_path: pathlib.Path, aiohttp_client: AiohttpClient ) -> None: # Tests the unauthorized access to a folder of static file server. # Try to list a folder content of static file server when server does not # have permissions to do so for the folder. - my_dir_path = os.path.join(tmp_dir_path, "my_dir") - os.mkdir(my_dir_path) + my_dir = tmp_path / "my_dir" + my_dir.mkdir() app = web.Application() @@ -341,34 +318,34 @@ async def test_unauthorized_folder_access( path_constructor.return_value = path # Register global static route: - app.router.add_static("/", tmp_dir_path, show_index=True) + app.router.add_static("/", str(tmp_path), show_index=True) client = await aiohttp_client(app) # Request the root of the static directory. - r = await client.get("/my_dir") + r = await client.get("/" + my_dir.name) assert r.status == 403 async def test_access_symlink_loop( - tmp_dir_path: pathlib.Path, aiohttp_client: AiohttpClient + tmp_path: pathlib.Path, aiohttp_client: AiohttpClient ) -> None: # Tests the access to a looped symlink, which could not be resolved. - my_dir_path = os.path.join(tmp_dir_path, "my_symlink") - os.symlink(my_dir_path, my_dir_path) + my_dir_path = tmp_path / "my_symlink" + pathlib.Path(str(my_dir_path)).symlink_to(str(my_dir_path), True) app = web.Application() # Register global static route: - app.router.add_static("/", tmp_dir_path, show_index=True) + app.router.add_static("/", str(tmp_path), show_index=True) client = await aiohttp_client(app) # Request the root of the static directory. - r = await client.get("/my_symlink") + r = await client.get("/" + my_dir_path.name) assert r.status == 404 async def test_access_special_resource( - tmp_dir_path: pathlib.Path, aiohttp_client: AiohttpClient + tmp_path: pathlib.Path, aiohttp_client: AiohttpClient ) -> None: # Tests the access to a resource that is neither a file nor a directory. # Checks that if a special resource is accessed (f.e. named pipe or UNIX @@ -388,7 +365,7 @@ async def test_access_special_resource( path_constructor.return_value = path # Register global static route: - app.router.add_static("/", tmp_dir_path, show_index=True) + app.router.add_static("/", str(tmp_path), show_index=True) client = await aiohttp_client(app) # Request the root of the static directory. @@ -623,20 +600,20 @@ async def test_static_absolute_url( # /static/\\machine_name\c$ or /static/D:\path # where the static dir is totally different app = web.Application() - fname = tmp_path / "file.txt" - fname.write_text("sample text", "ascii") + file_path = tmp_path / "file.txt" + file_path.write_text("sample text", "ascii") here = pathlib.Path(__file__).parent app.router.add_static("/static", here) client = await aiohttp_client(app) - resp = await client.get("/static/" + str(fname)) + resp = await client.get("/static/" + str(file_path.resolve())) assert resp.status == 403 async def test_for_issue_5250( - aiohttp_client: AiohttpClient, tmp_dir_path: pathlib.Path + aiohttp_client: AiohttpClient, tmp_path: pathlib.Path ) -> None: app = web.Application() - app.router.add_static("/foo", tmp_dir_path) + app.router.add_static("/foo", tmp_path) async def get_foobar(request: web.Request) -> web.Response: return web.Response(body="success!") diff --git a/tools/check_changes.py b/tools/check_changes.py index 118d1182b9a..6cc4d050cd8 100755 --- a/tools/check_changes.py +++ b/tools/check_changes.py @@ -22,7 +22,7 @@ def get_root(script_path): - folder = script_path.absolute().parent + folder = script_path.resolve().parent while not (folder / ".git").exists(): folder = folder.parent if folder == folder.anchor: