Skip to content

Commit

Permalink
add mock server to test that each dist is downloaded exactly once
Browse files Browse the repository at this point in the history
  • Loading branch information
cosmicexplorer committed Aug 2, 2023
1 parent 8a91de8 commit 0144ac3
Show file tree
Hide file tree
Showing 2 changed files with 94 additions and 0 deletions.
43 changes: 43 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -992,3 +992,46 @@ def html_index_for_packages(
f.write(pkg_index_content)

return html_dir


class OneTimeDownloadHandler(http.server.SimpleHTTPRequestHandler):
"""Serve files from the current directory, but error if a file is downloaded more
than once."""

_seen_paths: ClassVar[Set[str]] = set()

def do_GET(self) -> None:
if self.path in self._seen_paths:
self.send_error(
http.HTTPStatus.NOT_FOUND,
f"File {self.path} not available more than once!",
)
return
super().do_GET()
if not (self.path.endswith("/") or self.path.endswith(".metadata")):
self._seen_paths.add(self.path)


@pytest.fixture(scope="function")
def html_index_with_onetime_server(
html_index_for_packages: Path,
) -> Iterator[http.server.ThreadingHTTPServer]:
class InDirectoryServer(http.server.ThreadingHTTPServer):
def finish_request(self, request: Any, client_address: Any) -> None:
# raise Exception(request)
self.RequestHandlerClass(
request, client_address, self, directory=str(html_index_for_packages) # type: ignore[call-arg] # noqa: E501
)

class Handler(OneTimeDownloadHandler):
_seen_paths: ClassVar[Set[str]] = set()

with InDirectoryServer(("", 8000), Handler) as httpd:
server_thread = threading.Thread(target=httpd.serve_forever)
server_thread.start()

try:
yield httpd
finally:
httpd.shutdown()
server_thread.join()
51 changes: 51 additions & 0 deletions tests/functional/test_download.py
Original file line number Diff line number Diff line change
Expand Up @@ -1324,6 +1324,57 @@ def test_download_metadata(
assert sorted(os.listdir(download_dir)) == expected_outputs


@pytest.mark.parametrize(
"requirement_to_download, expected_outputs, doubled_path",
[
(
"simple2==1.0",
["simple-1.0.tar.gz", "simple2-1.0.tar.gz"],
"/simple2/simple2-1.0.tar.gz",
),
("simple==2.0", ["simple-2.0.tar.gz"], "/simple/simple-2.0.tar.gz"),
(
"colander",
["colander-0.9.9-py2.py3-none-any.whl", "translationstring-1.1.tar.gz"],
"/colander/colander-0.9.9-py2.py3-none-any.whl",
),
(
"compilewheel",
[
"compilewheel-1.0-py2.py3-none-any.whl",
"simple-1.0.tar.gz",
],
"/compilewheel/compilewheel-1.0-py2.py3-none-any.whl",
),
],
)
def test_download_metadata_server(
download_server_html_index: Callable[..., Tuple[TestPipResult, Path]],
requirement_to_download: str,
expected_outputs: List[str],
doubled_path: str,
) -> None:
"""Verify that if a data-dist-info-metadata attribute is present, then it is used
instead of the actual dist's METADATA.
Additionally, verify that each dist is downloaded exactly once using a mock server.
This is a regression test for issue https://github.com/pypa/pip/issues/11847.
"""
_, download_dir = download_server_html_index(
[requirement_to_download, "--no-cache-dir"],
)
assert sorted(os.listdir(download_dir)) == expected_outputs
shutil.rmtree(download_dir)
result, _ = download_server_html_index(
[requirement_to_download, "--no-cache-dir"],
allow_error=True,
)
assert result.returncode != 0
expected_msg = f"File {doubled_path} not available more than once!"
assert expected_msg in result.stderr


@pytest.mark.parametrize(
"requirement_to_download, real_hash",
[
Expand Down

0 comments on commit 0144ac3

Please sign in to comment.