comparison contrib/automation/hgautomation/windows.py @ 43076:2372284d9457

formatting: blacken the codebase

This is using my patch to black
(https://github.com/psf/black/pull/826) so we don't un-wrap collection
literals.

Done with:

  hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**"' | xargs black -S

# skip-blame mass-reformatting only

# no-check-commit reformats foo_bar functions

Differential Revision: https://phab.mercurial-scm.org/D6971

43075:57875cf423c9

import pathlib
import re
import subprocess
import tempfile

from .pypi import (
    upload as pypi_upload,
)
from .winrm import (
    run_powershell,
)


# PowerShell commands to activate a Visual Studio 2008 environment.
# This is essentially a port of vcvarsall.bat to PowerShell.
ACTIVATE_VC9_AMD64 = r'''

MERCURIAL_SCM_BASE_URL = 'https://mercurial-scm.org/release/windows'

X86_USER_AGENT_PATTERN = '.*Windows.*'
X64_USER_AGENT_PATTERN = '.*Windows.*(WOW|x)64.*'

X86_EXE_DESCRIPTION = ('Mercurial {version} Inno Setup installer - x86 Windows '
                       '- does not require admin rights')
X64_EXE_DESCRIPTION = ('Mercurial {version} Inno Setup installer - x64 Windows '
                       '- does not require admin rights')
X86_MSI_DESCRIPTION = ('Mercurial {version} MSI installer - x86 Windows '
                       '- requires admin rights')
X64_MSI_DESCRIPTION = ('Mercurial {version} MSI installer - x64 Windows '
                       '- requires admin rights')

def get_vc_prefix(arch):
    if arch == 'x86':
        return ACTIVATE_VC9_X86
    elif arch == 'x64':

        ssh_dir = temp_dir / '.ssh'
        ssh_dir.mkdir()
        ssh_dir.chmod(0o0700)

        # Generate SSH key to use for communication.
        subprocess.run([
            'ssh-keygen', '-t', 'rsa', '-b', '4096', '-N', '',
            '-f', str(ssh_dir / 'id_rsa')],
            check=True, capture_output=True)

        # Add it to ~/.ssh/authorized_keys on remote.
        # This assumes the file doesn't already exist.
        authorized_keys = r'c:\Users\Administrator\.ssh\authorized_keys'
        winrm_client.execute_cmd(r'mkdir c:\Users\Administrator\.ssh')

            fh.write('  StrictHostKeyChecking no\n')
            fh.write('  UserKnownHostsFile %s\n' % (ssh_dir / 'known_hosts'))
            fh.write('  IdentityFile %s\n' % (ssh_dir / 'id_rsa'))

        if not (hg_repo / '.hg').is_dir():
            raise Exception('%s is not a Mercurial repository; '
                            'synchronization not yet supported' % hg_repo)

        env = dict(os.environ)
        env['HGPLAIN'] = '1'
        env['HGENCODING'] = 'utf-8'

        hg_bin = hg_repo / 'hg'

        res = subprocess.run(
            ['python2.7', str(hg_bin), 'log', '-r', revision, '-T', '{node}'],
            cwd=str(hg_repo), env=env, check=True, capture_output=True)

        full_revision = res.stdout.decode('ascii')

        args = [
            'python2.7', hg_bin,
            '--config', 'ui.ssh=ssh -F %s' % ssh_config,
            '--config', 'ui.remotecmd=c:/hgdev/venv-bootstrap/Scripts/hg.exe',
            # Also ensure .hgtags changes are present so auto version
            # calculation works.
            'push', '-f', '-r', full_revision, '-r', 'file(.hgtags)',
            'ssh://%s/c:/hgdev/src' % public_ip,
        ]

        res = subprocess.run(args, cwd=str(hg_repo), env=env)

        # Allow 1 (no-op) to not trigger error.
        if res.returncode not in (0, 1):
            res.check_returncode()

        run_powershell(winrm_client,
                       HG_UPDATE_CLEAN.format(revision=full_revision))

        # TODO detect dirty local working directory and synchronize accordingly.


def purge_hg(winrm_client):

    dest = dest_path / latest
    print('copying %s to %s' % (source, dest))
    winrm_client.fetch(source, str(dest))


def build_inno_installer(winrm_client, arch: str, dest_path: pathlib.Path,
                         version=None):
    """Build the Inno Setup installer on a remote machine.

    Using a WinRM client, remote commands are executed to build
    a Mercurial Inno Setup installer.
    """

    extra_args = []
    if version:
        extra_args.extend(['--version', version])

    ps = get_vc_prefix(arch) + BUILD_INNO.format(arch=arch,
                                                 extra_args=' '.join(extra_args))
    run_powershell(winrm_client, ps)
    copy_latest_dist(winrm_client, '*.exe', dest_path)


def build_wheel(winrm_client, arch: str, dest_path: pathlib.Path):

    ps = get_vc_prefix(arch) + BUILD_WHEEL.format(arch=arch)
    run_powershell(winrm_client, ps)
    copy_latest_dist(winrm_client, '*.whl', dest_path)


def build_wix_installer(winrm_client, arch: str, dest_path: pathlib.Path,
                        version=None):
    """Build the WiX installer on a remote machine.

    Using a WinRM client, remote commands are executed to build a WiX installer.
    """
    print('Building WiX installer for %s' % arch)
    extra_args = []
    if version:
        extra_args.extend(['--version', version])

    ps = get_vc_prefix(arch) + BUILD_WIX.format(arch=arch,
                                                extra_args=' '.join(extra_args))
    run_powershell(winrm_client, ps)
    copy_latest_dist(winrm_client, '*.msi', dest_path)


def run_tests(winrm_client, python_version, arch, test_flags=''):

    ``arch`` is ``x86`` or ``x64``.
    ``test_flags`` is a str representing extra arguments to pass to
    ``run-tests.py``.
    """
    if not re.match(r'\d\.\d', python_version):
        raise ValueError(r'python_version must be \d.\d; got %s' %
                         python_version)

    if arch not in ('x86', 'x64'):
        raise ValueError('arch must be x86 or x64; got %s' % arch)

    python_path = 'python%s-%s' % (python_version.replace('.', ''), arch)

    ps = RUN_TESTS.format(
        python_path=python_path,
        test_flags=test_flags or '',
    )

    run_powershell(winrm_client, ps)


def resolve_wheel_artifacts(dist_path: pathlib.Path, version: str):

        (
            '10',
            version,
            X64_USER_AGENT_PATTERN,
            '%s/%s' % (MERCURIAL_SCM_BASE_URL, x64_msi_filename),
            X64_MSI_DESCRIPTION.format(version=version)
        )
    )

    lines = ['\t'.join(e) for e in entries]

    return '\n'.join(lines) + '\n'

    print('uploading wheels to PyPI (you may be prompted for credentials)')
    pypi_upload(wheel_paths)


def publish_artifacts_mercurial_scm_org(dist_path: pathlib.Path, version: str,
                                        ssh_username=None):
    """Publish Windows release artifacts to mercurial-scm.org."""
    all_paths = resolve_all_artifacts(dist_path, version)

    for p in all_paths:
        if not p.exists():

    latest_dat_path = '/var/www/release/windows/latest.dat'

    now = datetime.datetime.utcnow()
    backup_path = dist_path / (
        'latest-windows-%s.dat' % now.strftime('%Y%m%dT%H%M%S'))
    print('backing up %s to %s' % (latest_dat_path, backup_path))

    with sftp.open(latest_dat_path, 'rb') as fh:
        latest_dat_old = fh.read()

    with sftp.open(latest_dat_path, 'wb') as fh:
        fh.write(latest_dat_content.encode('ascii'))


def publish_artifacts(dist_path: pathlib.Path, version: str,
                      pypi=True, mercurial_scm_org=True,
                      ssh_username=None):
    """Publish Windows release artifacts.

    Files are found in `dist_path`. We will look for files with version string
    `version`.

    """
    if pypi:
        publish_artifacts_pypi(dist_path, version)

    if mercurial_scm_org:
        publish_artifacts_mercurial_scm_org(dist_path, version,
                                            ssh_username=ssh_username)

43076:2372284d9457

import pathlib
import re
import subprocess
import tempfile

from .pypi import upload as pypi_upload
from .winrm import run_powershell


# PowerShell commands to activate a Visual Studio 2008 environment.
# This is essentially a port of vcvarsall.bat to PowerShell.
ACTIVATE_VC9_AMD64 = r'''

MERCURIAL_SCM_BASE_URL = 'https://mercurial-scm.org/release/windows'

X86_USER_AGENT_PATTERN = '.*Windows.*'
X64_USER_AGENT_PATTERN = '.*Windows.*(WOW|x)64.*'

X86_EXE_DESCRIPTION = (
    'Mercurial {version} Inno Setup installer - x86 Windows '
    '- does not require admin rights'
)
X64_EXE_DESCRIPTION = (
    'Mercurial {version} Inno Setup installer - x64 Windows '
    '- does not require admin rights'
)
X86_MSI_DESCRIPTION = (
    'Mercurial {version} MSI installer - x86 Windows ' '- requires admin rights'
)
X64_MSI_DESCRIPTION = (
    'Mercurial {version} MSI installer - x64 Windows ' '- requires admin rights'
)


def get_vc_prefix(arch):
    if arch == 'x86':
        return ACTIVATE_VC9_X86
    elif arch == 'x64':

        ssh_dir = temp_dir / '.ssh'
        ssh_dir.mkdir()
        ssh_dir.chmod(0o0700)

        # Generate SSH key to use for communication.
        subprocess.run(
            [
                'ssh-keygen',
                '-t',
                'rsa',
                '-b',
                '4096',
                '-N',
                '',
                '-f',
                str(ssh_dir / 'id_rsa'),
            ],
            check=True,
            capture_output=True,
        )

        # Add it to ~/.ssh/authorized_keys on remote.
        # This assumes the file doesn't already exist.
        authorized_keys = r'c:\Users\Administrator\.ssh\authorized_keys'
        winrm_client.execute_cmd(r'mkdir c:\Users\Administrator\.ssh')

            fh.write('  StrictHostKeyChecking no\n')
            fh.write('  UserKnownHostsFile %s\n' % (ssh_dir / 'known_hosts'))
            fh.write('  IdentityFile %s\n' % (ssh_dir / 'id_rsa'))

        if not (hg_repo / '.hg').is_dir():
            raise Exception(
                '%s is not a Mercurial repository; '
                'synchronization not yet supported' % hg_repo
            )

        env = dict(os.environ)
        env['HGPLAIN'] = '1'
        env['HGENCODING'] = 'utf-8'

        hg_bin = hg_repo / 'hg'

        res = subprocess.run(
            ['python2.7', str(hg_bin), 'log', '-r', revision, '-T', '{node}'],
            cwd=str(hg_repo),
            env=env,
            check=True,
            capture_output=True,
        )

        full_revision = res.stdout.decode('ascii')

        args = [
            'python2.7',
            hg_bin,
            '--config',
            'ui.ssh=ssh -F %s' % ssh_config,
            '--config',
            'ui.remotecmd=c:/hgdev/venv-bootstrap/Scripts/hg.exe',
            # Also ensure .hgtags changes are present so auto version
            # calculation works.
            'push',
            '-f',
            '-r',
            full_revision,
            '-r',
            'file(.hgtags)',
            'ssh://%s/c:/hgdev/src' % public_ip,
        ]

        res = subprocess.run(args, cwd=str(hg_repo), env=env)

        # Allow 1 (no-op) to not trigger error.
        if res.returncode not in (0, 1):
            res.check_returncode()

        run_powershell(
            winrm_client, HG_UPDATE_CLEAN.format(revision=full_revision)
        )

        # TODO detect dirty local working directory and synchronize accordingly.


def purge_hg(winrm_client):

    dest = dest_path / latest
    print('copying %s to %s' % (source, dest))
    winrm_client.fetch(source, str(dest))


def build_inno_installer(
    winrm_client, arch: str, dest_path: pathlib.Path, version=None
):
    """Build the Inno Setup installer on a remote machine.

    Using a WinRM client, remote commands are executed to build
    a Mercurial Inno Setup installer.
    """

    extra_args = []
    if version:
        extra_args.extend(['--version', version])

    ps = get_vc_prefix(arch) + BUILD_INNO.format(
        arch=arch, extra_args=' '.join(extra_args)
    )
    run_powershell(winrm_client, ps)
    copy_latest_dist(winrm_client, '*.exe', dest_path)


def build_wheel(winrm_client, arch: str, dest_path: pathlib.Path):

    ps = get_vc_prefix(arch) + BUILD_WHEEL.format(arch=arch)
    run_powershell(winrm_client, ps)
    copy_latest_dist(winrm_client, '*.whl', dest_path)


def build_wix_installer(
    winrm_client, arch: str, dest_path: pathlib.Path, version=None
):
    """Build the WiX installer on a remote machine.

    Using a WinRM client, remote commands are executed to build a WiX installer.
    """
    print('Building WiX installer for %s' % arch)
    extra_args = []
    if version:
        extra_args.extend(['--version', version])

    ps = get_vc_prefix(arch) + BUILD_WIX.format(
        arch=arch, extra_args=' '.join(extra_args)
    )
    run_powershell(winrm_client, ps)
    copy_latest_dist(winrm_client, '*.msi', dest_path)


def run_tests(winrm_client, python_version, arch, test_flags=''):

    ``arch`` is ``x86`` or ``x64``.
    ``test_flags`` is a str representing extra arguments to pass to
    ``run-tests.py``.
    """
    if not re.match(r'\d\.\d', python_version):
        raise ValueError(
            r'python_version must be \d.\d; got %s' % python_version
        )

    if arch not in ('x86', 'x64'):
        raise ValueError('arch must be x86 or x64; got %s' % arch)

    python_path = 'python%s-%s' % (python_version.replace('.', ''), arch)

    ps = RUN_TESTS.format(python_path=python_path, test_flags=test_flags or '',)

    run_powershell(winrm_client, ps)


def resolve_wheel_artifacts(dist_path: pathlib.Path, version: str):

        (
            '10',
            version,
            X64_USER_AGENT_PATTERN,
            '%s/%s' % (MERCURIAL_SCM_BASE_URL, x64_msi_filename),
            X64_MSI_DESCRIPTION.format(version=version),
        ),
    )

    lines = ['\t'.join(e) for e in entries]

    return '\n'.join(lines) + '\n'

    print('uploading wheels to PyPI (you may be prompted for credentials)')
    pypi_upload(wheel_paths)


def publish_artifacts_mercurial_scm_org(
    dist_path: pathlib.Path, version: str, ssh_username=None
):
    """Publish Windows release artifacts to mercurial-scm.org."""
    all_paths = resolve_all_artifacts(dist_path, version)

    for p in all_paths:
        if not p.exists():

    latest_dat_path = '/var/www/release/windows/latest.dat'

    now = datetime.datetime.utcnow()
    backup_path = dist_path / (
        'latest-windows-%s.dat' % now.strftime('%Y%m%dT%H%M%S')
    )
    print('backing up %s to %s' % (latest_dat_path, backup_path))

    with sftp.open(latest_dat_path, 'rb') as fh:
        latest_dat_old = fh.read()

    with sftp.open(latest_dat_path, 'wb') as fh:
        fh.write(latest_dat_content.encode('ascii'))


def publish_artifacts(
    dist_path: pathlib.Path,
    version: str,
    pypi=True,
    mercurial_scm_org=True,
    ssh_username=None,
):
    """Publish Windows release artifacts.

    Files are found in `dist_path`. We will look for files with version string
    `version`.

    """
    if pypi:
        publish_artifacts_pypi(dist_path, version)

    if mercurial_scm_org:
        publish_artifacts_mercurial_scm_org(
            dist_path, version, ssh_username=ssh_username
        )