forked from github.com/pypiserver
377 lines
10 KiB
Python
377 lines
10 KiB
Python
#! /usr/bin/env py.test
|
|
"""
|
|
Checks an actual pypi-server against various clients.
|
|
|
|
The tests below are using 3 ways to startup pypi-servers:
|
|
|
|
- "open": a per-module server instance without any authed operations,
|
|
serving a single `wheel` package, on a fixed port.
|
|
- "open": a per-module server instance with authed 'download/upload'
|
|
operations, serving a single `wheel` package, on a fixed port.
|
|
- "new_server": starting a new server with any configurations on each test.
|
|
|
|
"""
|
|
import contextlib
|
|
import itertools
|
|
import os
|
|
import shutil
|
|
import socket
|
|
import sys
|
|
import time
|
|
import typing as t
|
|
from collections import namedtuple
|
|
from pathlib import Path
|
|
from shlex import split
|
|
from subprocess import Popen
|
|
from urllib.error import URLError
|
|
from urllib.request import urlopen
|
|
|
|
import pytest
|
|
|
|
# ######################################################################
|
|
# Fixtures & Helper Functions
|
|
# ######################################################################
|
|
|
|
|
|
CURRENT_PATH = Path(__file__).parent
|
|
ports = itertools.count(10000)
|
|
Srv = namedtuple("Srv", ("port", "root"))
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def run_server(root, authed=False, other_cli=""):
|
|
"""Run a server, optionally with partial auth enabled."""
|
|
htpasswd = (
|
|
CURRENT_PATH.joinpath("../fixtures/htpasswd.a.a").expanduser().resolve()
|
|
)
|
|
pswd_opt_choices = {
|
|
True: f"-P {htpasswd} -a update,download",
|
|
False: "-P. -a.",
|
|
"partial": f"-P {htpasswd} -a update",
|
|
}
|
|
pswd_opts = pswd_opt_choices[authed]
|
|
|
|
port = next(ports)
|
|
cmd = (
|
|
f"{sys.executable} -m pypiserver.__main__ "
|
|
f"run -vvv --overwrite -i 127.0.0.1 "
|
|
f"-p {port} {pswd_opts} {other_cli} {root}"
|
|
)
|
|
proc = Popen(cmd.split(), bufsize=2**16)
|
|
srv = Srv(port, root)
|
|
try:
|
|
wait_until_ready(srv)
|
|
assert proc.poll() is None
|
|
yield srv
|
|
finally:
|
|
print(f"Killing {srv}")
|
|
_kill_proc(proc)
|
|
|
|
|
|
def wait_until_ready(srv: Srv, n_tries=10):
|
|
for _ in range(n_tries):
|
|
if is_ready(srv):
|
|
return True
|
|
time.sleep(0.5)
|
|
raise TimeoutError
|
|
|
|
|
|
def is_ready(srv: Srv):
|
|
try:
|
|
return urlopen(build_url(srv.port), timeout=0.5).getcode() in (
|
|
200,
|
|
401,
|
|
)
|
|
except (URLError, socket.timeout):
|
|
return False
|
|
|
|
|
|
def _kill_proc(proc):
|
|
proc.terminate()
|
|
try:
|
|
proc.wait(timeout=1)
|
|
finally:
|
|
proc.kill()
|
|
|
|
|
|
def build_url(port: t.Union[int, str], user: str = "", pswd: str = "") -> str:
|
|
auth = f"{user}:{pswd}@" if user or pswd else ""
|
|
return f"http://{auth}localhost:{port}"
|
|
|
|
|
|
def run_setup_py(path: Path, arguments: str):
|
|
return os.system(f"{sys.executable} {path / 'setup.py'} {arguments}")
|
|
|
|
|
|
# A test-distribution to check if
|
|
# bottle supports uploading 100's of packages,
|
|
# see: https://github.com/pypiserver/pypiserver/issues/82
|
|
#
|
|
# Has been run once `pip wheel .`, just to generate:
|
|
# ./wheelhouse/centodeps-0.0.0-cp34-none-win_amd64.whl
|
|
#
|
|
SETUP_PY = """\
|
|
from setuptools import setup
|
|
|
|
setup(
|
|
name="centodeps",
|
|
install_requires=["a==1.0"] * 200,
|
|
options={
|
|
"bdist_wheel": {"universal": True},
|
|
},
|
|
packages=[],
|
|
)
|
|
"""
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def project(tmp_path_factory):
|
|
projdir = tmp_path_factory.mktemp("project") / "centodeps"
|
|
projdir.mkdir(parents=True, exist_ok=True)
|
|
projdir.joinpath("setup.py").write_text(SETUP_PY)
|
|
return projdir
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def server_root(tmp_path_factory):
|
|
return tmp_path_factory.mktemp("root")
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def wheel_file(project, tmp_path_factory):
|
|
distdir = tmp_path_factory.mktemp("dist")
|
|
assert run_setup_py(project, f"bdist_wheel -d {distdir}") == 0
|
|
return list(distdir.glob("centodeps*.whl"))[0]
|
|
|
|
|
|
@pytest.fixture()
|
|
def hosted_wheel_file(wheel_file, server_root):
|
|
dst = server_root / wheel_file.name
|
|
shutil.copy(wheel_file, dst)
|
|
yield dst
|
|
if dst.is_file():
|
|
dst.unlink()
|
|
|
|
|
|
def clear_directory(root: Path):
|
|
for path in root.iterdir():
|
|
if path.is_file():
|
|
path.unlink()
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def _open_server(server_root):
|
|
with run_server(server_root, authed=False) as srv:
|
|
yield srv
|
|
|
|
|
|
@pytest.fixture
|
|
def open_server(_open_server: Srv):
|
|
yield _open_server
|
|
clear_directory(_open_server.root)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def _authed_server(server_root):
|
|
with run_server(server_root, authed=True) as srv:
|
|
yield srv
|
|
|
|
|
|
@pytest.fixture
|
|
def authed_server(_authed_server):
|
|
yield _authed_server
|
|
clear_directory(_authed_server.root)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def _partial_auth_server(server_root):
|
|
with run_server(server_root, authed="partial") as srv:
|
|
yield srv
|
|
|
|
|
|
@pytest.fixture
|
|
def partial_authed_server(_partial_auth_server):
|
|
yield _partial_auth_server
|
|
clear_directory(_partial_auth_server.root)
|
|
|
|
|
|
@pytest.fixture
|
|
def empty_packdir(tmp_path_factory):
|
|
return tmp_path_factory.mktemp("dists")
|
|
|
|
|
|
def pip_download(
|
|
cmd: str,
|
|
port: t.Union[int, str],
|
|
install_dir: str,
|
|
user: str = None,
|
|
pswd: str = None,
|
|
) -> int:
|
|
url = build_url(port, user, pswd)
|
|
return _run_pip(f"-vv download -d {install_dir} -i {url} {cmd}")
|
|
|
|
|
|
def _run_pip(cmd: str) -> int:
|
|
ncmd = (
|
|
"pip --no-cache-dir --disable-pip-version-check "
|
|
f"--retries 0 --timeout 5 --no-input {cmd}"
|
|
)
|
|
print(f"PIP: {ncmd}")
|
|
proc = Popen(split(ncmd))
|
|
proc.communicate()
|
|
return proc.returncode
|
|
|
|
|
|
@pytest.fixture
|
|
def pipdir(tmp_path_factory):
|
|
return tmp_path_factory.mktemp("pip")
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def pypirc_file(repo, username="''", password="''"):
|
|
pypirc_path = Path.home() / ".pypirc"
|
|
old_pypirc = pypirc_path.read_text() if pypirc_path.is_file() else None
|
|
pypirc_path.write_text(
|
|
"\n".join(
|
|
(
|
|
"[distutils]",
|
|
"index-servers: test",
|
|
"",
|
|
"[test]",
|
|
f"repository: {repo}",
|
|
f"username: {username}",
|
|
f"password: {password}",
|
|
)
|
|
)
|
|
)
|
|
try:
|
|
yield pypirc_path
|
|
finally:
|
|
if old_pypirc:
|
|
pypirc_path.write_text(old_pypirc)
|
|
else:
|
|
pypirc_path.unlink()
|
|
|
|
|
|
@pytest.fixture
|
|
def open_pypirc(open_server):
|
|
with pypirc_file(repo=build_url(open_server.port)) as path:
|
|
yield path
|
|
|
|
|
|
@pytest.fixture
|
|
def authed_pypirc(authed_server):
|
|
username, password = "a", "a"
|
|
with pypirc_file(
|
|
repo=build_url(authed_server.port),
|
|
username=username,
|
|
password=password,
|
|
) as path:
|
|
yield path
|
|
|
|
|
|
def run_twine(command: str, package: str, conf: str) -> None:
|
|
proc = Popen(
|
|
split(
|
|
f"twine {command} --repository test --config-file {conf} {package}"
|
|
)
|
|
)
|
|
proc.communicate()
|
|
assert not proc.returncode, f"Twine {command} failed. See stdout/err"
|
|
|
|
|
|
# ######################################################################
|
|
# Tests
|
|
# ######################################################################
|
|
|
|
all_servers = [
|
|
("open_server", "open_pypirc"),
|
|
("authed_server", "authed_pypirc"),
|
|
("partial_authed_server", "authed_pypirc"),
|
|
]
|
|
|
|
|
|
def test_pip_install_package_not_found(open_server, pipdir):
|
|
assert pip_download("centodeps", open_server.port, pipdir) != 0
|
|
assert not list(pipdir.iterdir())
|
|
|
|
|
|
def test_pip_install_open_succeeds(open_server, hosted_wheel_file, pipdir):
|
|
assert pip_download("centodeps", open_server.port, pipdir) == 0
|
|
assert pipdir.joinpath(hosted_wheel_file.name).is_file()
|
|
|
|
|
|
@pytest.mark.usefixtures("wheel_file")
|
|
def test_pip_install_authed_fails(authed_server, pipdir):
|
|
assert pip_download("centodeps", authed_server.port, pipdir) != 0
|
|
assert not list(pipdir.iterdir())
|
|
|
|
|
|
def test_pip_install_authed_succeeds(authed_server, hosted_wheel_file, pipdir):
|
|
assert (
|
|
pip_download(
|
|
"centodeps", authed_server.port, pipdir, user="a", pswd="a"
|
|
)
|
|
== 0
|
|
)
|
|
assert pipdir.joinpath(hosted_wheel_file.name).is_file()
|
|
|
|
|
|
@pytest.mark.parametrize("pkg_frmt", ["bdist", "bdist_wheel"])
|
|
@pytest.mark.parametrize(["server_fixture", "pypirc_fixture"], all_servers)
|
|
def test_setuptools_upload(
|
|
server_fixture, pypirc_fixture, project, pkg_frmt, server_root, request
|
|
):
|
|
request.getfixturevalue(server_fixture)
|
|
request.getfixturevalue(pypirc_fixture)
|
|
|
|
assert len(list(server_root.iterdir())) == 0
|
|
|
|
for i in range(5):
|
|
print(f"++Attempt #{i}")
|
|
assert run_setup_py(project, f"-vvv {pkg_frmt} upload -r test") == 0
|
|
assert len(list(server_root.iterdir())) == 1
|
|
|
|
|
|
def test_partial_authed_open_download(partial_authed_server):
|
|
"""Validate that partial auth still allows downloads."""
|
|
url = build_url(partial_authed_server.port) + "/simple"
|
|
resp = urlopen(url)
|
|
assert resp.getcode() == 200
|
|
|
|
|
|
@pytest.mark.parametrize("hash_algo", ("md5", "sha256", "sha512"))
|
|
@pytest.mark.usefixtures("hosted_wheel_file")
|
|
def test_hash_algos(server_root, pipdir, hash_algo):
|
|
"""Test twine upload with no authentication"""
|
|
with run_server(
|
|
server_root, other_cli="--hash-algo {}".format(hash_algo)
|
|
) as srv:
|
|
assert pip_download("centodeps", srv.port, pipdir) == 0
|
|
|
|
|
|
@pytest.mark.parametrize(["server_fixture", "pypirc_fixture"], all_servers)
|
|
def test_twine_upload(
|
|
server_fixture, pypirc_fixture, server_root, wheel_file, request
|
|
):
|
|
"""Test twine upload with no authentication"""
|
|
assert len(list(server_root.iterdir())) == 0
|
|
request.getfixturevalue(server_fixture)
|
|
pypirc = request.getfixturevalue(pypirc_fixture)
|
|
|
|
run_twine("upload", wheel_file, conf=pypirc)
|
|
|
|
assert len(list(server_root.iterdir())) == 1
|
|
assert server_root.joinpath(wheel_file.name).is_file(), (
|
|
wheel_file.name,
|
|
list(server_root.iterdir()),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(["server_fixture", "pypirc_fixture"], all_servers)
|
|
def test_twine_register(server_fixture, pypirc_fixture, wheel_file, request):
|
|
"""Test unauthenticated twine registration"""
|
|
request.getfixturevalue(server_fixture)
|
|
pypirc = request.getfixturevalue(pypirc_fixture)
|
|
run_twine("register", wheel_file, conf=pypirc)
|