• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""Reusable functions and classes for different types of integration tests.
2
3For example ``Archive`` can be used to check the contents of distribution built
4with setuptools, and ``run`` will always try to be as verbose as possible to
5facilitate debugging.
6"""
7import os
8import subprocess
9import tarfile
10from zipfile import ZipFile
11from pathlib import Path
12
13
14def run(cmd, env=None):
15    r = subprocess.run(
16        cmd,
17        stdout=subprocess.PIPE,
18        stderr=subprocess.PIPE,
19        universal_newlines=True,
20        env={**os.environ, **(env or {})}
21        # ^-- allow overwriting instead of discarding the current env
22    )
23
24    out = r.stdout + "\n" + r.stderr
25    # pytest omits stdout/err by default, if the test fails they help debugging
26    print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")
27    print(f"Command: {cmd}\nreturn code: {r.returncode}\n\n{out}")
28
29    if r.returncode == 0:
30        return out
31    raise subprocess.CalledProcessError(r.returncode, cmd, r.stdout, r.stderr)
32
33
34class Archive:
35    """Compatibility layer for ZipFile/Info and TarFile/Info"""
36    def __init__(self, filename):
37        self._filename = filename
38        if filename.endswith("tar.gz"):
39            self._obj = tarfile.open(filename, "r:gz")
40        elif filename.endswith("zip"):
41            self._obj = ZipFile(filename)
42        else:
43            raise ValueError(f"{filename} doesn't seem to be a zip or tar.gz")
44
45    def __iter__(self):
46        if hasattr(self._obj, "infolist"):
47            return iter(self._obj.infolist())
48        return iter(self._obj)
49
50    def get_name(self, zip_or_tar_info):
51        if hasattr(zip_or_tar_info, "filename"):
52            return zip_or_tar_info.filename
53        return zip_or_tar_info.name
54
55    def get_content(self, zip_or_tar_info):
56        if hasattr(self._obj, "extractfile"):
57            content = self._obj.extractfile(zip_or_tar_info)
58            if content is None:
59                msg = f"Invalid {zip_or_tar_info.name} in {self._filename}"
60                raise ValueError(msg)
61            return str(content.read(), "utf-8")
62        return str(self._obj.read(zip_or_tar_info), "utf-8")
63
64
65def get_sdist_members(sdist_path):
66    with tarfile.open(sdist_path, "r:gz") as tar:
67        files = [Path(f) for f in tar.getnames()]
68    # remove root folder
69    relative_files = ("/".join(f.parts[1:]) for f in files)
70    return {f for f in relative_files if f}
71
72
73def get_wheel_members(wheel_path):
74    with ZipFile(wheel_path) as zipfile:
75        return set(zipfile.namelist())
76