diff --git a/colcon_core/python_project/wheel.py b/colcon_core/python_project/wheel.py new file mode 100644 index 00000000..597f3eec --- /dev/null +++ b/colcon_core/python_project/wheel.py @@ -0,0 +1,250 @@ +# Copyright 2026 Open Source Robotics Foundation, Inc. +# Licensed under the Apache License, Version 2.0 + +"""Uninstallation and cleanup utilities for Python packages.""" + +from base64 import urlsafe_b64encode +from configparser import ConfigParser +from email import message_from_binary_file +from functools import lru_cache +from hashlib import sha256 +from io import TextIOWrapper +import os +from pathlib import Path +import shutil +import warnings +from zipfile import ZIP_DEFLATED +from zipfile import ZipFile + +from colcon_core.logging import colcon_logger +from colcon_core.python_project.distribution import _get_install_path +from colcon_core.python_project.distribution import InstalledDistribution +from distlib.scripts import ScriptMaker + +logger = colcon_logger.getChild(__name__) + + +def remove_distributions(name, install_base): + """ + Remove any installed distributions with the given name. + + :param str name: Name of the distribution + :param Path install_base: Path to the base directory to uninstall from + """ + install_base = Path(install_base) + libdirs = [ + Path(_get_install_path('purelib', install_base)), + Path(_get_install_path('platlib', install_base)), + ] + + dists = list( + InstalledDistribution.discover( + name=name, + path=[str(d) for d in libdirs], + prefix_path=install_base, + ) + ) + # InstalledDistribution.discover yields all distributions found on + # the path; filter for the target package name case-insensitively + # with normalization. + target_norm = name.lower().replace('_', '-') + dists = [ + d for d in dists + if d.name.lower().replace('_', '-') == target_norm + ] + + deleted_files = [] + for dist in dists: + libdir = dist.path.parent + files = dist.get_installed_files() + for file in files: + path = libdir / file + if path.is_file() or path.is_symlink(): + logger.debug(f'Removing {path}') + try: + path.unlink() + deleted_files.append(path) + except OSError as e: + logger.warning(f"Could not remove file '{path}': {e}") + + # Clean up empty parent directories recursively + parent_dirs = set() + for file in deleted_files: + for parent in enumerate_parent_dirs(file, install_base): + parent_dirs.add(parent) + + for parent in sorted(parent_dirs, reverse=True): + try: + parent.rmdir() + logger.debug(f'Removing empty directory {parent}' + os.path.sep) + except OSError: + # Directory is not empty or cannot be removed + pass + + +def enumerate_parent_dirs(file, base): + """ + Enumerate all recursive directories under a base directory to a file. + + The base directory itself is not enumerated. + + :param Path file: The file under the base directory + :param Path base: The base directory + :returns: Generator of parent directories + :rtype: Generator[Path, None, None] + """ + try: + rel = file.parent.relative_to(base) + except ValueError: + return + for i in range(1, len(rel.parts) + 1): + yield base.joinpath(*rel.parts[:i]) + + +@lru_cache(maxsize=32) +def _get_script_maker(script_dir, dry_run=False): + """ + Get a ScriptMaker instance. + + :param Path script_dir: Directory where scripts will be created + :param bool dry_run: If True, do not write scripts to disk + :returns: ScriptMaker instance + :rtype: ScriptMaker + """ + sm = ScriptMaker(None, str(script_dir), dry_run=dry_run) + sm.clobber = True + sm.variants = {''} + return sm + + +def write_and_record(libdir, path, lines): + """ + Write file content to disk and compute a fully-qualified RECORD entry. + + :param Path libdir: Library directory where package is installed + :param str/Path path: Path to file to write relative to libdir + :param list[str] lines: Lines of text to write + :returns: Three-element tuple constituting the file's RECORD entry + :rtype: tuple[str, str, str] + """ + path = libdir / path + raw = (os.linesep.join(lines) + os.linesep).encode() + digest = urlsafe_b64encode(sha256(raw).digest()).rstrip(b'=').decode() + path.write_bytes(raw) + return ( + Path(os.path.relpath(path, libdir)).as_posix(), + f'sha256={digest}', + f'{len(raw)}') + + +def install_wheel(wheel_path, install_base, script_dir_override=None): + """ + Install a wheel file under the given installation base directory. + + :param Path wheel_path: Path to the wheel file to be installed + :param Path install_base: Path to the base directory to install under + :param str script_dir_override: Optional path override for scripts + :returns: Path to the installed distribution info directory + :rtype: Path + :raises RuntimeError: If the wheel format is invalid or unsupported + """ + wheel_name = wheel_path.name.split('-') + if len(wheel_name) not in (5, 6): + raise RuntimeError('Invalid wheel file name') + distribution, version = wheel_name[:2] + dist_info_dir = f'{distribution}-{version}.dist-info/' + data_dir = f'{distribution}-{version}.data/' + wheel_file = dist_info_dir + 'WHEEL' + record_file = dist_info_dir + 'RECORD' + entry_points_file = dist_info_dir + 'entry_points.txt' + + install_base = Path(install_base) + + remove_distributions(distribution, install_base) + + with ZipFile( + wheel_path, mode='r', compression=ZIP_DEFLATED, allowZip64=True + ) as wf: + with wf.open(wheel_file) as wf_mf: + wheel_metadata = message_from_binary_file(wf_mf) + + wheel_version = wheel_metadata.get('Wheel-Version', '').split('.') + if len(wheel_version) < 2 or wheel_version[0] != '1': + raise RuntimeError('Wheel file is not supported') + elif wheel_version[1] != '0': + warnings.warn('Wheel format is newer than supported version') + + if wheel_metadata.get('Root-Is-Purelib') in ('true',): + libdir = Path(_get_install_path('purelib', install_base)) + else: + libdir = Path(_get_install_path('platlib', install_base)) + + records = [] + with wf.open(record_file) as wf_rec_bin: + with TextIOWrapper(wf_rec_bin) as wf_rec: + for line in wf_rec: + if ',' in line: + records.append(line.strip().split(',')) + + for record in records: + if record[0] == record_file: + continue + elif not record[0].startswith(data_dir): + info = wf.getinfo(record[0]) + wf.extract(info, libdir) + mode = info.external_attr >> 16 + if mode: + try: + (libdir / record[0]).chmod(mode) + except OSError: + pass + continue + + _, key, subpath = record[0].split('/', 2) + target = Path(_get_install_path(key, install_base)) + target /= subpath + target.parent.mkdir(parents=True, exist_ok=True) + info = wf.getinfo(record[0]) + with wf.open(info) as fsrc: + with target.open('wb') as fdst: + shutil.copyfileobj(fsrc, fdst) + mode = info.external_attr >> 16 + if mode: + try: + target.chmod(mode) + except OSError: + pass + record[0] = Path(os.path.relpath(target, start=libdir)).as_posix() + + records.append(write_and_record( + libdir, + dist_info_dir + 'INSTALLER', + ('colcon-core',))) + + if entry_points_file in wf.namelist(): + ep = ConfigParser() + with wf.open(entry_points_file) as wf_ep_bin: + with TextIOWrapper(wf_ep_bin) as wf_ep: + ep.read_file(wf_ep) + if ep.has_section('console_scripts'): + if script_dir_override: + script_dir = install_base / script_dir_override + else: + script_dir = Path( + _get_install_path('scripts', install_base)) + sm = _get_script_maker(script_dir) + specs = [ + '%s = %s' % pair + for pair in ep.items('console_scripts') + ] + scripts_made = sm.make_multiple(specs) + + records += [ + (Path(os.path.relpath(s, libdir)).as_posix(), '', '') + for s in scripts_made + ] + + with (libdir / record_file).open('w') as f: + f.writelines(','.join(rec) + '\n' for rec in records) + + return libdir / dist_info_dir diff --git a/test/spell_check.words b/test/spell_check.words index 7b72c195..eb55ce8c 100644 --- a/test/spell_check.words +++ b/test/spell_check.words @@ -13,12 +13,14 @@ callables capsys catched changelog +chmod classname colcon coloredlogs compileall configparser contextlib +copyfileobj copytree coroutine coroutines @@ -40,18 +42,22 @@ docstring executables exitstatus fdopen +fdst ffoo filterwarnings foobar fooo fromhex +fsrc functools getcategory +getinfo getpid getpreferredencoding getsignal github hardcodes +hashlib hookimpl hookwrapper https @@ -61,8 +67,11 @@ isatty isdigit iterdir itertools +joinpath junit levelname +libdir +libdirs libexec lineno linter @@ -73,6 +82,7 @@ minversion mkdtemp monkeypatch namedtuple +namelist namespaces nargs noop @@ -140,6 +150,7 @@ stylizers subclassing subparser subparsers +subpath subprocesses symlink symlinks @@ -160,10 +171,14 @@ traceback tryfirst tuples uninstall +uninstallation unittest unittests unlinking unrenamed +urlsafe usefixtures wildcards workaround +writestr +zipfile diff --git a/test/test_install_wheel.py b/test/test_install_wheel.py new file mode 100644 index 00000000..00ec0d20 --- /dev/null +++ b/test/test_install_wheel.py @@ -0,0 +1,225 @@ +# Copyright 2026 Open Source Robotics Foundation, Inc. +# Licensed under the Apache License, Version 2.0 + +"""Tests for package installation utilities.""" + +from base64 import urlsafe_b64encode +import hashlib +from pathlib import Path +from zipfile import ZIP_DEFLATED +from zipfile import ZipFile + +from colcon_core.python_project.distribution import _get_install_path +from colcon_core.python_project.wheel import install_wheel + + +def create_dummy_wheel( + wheel_path, name, version, purelib=True, extra_files=None, + console_scripts=None +): + """ + Create a minimal dummy wheel file. + + :param Path wheel_path: Target path to save the wheel file + :param str name: Distribution name + :param str version: Distribution version + :param bool purelib: True for purelib, False for platlib + :param dict extra_files: Mapping of path in zip to file content + :param dict console_scripts: Mapping of script names to entry points + """ + dist_info = f'{name}-{version}.dist-info' + + files = {} + + # WHEEL file + wheel_content = ( + 'Wheel-Version: 1.0\n' + 'Generator: test\n' + f"Root-Is-Purelib: {'true' if purelib else 'false'}\n" + 'Tag: py3-none-any\n' + ) + files[f'{dist_info}/WHEEL'] = wheel_content + + # METADATA file + metadata_content = ( + 'Metadata-Version: 2.1\n' + f'Name: {name}\n' + f'Version: {version}\n' + ) + files[f'{dist_info}/METADATA'] = metadata_content + + # entry_points.txt + if console_scripts: + ep_content = '[console_scripts]\n' + for script_name, target in console_scripts.items(): + ep_content += f'{script_name} = {target}\n' + files[f'{dist_info}/entry_points.txt'] = ep_content + + # Extra files + if extra_files: + for path, content in extra_files.items(): + files[path] = content + + # Compute RECORD + record_lines = [] + with ZipFile(wheel_path, 'w', compression=ZIP_DEFLATED) as zf: + for path, content in files.items(): + content_bytes = ( + content.encode('utf-8') + if isinstance(content, str) else content + ) + zf.writestr(path, content_bytes) + + digest = urlsafe_b64encode( + hashlib.sha256(content_bytes).digest() + ).rstrip(b'=').decode('utf-8') + size = len(content_bytes) + record_lines.append(f'{path},sha256={digest},{size}') + + # RECORD itself is listed with empty hash and size + record_lines.append(f'{dist_info}/RECORD,,') + zf.writestr( + f'{dist_info}/RECORD', + '\n'.join(record_lines) + '\n' + ) + + +def test_install_wheel_purelib(tmp_path): + """ + Test installation of a pure Python wheel. + + :param Path tmp_path: pytest fixture providing a temp directory + """ + wheel_path = tmp_path / 'dummy_pkg-1.0.0-py3-none-any.whl' + extra_files = { + 'dummy_pkg/__init__.py': '# init\n', + 'dummy_pkg/module.py': 'def foo(): pass\n', + } + create_dummy_wheel( + wheel_path, 'dummy_pkg', '1.0.0', extra_files=extra_files) + + install_base = tmp_path / 'install' + dist_info = install_wheel(wheel_path, install_base) + + libdir = Path(_get_install_path('purelib', install_base)) + + # Verify files were extracted to purelib + assert dist_info.is_dir() + assert dist_info.name == 'dummy_pkg-1.0.0.dist-info' + assert (libdir / 'dummy_pkg' / '__init__.py').is_file() + assert (libdir / 'dummy_pkg' / 'module.py').is_file() + + # Verify INSTALLER metadata exists + installer_file = dist_info / 'INSTALLER' + assert installer_file.is_file() + assert installer_file.read_text(encoding='utf-8').strip() == 'colcon-core' + + # Verify RECORD exists + record_file = dist_info / 'RECORD' + assert record_file.is_file() + + +def test_install_wheel_platlib(tmp_path): + """ + Test installation of a platform-specific wheel. + + :param Path tmp_path: pytest fixture providing a temp directory + """ + wheel_path = tmp_path / 'dummy_pkg-1.0.0-py3-none-any.whl' + extra_files = { + 'dummy_pkg/__init__.py': '# init\n', + } + create_dummy_wheel( + wheel_path, 'dummy_pkg', '1.0.0', purelib=False, + extra_files=extra_files + ) + + install_base = tmp_path / 'install' + dist_info = install_wheel(wheel_path, install_base) + + libdir = Path(_get_install_path('platlib', install_base)) + + assert dist_info.is_dir() + assert (libdir / 'dummy_pkg' / '__init__.py').is_file() + + +def test_install_wheel_with_console_scripts(tmp_path): + """ + Test installation of a wheel with entry point console scripts. + + :param Path tmp_path: pytest fixture providing a temp directory + """ + wheel_path = tmp_path / 'dummy_pkg-1.0.0-py3-none-any.whl' + console_scripts = { + 'dummy-cli': 'dummy_pkg.main:cli_entry', + } + create_dummy_wheel( + wheel_path, 'dummy_pkg', '1.0.0', console_scripts=console_scripts + ) + + install_base = tmp_path / 'install' + install_wheel(wheel_path, install_base) + + script_dir = Path(_get_install_path('scripts', install_base)) + scripts = list(script_dir.glob('dummy-cli*')) + assert len(scripts) > 0 + + +def test_install_wheel_with_data_files(tmp_path): + """ + Test installation of a wheel with package data files. + + :param Path tmp_path: pytest fixture providing a temp directory + """ + wheel_path = tmp_path / 'dummy_pkg-1.0.0-py3-none-any.whl' + extra_files = { + 'dummy_pkg-1.0.0.data/data/share/dummy_pkg/resource.txt': + 'resource data\n', + } + create_dummy_wheel( + wheel_path, 'dummy_pkg', '1.0.0', extra_files=extra_files + ) + + install_base = tmp_path / 'install' + install_wheel(wheel_path, install_base) + + # Path gets resolved under install_base/share/dummy_pkg/resource.txt + data_file = install_base / 'share' / 'dummy_pkg' / 'resource.txt' + assert data_file.is_file() + assert data_file.read_text(encoding='utf-8') == 'resource data\n' + + +def test_install_wheel_cleanup_old(tmp_path): + """ + Test that installing a package cleans up the older version. + + :param Path tmp_path: pytest fixture providing a temp directory + """ + install_base = tmp_path / 'install' + libdir = Path(_get_install_path('purelib', install_base)) + + # Install version 1.0.0 + wheel_v1 = tmp_path / 'dummy_pkg-1.0.0-py3-none-any.whl' + create_dummy_wheel( + wheel_v1, 'dummy_pkg', '1.0.0', + extra_files={'dummy_pkg/v1.py': '# v1\n'} + ) + dist_info_v1 = install_wheel(wheel_v1, install_base) + assert dist_info_v1.is_dir() + assert (libdir / 'dummy_pkg' / 'v1.py').is_file() + + # Install version 2.0.0 + wheel_v2 = tmp_path / 'dummy_pkg-2.0.0-py3-none-any.whl' + create_dummy_wheel( + wheel_v2, 'dummy_pkg', '2.0.0', + extra_files={'dummy_pkg/v2.py': '# v2\n'} + ) + dist_info_v2 = install_wheel(wheel_v2, install_base) + + # Verify v1 files are deleted + assert not dist_info_v1.exists() + assert not (libdir / 'dummy_pkg' / 'v1.py').exists() + + # Verify v2 files are installed + assert dist_info_v2.is_dir() + assert (libdir / 'dummy_pkg' / 'v2.py').is_file() diff --git a/test/test_remove_wheel.py b/test/test_remove_wheel.py new file mode 100644 index 00000000..c6f5d3d6 --- /dev/null +++ b/test/test_remove_wheel.py @@ -0,0 +1,192 @@ +# Copyright 2026 Open Source Robotics Foundation, Inc. +# Licensed under the Apache License, Version 2.0 + +"""Tests for package uninstallation utilities.""" + +import os +from pathlib import Path +import shutil + +from colcon_core.python_project.distribution import _get_install_path +from colcon_core.python_project.distribution import InstalledDistribution +from colcon_core.python_project.wheel import remove_distributions +import pytest + +TEST_DISTS_ROOT = Path(__file__).parent / 'mock_distributions' + + +@pytest.fixture +def temp_workspace(tmp_path): + """ + Set up a temporary workspace. + + This copies mock distributions into the real python path. + + :param Path tmp_path: pytest fixture providing a temp directory + :returns: Path to the temporary workspace root + """ + real_purelib = Path(_get_install_path('purelib', tmp_path)) + real_purelib.mkdir(parents=True, exist_ok=True) + + # Copy site-packages contents into the real library path + mock_site_packages = TEST_DISTS_ROOT / 'lib' / 'python' / 'site-packages' + shutil.copytree(mock_site_packages, real_purelib, dirs_exist_ok=True) + + # Copy non-lib files (e.g. bin, Scripts, src) to workspace root + for item in TEST_DISTS_ROOT.iterdir(): + if item.name == 'lib': + continue + dest = tmp_path / item.name + if item.is_dir(): + shutil.copytree(item, dest, dirs_exist_ok=True) + else: + shutil.copy(item, dest) + + # Dynamically update the relative path inside egg-link files + # to match the layout depth of the current platform/Python version + src_dir = tmp_path / 'src' / 'typical_egg_link' + for egg_link in real_purelib.glob('*.egg-link'): + try: + rel_path = os.path.relpath(str(src_dir), start=str(real_purelib)) + except ValueError: + rel_path = str(src_dir) + egg_link.write_text(f'{rel_path}\n.\n', encoding='utf-8') + + yield tmp_path + + +def test_remove_typical_dist_info(temp_workspace): + """ + Test removal of a standard .dist-info package. + + :param Path temp_workspace: Temporary workspace fixture + """ + name = 'typical-dist-info' + libdir = Path(_get_install_path('purelib', temp_workspace)) + + # Discover and verify distribution exists + dists = list( + InstalledDistribution.discover( + name=name, path=[str(libdir)], prefix_path=temp_workspace + ) + ) + dists = [ + d for d in dists + if d.name.lower().replace('_', '-') == name.lower().replace('_', '-') + ] + assert len(dists) == 1 + dist = dists[0] + + # Keep track of expected file paths + files = dist.get_installed_files() + assert len(files) > 0 + file_paths = [dist.path.parent / f for f in files] + for p in file_paths: + assert p.exists() + + # Call uninstallation utility + remove_distributions(name, temp_workspace) + + # Assert all files are deleted + for p in file_paths: + assert not p.exists() + + # Assert empty parent directories are cleaned up + assert not (libdir / 'typical_dist_info').exists() + assert not (libdir / 'typical_dist_info-0.0.0.dist-info').exists() + + +def test_remove_typical_egg_info(temp_workspace): + """ + Test removal of a legacy .egg-info package. + + :param Path temp_workspace: Temporary workspace fixture + """ + name = 'typical-egg-info' + libdir = Path(_get_install_path('purelib', temp_workspace)) + + # Discover and verify distribution exists + dists = list( + InstalledDistribution.discover( + name=name, path=[str(libdir)], prefix_path=temp_workspace + ) + ) + dists = [ + d for d in dists + if d.name.lower().replace('_', '-') == name.lower().replace('_', '-') + ] + assert len(dists) == 1 + dist = dists[0] + + # Keep track of expected file paths + files = dist.get_installed_files() + assert len(files) > 0 + file_paths = [dist.path.parent / f for f in files] + for p in file_paths: + assert p.exists() + + # Call uninstallation utility + remove_distributions(name, temp_workspace) + + # Assert all files are deleted + for p in file_paths: + assert not p.exists() + + # Assert empty parent directories are cleaned up + assert not (libdir / 'typical_egg_info').exists() + assert not (libdir / 'typical_egg_info-0.0.0.egg-info').exists() + + +def test_remove_typical_egg_link(temp_workspace): + """ + Test removal of a legacy develop egg-link. + + :param Path temp_workspace: Temporary workspace fixture + """ + name = 'typical-egg-link' + libdir = Path(_get_install_path('purelib', temp_workspace)) + egg_link = libdir / 'typical-egg-link.egg-link' + + # Verify egg link exists initially + assert egg_link.is_file() + + # Discover and verify distribution exists + dists = list( + InstalledDistribution.discover( + name=name, path=[str(libdir)], prefix_path=temp_workspace + ) + ) + dists = [ + d for d in dists + if d.name.lower().replace('_', '-') == name.lower().replace('_', '-') + ] + assert len(dists) == 1 + dist = dists[0] + + # Keep track of expected file paths + files = dist.get_installed_files() + assert len(files) > 0 + file_paths = [dist.path.parent / f for f in files] + for p in file_paths: + assert p.exists() + + # Call uninstallation utility + remove_distributions(name, temp_workspace) + + # Assert all files are deleted + for p in file_paths: + assert not p.exists() + + # Assert egg-link file is deleted + assert not egg_link.exists() + + +def test_remove_nonexistent_distribution(temp_workspace): + """ + Test uninstallation of a package that does not exist on disk. + + :param Path temp_workspace: Temporary workspace fixture + """ + name = 'nonexistent-package' + # Should complete without error + remove_distributions(name, temp_workspace)