def read_pkg_info_bytes(bytestr: Union[bytes, str]) -> Message:
    """Parse a PKG-INFO or METADATA data string."""
    if isinstance(bytestr, bytes):
        bytestr = bytestr.decode("utf-8")
    return Parser().parsestr(bytestr) class InTemporaryDirectory(TemporaryDirectory):
    """Create, return, and change directory to a temporary directory

    Examples
    --------
    >>> import os
    >>> my_cwd = os.getcwd()
    >>> with InTemporaryDirectory() as tmpdir:
    ...     _ = open('test.txt', 'wt').write('some text')
    ...     assert os.path.isfile('test.txt')
    ...     assert os.path.isfile(os.path.join(tmpdir, 'test.txt'))
    >>> os.path.exists(tmpdir)
    False
    >>> os.getcwd() == my_cwd
    True
    """

    def __enter__(self):
        self._pwd = os.getcwd()
        os.chdir(self.name)
        return super(InTemporaryDirectory, self).__enter__()

    def __exit__(self, exc, value, tb):
        os.chdir(self._pwd)
        return super(InTemporaryDirectory, self).__exit__(exc, value, tb) T = TypeVar("T")
IN_RE = re.compile(
    r"(.*) \(compatibility version (\d+\.\d+\.\d+), "
    r"current version (\d+\.\d+\.\d+)(?:, \w+)?\)"
)
_OTOOL_ARCHITECTURE_RE = re.compile(
    r"^(?P<name>.*?)(?: \(architecture (?P<architecture>\w+)\))?:$"
)
RE_PERM_DEN = re.compile(r"Permission denied[.) ]*$")
BAD_OBJECT_TESTS = [
    # otool version cctools-862
    lambda s: "is not an object file" in s,
    # cctools-862 (.ico)
    lambda s: "The end of the file was unexpectedly encountered" in s,
    # cctools-895
    lambda s: "The file was not recognized as a valid object file" in s,
    # 895 binary file
    lambda s: "Invalid data was encountered while parsing the file" in s,
    # cctools-900
    lambda s: "Object is not a Mach-O file type" in s,
    # cctools-949
    lambda s: "object is not a Mach-O file type" in s,
    # File may not have read permissions
    lambda s: RE_PERM_DEN.search(s) is not None,
]
_LINE0_RE = re.compile(r"^(?: \(architecture .*\))?:(?P<further_report>.*)")
RPATH_RE = re.compile(r"path (?P<rpath>.*) \(offset \d+\)")
class InstallNameError(Exception):
def back_tick(
    cmd: Union[str, Sequence[str]],
    ret_err: bool = False,
    as_str: bool = True,
    raise_err: Optional[bool] = None,
) -> Any:
def unique_by_index(sequence):
def chmod_perms(fname):
def ensure_permissions(mode_flags=stat.S_IWUSR):
    def decorator(f):
        def modify(filename, *args, **kwargs):
def parse_install_name(line: str) -> Tuple[str, str, str]:
def _parse_otool_listing(stdout: str) -> Dict[str, List[str]]:
def _check_ignore_archs(input: Dict[str, T]) -> T:
def _parse_otool_install_names(
    stdout: str,
) -> Dict[str, List[Tuple[str, str, str]]]:
def _line0_says_object(stdout_stderr: str, filename: str) -> bool:
def get_install_names(filename: str) -> Tuple[str, ...]:
def get_install_id(filename: str) -> Optional[str]:
def _get_install_ids(filename: str) -> Dict[str, str]:
def set_install_name(
    filename: str, oldname: str, newname: str, ad_hoc_sign: bool = True
) -> None:
def set_install_id(filename: str, install_id: str, ad_hoc_sign: bool = True):
def _parse_otool_rpaths(stdout: str) -> Dict[str, List[str]]:
def get_rpaths(filename: str) -> Tuple[str, ...]:
def get_environment_variable_paths():
def add_rpath(filename: str, newpath: str, ad_hoc_sign: bool = True) -> None:
def zip2dir(zip_fname: str, out_dir: str) -> None:
def dir2zip(in_dir, zip_fname):
def find_package_dirs(root_path: str) -> Set[str]:
def cmp_contents(filename1, filename2):
def get_archs(libname: str) -> FrozenSet[str]:
def lipo_fuse(
    in_fname1: str, in_fname2: str, out_fname: str, ad_hoc_sign: bool = True
) -> str:
def replace_signature(filename: str, identity: str) -> None:
def validate_signature(filename: str) -> None: class InWheel(InTemporaryDirectory):
    """Context manager for doing things inside wheels

    On entering, you'll find yourself in the root tree of the wheel.  If you've
    asked for an output wheel, then on exit we'll rewrite the wheel record and
    pack stuff up for you.
    """

    def __init__(self, in_wheel, out_wheel=None, ret_self=False):
        """Initialize in-wheel context manager

        Parameters
        ----------
        in_wheel : str
            filename of wheel to unpack and work inside
        out_wheel : None or str:
            filename of wheel to write after exiting.  If None, don't write and
            discard
        ret_self : bool, optional
            If True, return ``self`` from ``__enter__``, otherwise return the
            directory path.
        """
        self.in_wheel = abspath(in_wheel)
        self.out_wheel = None if out_wheel is None else abspath(out_wheel)
        super(InWheel, self).__init__()

    def __enter__(self):
        zip2dir(self.in_wheel, self.name)
        return super(InWheel, self).__enter__()

    def __exit__(self, exc, value, tb):
        if self.out_wheel is not None:
            rewrite_record(self.name)
            dir2zip(self.name, self.out_wheel)
        return super(InWheel, self).__exit__(exc, value, tb) class InWheelCtx(InWheel):
    """Context manager for doing things inside wheels

    On entering, you'll find yourself in the root tree of the wheel.  If you've
    asked for an output wheel, then on exit we'll rewrite the wheel record and
    pack stuff up for you.

    The context manager returns itself from the __enter__ method, so you can
    set things like ``out_wheel``.  This is useful when processing in the wheel
    will dicate what the output wheel name is, or whether you want to save at
    all.

    The current path of the wheel contents is set in the attribute
    ``wheel_path``.
    """

    def __init__(self, in_wheel, out_wheel=None):
        """Init in-wheel context manager returning self from enter

        Parameters
        ----------
        in_wheel : str
            filename of wheel to unpack and work inside
        out_wheel : None or str:
            filename of wheel to write after exiting.  If None, don't write and
            discard
        """
        super(InWheelCtx, self).__init__(in_wheel, out_wheel)
        self.wheel_path = None

    def __enter__(self):
        self.wheel_path = super(InWheelCtx, self).__enter__()
        return self class WheelToolsError(Exception):
    pass def _get_wheelinfo_name(wheelfile: WheelFile):
    return wheelfile.dist_info_path + "/WHEEL" def add_platforms(in_wheel, platforms, out_path=None, clobber=False):
    """Add platform tags `platforms` to `in_wheel` filename and WHEEL tags

    Add any platform tags in `platforms` that are missing from `in_wheel`
    filename.

    Add any platform tags in `platforms` that are missing from `in_wheel`
    ``WHEEL`` file.

    Parameters
    ----------
    in_wheel : str
        Filename of wheel to which to add platform tags
    platforms : iterable
        platform tags to add to wheel filename and WHEEL tags - e.g.
        ``('macosx_10_9_intel', 'macosx_10_9_x86_64')
    out_path : None or str, optional
        Directory to which to write new wheel.  Default is directory containing
        `in_wheel`
    clobber : bool, optional
        If True, overwrite existing output filename, otherwise raise error

    Returns
    -------
    out_wheel : None or str
        Absolute path of wheel file written, or None if no wheel file written.
    """
    in_wheel = abspath(in_wheel)
    out_path = dirname(in_wheel) if out_path is None else abspath(out_path)
    wf = WheelFile(in_wheel)
    info_fname = _get_wheelinfo_name(wf)
    # Check what tags we have
    in_fname_tags = wf.parsed_filename.groupdict()["plat"].split(".")
    extra_fname_tags = [tag for tag in platforms if tag not in in_fname_tags]
    in_wheel_base, ext = splitext(basename(in_wheel))
    out_wheel_base = ".".join([in_wheel_base] + list(extra_fname_tags))
    out_wheel = pjoin(out_path, out_wheel_base + ext)
    if exists(out_wheel) and not clobber:
        raise WheelToolsError(
            "Not overwriting {0}; set clobber=True to overwrite".format(
                out_wheel
            )
        )
    with InWheelCtx(in_wheel) as ctx:
        info = read_pkg_info(info_fname)
        if info["Root-Is-Purelib"] == "true":
            raise WheelToolsError("Cannot add platforms to pure wheel")
        in_info_tags = [tag for name, tag in info.items() if name == "Tag"]
        # Python version, C-API version combinations
        pyc_apis = ["-".join(tag.split("-")[:2]) for tag in in_info_tags]
        # unique Python version, C-API version combinations
        pyc_apis = unique_by_index(pyc_apis)
        # Add new platform tags for each Python version, C-API combination
        required_tags = ["-".join(tup) for tup in product(pyc_apis, platforms)]
        needs_write = False
        for req_tag in required_tags:
            if req_tag in in_info_tags:
                continue
            needs_write = True
            info.add_header("Tag", req_tag)
        if needs_write:
            write_pkg_info(info_fname, info)
            # Tell context manager to write wheel on exit by setting filename
            ctx.out_wheel = out_wheel
    return ctx.out_wheel def rewrite_record(bdist_dir):
    """Rewrite RECORD file with hashes for all files in `wheel_sdir`

    Copied from :method:`wheel.bdist_wheel.bdist_wheel.write_record`

    Will also unsign wheel

    Parameters
    ----------
    bdist_dir : str
        Path of unpacked wheel file
    """
    info_dirs = glob.glob(pjoin(bdist_dir, "*.dist-info"))
    if len(info_dirs) != 1:
        raise WheelToolsError("Should be exactly one `*.dist_info` directory")
    record_path = pjoin(info_dirs[0], "RECORD")
    record_relpath = relpath(record_path, bdist_dir)
    # Unsign wheel - because we're invalidating the record hash
    sig_path = pjoin(info_dirs[0], "RECORD.jws")
    if exists(sig_path):
        os.unlink(sig_path)

    def walk():
        for dir, dirs, files in os.walk(bdist_dir):
            for f in files:
                yield pjoin(dir, f)

    def skip(path):
        """Wheel hashes every possible file."""
        return path == record_relpath

    with _open_for_csv(record_path, "w+") as record_file:
        writer = csv.writer(record_file)
        for path in walk():
            relative_path = relpath(path, bdist_dir)
            if skip(relative_path):
                hash = ""
                size = ""
            else:
                with open(path, "rb") as f:
                    data = f.read()
                digest = hashlib.sha256(data).digest()
                hash = "sha256=" + native(urlsafe_b64encode(digest))
                size = len(data)
            path_for_record = relpath(path, bdist_dir).replace(psep, "/")
            writer.writerow((path_for_record, hash, size)) def assert_equal(first, second):
    __tracebackhide__ = True
    assert first == second def assert_false(condition):
    __tracebackhide__ = True
    assert not condition def assert_raises(expected_exception, *args, **kwargs):
    __tracebackhide__ = True
    return pytest.raises(expected_exception, *args, **kwargs) def assert_true(condition):
    __tracebackhide__ = True
    assert condition PLAT_WHEEL = _collect_wheel("fakepkg1-1.0-cp*.whl") PURE_WHEEL = _collect_wheel("fakepkg2-1.0-py*.whl")
import os
import shutil
from email.message import Message
from os.path import basename, exists, isfile
from os.path import join as pjoin
from os.path import realpath, splitext
from typing import AnyStr
from delocate.pkginfo import read_pkg_info_bytes
from wheel.wheelfile import WheelFile
from ..tmpdirs import InTemporaryDirectory
from ..tools import open_readable, zip2dir
from ..wheeltools import (
    InWheel,
    InWheelCtx,
    WheelToolsError,
    _get_wheelinfo_name,
    add_platforms,
    rewrite_record,
)
from .pytest_tools import assert_equal, assert_false, assert_raises, assert_true
from .test_wheelies import PLAT_WHEEL, PURE_WHEEL
""" Tests for wheeltools utilities
"""




# Template for testing expected wheel information
EXP_PLAT = splitext(PLAT_WHEEL)[0].split("-")[-1]
EXP_ITEMS = [
    ("Generator", "bdist_wheel {pip_version}"),
    ("Root-Is-Purelib", "false"),
    ("Tag", "{pyver}-{abi}-" + EXP_PLAT),
    ("Wheel-Version", "1.0"),
]
# Extra platforms to add
EXTRA_PLATS = ("macosx_10_11_universal2", "macosx_10_11_x86_64")
# Expected outputs for plat added wheels minus wheel-version (that might
# change)
EXTRA_EXPS = [
    ("Generator", "bdist_wheel {pip_version}"),
    ("Root-Is-Purelib", "false"),
] + [("Tag", "{pyver}-{abi}-" + plat) for plat in (EXP_PLAT,) + EXTRA_PLATS]


def assert_record_equal(record_orig: AnyStr, record_new: AnyStr) -> None:
    assert sorted(record_orig.splitlines()) == sorted(record_new.splitlines())


def test_rewrite_record():
    dist_info_sdir = "fakepkg2-1.0.dist-info"
    with InTemporaryDirectory():
        zip2dir(PURE_WHEEL, "wheel")
        record_fname = pjoin("wheel", dist_info_sdir, "RECORD")
        with open_readable(record_fname, "rt") as fobj:
            record_orig = fobj.read()
        # Test we get the same record by rewriting
        os.unlink(record_fname)
        rewrite_record("wheel")
        with open_readable(record_fname, "rt") as fobj:
            record_new = fobj.read()
        assert_record_equal(record_orig, record_new)
        # Test that signature gets deleted
        sig_fname = pjoin("wheel", dist_info_sdir, "RECORD.jws")
        with open(sig_fname, "wt") as fobj:
            fobj.write("something")
        rewrite_record("wheel")
        with open_readable(record_fname, "rt") as fobj:
            record_new = fobj.read()
        assert_record_equal(record_orig, record_new)
        assert_false(exists(sig_fname))
        # Test error for too many dist-infos
        shutil.copytree(
            pjoin("wheel", dist_info_sdir),
            pjoin("wheel", "anotherpkg-2.0.dist-info"),
        )
        assert_raises(WheelToolsError, rewrite_record, "wheel")


def test_in_wheel():
    # Test in-wheel context managers
    # Stuff they share
    for ctx_mgr in InWheel, InWheelCtx:
        with ctx_mgr(PURE_WHEEL):  # No output wheel
            shutil.rmtree("fakepkg2")
            res = sorted(os.listdir("."))
        assert_equal(res, ["fakepkg2-1.0.dist-info"])
        # The original wheel unchanged
        with ctx_mgr(PURE_WHEEL):  # No output wheel
            res = sorted(os.listdir("."))
        assert_equal(res, ["fakepkg2", "fakepkg2-1.0.dist-info"])
        # Make an output wheel file in a temporary directory
        with InTemporaryDirectory():
            mod_path = pjoin("fakepkg2", "module1.py")
            with ctx_mgr(PURE_WHEEL, "mungled.whl"):
                assert_true(isfile(mod_path))
                os.unlink(mod_path)
            with ctx_mgr("mungled.whl"):
                assert_false(isfile(mod_path))
    # Different return from context manager
    with InWheel(PURE_WHEEL) as wheel_path:
        assert_equal(realpath(wheel_path), realpath(os.getcwd()))
    with InWheelCtx(PURE_WHEEL) as ctx:
        assert_equal(realpath(ctx.wheel_path), realpath(os.getcwd()))
    # Set the output wheel inside the with block
    with InTemporaryDirectory() as tmpdir:
        mod_path = pjoin("fakepkg2", "module1.py")

        with InWheelCtx(PURE_WHEEL) as ctx: