codekingpro/portable-devtools

1"""Reusable functions and classes for different types of integration tests.2 3For example ``Archive`` can be used to check the contents of distribution built4with setuptools, and ``run`` will always try to be as verbose as possible to5facilitate debugging.6"""7 8from __future__ import annotations9 10import os11import subprocess12import tarfile13from collections.abc import Iterator14from pathlib import Path15from zipfile import ZipFile, ZipInfo16 17 18def run(cmd, env=None):19    r = subprocess.run(20        cmd,21        capture_output=True,22        text=True,23        encoding="utf-8",24        env={**os.environ, **(env or {})},25        # ^-- allow overwriting instead of discarding the current env26    )27 28    out = r.stdout + "\n" + r.stderr29    # pytest omits stdout/err by default, if the test fails they help debugging30    print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")31    print(f"Command: {cmd}\nreturn code: {r.returncode}\n\n{out}")32 33    if r.returncode == 0:34        return out35    raise subprocess.CalledProcessError(r.returncode, cmd, r.stdout, r.stderr)36 37 38class Archive:39    """Compatibility layer for ZipFile/Info and TarFile/Info"""40 41    def __init__(self, filename) -> None:42        self._filename = filename43        if filename.endswith("tar.gz"):44            self._obj: tarfile.TarFile | ZipFile = tarfile.open(filename, "r:gz")45        elif filename.endswith("zip"):46            self._obj = ZipFile(filename)47        else:48            raise ValueError(f"{filename} doesn't seem to be a zip or tar.gz")49 50    def __iter__(self) -> Iterator[ZipInfo] | Iterator[tarfile.TarInfo]:51        if hasattr(self._obj, "infolist"):52            return iter(self._obj.infolist())53        return iter(self._obj)54 55    def get_name(self, zip_or_tar_info):56        if hasattr(zip_or_tar_info, "filename"):57            return zip_or_tar_info.filename58        return zip_or_tar_info.name59 60    def get_content(self, zip_or_tar_info):61        if hasattr(self._obj, "extractfile"):62            content = self._obj.extractfile(zip_or_tar_info)63            if content is None:64                msg = f"Invalid {zip_or_tar_info.name} in {self._filename}"65                raise ValueError(msg)66            return str(content.read(), "utf-8")67        return str(self._obj.read(zip_or_tar_info), "utf-8")68 69 70def get_sdist_members(sdist_path):71    with tarfile.open(sdist_path, "r:gz") as tar:72        files = [Path(f) for f in tar.getnames()]73    # remove root folder74    relative_files = ("/".join(f.parts[1:]) for f in files)75    return {f for f in relative_files if f}76 77 78def get_wheel_members(wheel_path):79    with ZipFile(wheel_path) as zipfile:80        return set(zipfile.namelist())81