##// END OF EJS Templates
run-tests: use the same python version for shebang lines on Windows...
run-tests: use the same python version for shebang lines on Windows The latest py3 is used if the minor number isn't specified. After running the script to install all of the build dependencies, that moved the default from 3.8 to 3.9 on the CI system. That in turn caused a bunch of tests to be skipped that were running prior, even when the test runner was invoked with `py -3.8`. While we should almost always use the latest version, we really shouldn't make it hard to test different versions or allow things to randomly break in subtle ways like that. Differential Revision: https://phab.mercurial-scm.org/D10702

File last commit:

r43346:2372284d default
r47953:6f976d54 default
Show More
ssh.py
72 lines | 2.1 KiB | text/x-python | PythonLexer
# ssh.py - Interact with remote SSH servers
#
# Copyright 2019 Gregory Szorc <gregory.szorc@gmail.com>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.
# no-check-code because Python 3 native.
import socket
import time
import warnings
from cryptography.utils import CryptographyDeprecationWarning
import paramiko
def wait_for_ssh(hostname, port, timeout=60, username=None, key_filename=None):
"""Wait for an SSH server to start on the specified host and port."""
class IgnoreHostKeyPolicy(paramiko.MissingHostKeyPolicy):
def missing_host_key(self, client, hostname, key):
return
end_time = time.time() + timeout
# paramiko triggers a CryptographyDeprecationWarning in the cryptography
# package. Let's suppress
with warnings.catch_warnings():
warnings.filterwarnings(
'ignore', category=CryptographyDeprecationWarning
)
while True:
client = paramiko.SSHClient()
client.set_missing_host_key_policy(IgnoreHostKeyPolicy())
try:
client.connect(
hostname,
port=port,
username=username,
key_filename=key_filename,
timeout=5.0,
allow_agent=False,
look_for_keys=False,
)
return client
except socket.error:
pass
except paramiko.AuthenticationException:
raise
except paramiko.SSHException:
pass
if time.time() >= end_time:
raise Exception('Timeout reached waiting for SSH')
time.sleep(1.0)
def exec_command(client, command):
"""exec_command wrapper that combines stderr/stdout and returns channel"""
chan = client.get_transport().open_session()
chan.exec_command(command)
chan.set_combine_stderr(True)
stdin = chan.makefile('wb', -1)
stdout = chan.makefile('r', -1)
return chan, stdin, stdout