ansible/test/sanity/code-smell/package-data.py

from __future__ import annotations

import contextlib
import fnmatch
import glob
import os
import pathlib
import re
import shutil
import subprocess
import sys
import tarfile
import tempfile

import packaging.version

from ansible.release import __version__


def assemble_files_to_ship(complete_file_list):
    """
    This looks for all files which should be shipped in the sdist
    """
    # All files which are in the repository except these:
    ignore_patterns = (
        # Developer-only tools
        '.azure-pipelines/*',
        '.github/*',
        '.github/*/*',
        'changelogs/fragments/*',
        'hacking/*',
        'hacking/*/*',
        'test/results/.tmp/*',
        'test/results/.tmp/*/*',
        'test/results/.tmp/*/*/*',
        'test/results/.tmp/*/*/*/*',
        'test/results/.tmp/*/*/*/*/*',
        '.git*',
    )
    ignore_files = frozenset((
        # Developer-only tools
        'changelogs/README.md',
        'changelogs/config.yaml',
        '.cherry_picker.toml',
        '.mailmap',
    ))

    # These files are generated and then intentionally added to the sdist

    # Manpages
    ignore_script = ('ansible-connection', 'ansible-test')
    script_names = [os.path.basename(path) for path in complete_file_list if path.startswith('bin/')]
    manpages = [f'docs/man/man1/{name}.1' for name in script_names if name not in ignore_script]

    # Misc
    misc_generated_files = [
        'PKG-INFO',
    ]

    shipped_files = manpages + misc_generated_files

    for path in complete_file_list:
        if path not in ignore_files:
            for ignore in ignore_patterns:
                if fnmatch.fnmatch(path, ignore):
                    break
            else:
                shipped_files.append(path)

    return shipped_files


def assemble_files_to_install(complete_file_list):
    """
    This looks for all of the files which should show up in an installation of ansible
    """
    ignore_patterns = (
        # Tests excluded from sdist
    )

    pkg_data_files = []
    for path in complete_file_list:

        if path.startswith("lib/ansible"):
            prefix = 'lib'
        elif path.startswith("test/lib/ansible_test"):
            prefix = 'test/lib'
        else:
            continue

        for ignore in ignore_patterns:
            if fnmatch.fnmatch(path, ignore):
                break
        else:
            pkg_data_files.append(os.path.relpath(path, prefix))

    return pkg_data_files


@contextlib.contextmanager
def clean_repository(file_list):
    """Copy the repository to clean it of artifacts"""
    # Create a tempdir that will be the clean repo
    with tempfile.TemporaryDirectory() as repo_root:
        directories = {repo_root + os.path.sep}

        for filename in file_list:
            # Determine if we need to create the directory
            directory = os.path.dirname(filename)
            dest_dir = os.path.join(repo_root, directory)
            if dest_dir not in directories:
                os.makedirs(dest_dir)

                # Keep track of all the directories that now exist
                path_components = directory.split(os.path.sep)
                path = repo_root
                for component in path_components:
                    path = os.path.join(path, component)
                    if path not in directories:
                        directories.add(path)

            # Copy the file
            shutil.copy2(filename, dest_dir, follow_symlinks=False)

        yield repo_root


def create_sdist(tmp_dir):
    """Create an sdist in the repository"""
    # Make sure a changelog exists for this version when testing from devel.
    # When testing from a stable branch the changelog will already exist.
    version = packaging.version.Version(__version__)
    pathlib.Path(f'changelogs/CHANGELOG-v{version.major}.{version.minor}.rst').touch()

    create = subprocess.run(
        [sys.executable, '-m', 'build', '--sdist', '--no-isolation', '--config-setting=--build-manpages', '--outdir', tmp_dir],
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        check=False,
    )

    stderr = create.stderr
    stdout = create.stdout

    if create.returncode != 0:
        raise Exception('make snapshot failed:\n%s' % stderr + '\n' + stdout)

    # Determine path to sdist
    tmp_dir_files = os.listdir(tmp_dir)

    if not tmp_dir_files:
        raise Exception('sdist was not created in the temp dir')
    elif len(tmp_dir_files) > 1:
        raise Exception('Unexpected extra files in the temp dir')

    return os.path.join(tmp_dir, tmp_dir_files[0])


def extract_sdist(sdist_path, tmp_dir):
    """Untar the sdist"""
    # Untar the sdist from the tmp_dir
    with tarfile.open(os.path.join(tmp_dir, sdist_path), 'r|*') as sdist:
        sdist.extractall(path=tmp_dir)

    # Determine the sdist directory name
    sdist_filename = os.path.basename(sdist_path)
    tmp_dir_files = os.listdir(tmp_dir)
    try:
        tmp_dir_files.remove(sdist_filename)
    except ValueError:
        # Unexpected could not find original sdist in temp dir
        raise

    if len(tmp_dir_files) > 1:
        raise Exception('Unexpected extra files in the temp dir')
    elif len(tmp_dir_files) < 1:
        raise Exception('sdist extraction did not occur i nthe temp dir')

    return os.path.join(tmp_dir, tmp_dir_files[0])


def install_sdist(tmp_dir, sdist_dir):
    """Install the extracted sdist into the temporary directory"""
    install = subprocess.run(
        ['python', 'setup.py', 'install', '--root=%s' % tmp_dir],
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        cwd=os.path.join(tmp_dir, sdist_dir),
        check=False,
    )

    stdout, stderr = install.stdout, install.stderr

    if install.returncode != 0:
        raise Exception('sdist install failed:\n%s' % stderr)

    # Determine the prefix for the installed files
    match = re.search('^copying .* -> (%s/.*?/(?:site|dist)-packages)/ansible$' %
                      tmp_dir, stdout, flags=re.M)

    return match.group(1)


def check_sdist_contains_expected(sdist_dir, to_ship_files):
    """Check that the files we expect to ship are present in the sdist"""
    results = []
    for filename in to_ship_files:
        path = os.path.join(sdist_dir, filename)
        if not os.path.exists(path):
            results.append('%s: File was not added to sdist' % filename)

    # Also changelog
    changelog_files = glob.glob(os.path.join(sdist_dir, 'changelogs/CHANGELOG-v2.[0-9]*.rst'))
    if not changelog_files:
        results.append('changelogs/CHANGELOG-v2.*.rst: Changelog file was not added to the sdist')
    elif len(changelog_files) > 1:
        results.append('changelogs/CHANGELOG-v2.*.rst: Too many changelog files: %s'
                       % changelog_files)

    return results


def check_sdist_files_are_wanted(sdist_dir, to_ship_files):
    """Check that all files in the sdist are desired"""
    results = []
    for dirname, dummy, files in os.walk(sdist_dir):
        dirname = os.path.relpath(dirname, start=sdist_dir)
        if dirname == '.':
            dirname = ''

        for filename in files:
            if filename == 'setup.cfg':
                continue

            path = os.path.join(dirname, filename)
            if path not in to_ship_files:

                if fnmatch.fnmatch(path, 'changelogs/CHANGELOG-v2.[0-9]*.rst'):
                    # changelog files are expected
                    continue

                if fnmatch.fnmatch(path, 'lib/ansible_core.egg-info/*'):
                    continue

                # FIXME: ansible-test doesn't pass the paths of symlinks to us so we aren't
                # checking those
                if not os.path.islink(os.path.join(sdist_dir, path)):
                    results.append('%s: File in sdist was not in the repository' % path)

    return results


def check_installed_contains_expected(install_dir, to_install_files):
    """Check that all the files we expect to be installed are"""
    results = []
    for filename in to_install_files:
        path = os.path.join(install_dir, filename)
        if not os.path.exists(path):
            results.append('%s: File not installed' % os.path.join('lib', filename))

    return results


EGG_RE = re.compile('ansible[^/]+\\.egg-info/(PKG-INFO|SOURCES.txt|'
                    'dependency_links.txt|not-zip-safe|requires.txt|top_level.txt|entry_points.txt)$')


def check_installed_files_are_wanted(install_dir, to_install_files):
    """Check that all installed files were desired"""
    results = []

    for dirname, dummy, files in os.walk(install_dir):
        dirname = os.path.relpath(dirname, start=install_dir)
        if dirname == '.':
            dirname = ''

        for filename in files:
            # If this is a byte code cache, look for the python file's name
            directory = dirname
            if filename.endswith('.pyc') or filename.endswith('.pyo'):
                # Remove the trailing "o" or c"
                filename = filename[:-1]

                if directory.endswith('%s__pycache__' % os.path.sep):
                    # Python3 byte code cache, look for the basename of
                    # __pycache__/__init__.cpython-36.py
                    segments = filename.rsplit('.', 2)
                    if len(segments) >= 3:
                        filename = '.'.join((segments[0], segments[2]))
                        directory = os.path.dirname(directory)

            path = os.path.join(directory, filename)

            # Test that the file was listed for installation
            if path not in to_install_files:
                # FIXME: ansible-test doesn't pass the paths of symlinks to us so we
                # aren't checking those
                if not os.path.islink(os.path.join(install_dir, path)):
                    if not EGG_RE.match(path):
                        results.append('%s: File was installed but was not supposed to be' % path)

    return results


def main():
    """All of the files in the repository"""
    complete_file_list = sys.argv[1:] or sys.stdin.read().splitlines()

    # Limit visible files to those reported by ansible-test.
    # This avoids including files which are not committed to git.
    with clean_repository(complete_file_list) as clean_repo_dir:
        os.chdir(clean_repo_dir)

        to_ship_files = assemble_files_to_ship(complete_file_list)
        to_install_files = assemble_files_to_install(complete_file_list)

        results = []
        with tempfile.TemporaryDirectory() as tmp_dir:
            sdist_path = create_sdist(tmp_dir)
            sdist_dir = extract_sdist(sdist_path, tmp_dir)

            # Check that the files that are supposed to be in the sdist are there
            results.extend(check_sdist_contains_expected(sdist_dir, to_ship_files))

            # Check that the files that are in the sdist are in the repository
            results.extend(check_sdist_files_are_wanted(sdist_dir, to_ship_files))

            # install the sdist
            install_dir = install_sdist(tmp_dir, sdist_dir)

            # Check that the files that are supposed to be installed are there
            results.extend(check_installed_contains_expected(install_dir, to_install_files))

            # Check that the files that are installed are supposed to be installed
            results.extend(check_installed_files_are_wanted(install_dir, to_install_files))

        for message in results:
            print(message)


if __name__ == '__main__':
    main()