diff --git a/tests/test_connection.py b/tests/test_connection.py index b4a3c4b..31e85a8 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -25,7 +25,6 @@ import os from pathlib import Path import socket -import subprocess import sys import unittest from unittest.mock import patch @@ -51,17 +50,10 @@ from .server import Server, ServerTestCase -from .util import asynctest, gss_available, patch_gss, run +from .util import asynctest, gss_available, nc_available, patch_gss from .util import patch_getnameinfo, x509_available -try: - run('which nc') - _nc_available = True -except subprocess.CalledProcessError: # pragma: no cover - _nc_available = False - - class _CheckAlgsClientConnection(asyncssh.SSHClientConnection): """Test specification of encryption algorithms""" @@ -609,7 +601,7 @@ async def test_get_server_host_key_connect_failure(self): with self.assertRaises(OSError): await asyncssh.get_server_host_key('\xff') - @unittest.skipUnless(_nc_available, 'Netcat not available') + @unittest.skipUnless(nc_available, 'Netcat not available') @asynctest async def test_get_server_host_key_proxy(self): """Test retrieving a server host key using proxy command""" @@ -621,7 +613,7 @@ async def test_get_server_host_key_proxy(self): self.assertEqual(key, keylist[0]) - @unittest.skipUnless(_nc_available, 'Netcat not available') + @unittest.skipUnless(nc_available, 'Netcat not available') @asynctest async def test_get_server_host_key_proxy_failure(self): """Test failure retrieving a server host key using proxy command""" @@ -1624,7 +1616,7 @@ async def test_run_server(self): async with self.run_server(sock): pass - @unittest.skipUnless(_nc_available, 'Netcat not available') + @unittest.skipUnless(nc_available, 'Netcat not available') @asynctest async def test_connect_reverse_proxy(self): """Test reverse direction SSH connection with proxy command""" diff --git a/tests/test_connection_auth.py b/tests/test_connection_auth.py index 4301ea5..de806e7 100644 --- a/tests/test_connection_auth.py +++ b/tests/test_connection_auth.py @@ -34,7 +34,7 @@ from .keysign_stub import create_subprocess_exec_stub from .server import Server, ServerTestCase from .util import asynctest, gss_available, patch_getnameinfo, patch_gss -from .util import make_certificate, x509_available +from .util import make_certificate, nc_available, x509_available class _FailValidateHostSSHServerConnection(asyncssh.SSHServerConnection): @@ -658,6 +658,7 @@ async def test_get_server_auth_methods(self): self.assertEqual(auth_methods, ['hostbased']) + @unittest.skipUnless(nc_available, 'Netcat not available') @asynctest async def test_get_server_auth_methods_no_sockname(self): """Test getting auth methods from the test server""" diff --git a/tests/util.py b/tests/util.py index 46eaade..a869944 100644 --- a/tests/util.py +++ b/tests/util.py @@ -24,6 +24,7 @@ import binascii import functools import os +import shutil import subprocess import sys import tempfile @@ -33,7 +34,14 @@ from cryptography.hazmat.backends.openssl import backend -# pylint: disable=unused-import +from asyncssh.gss import gss_available +from asyncssh.logging import logger +from asyncssh.misc import ConnectionLost, SignalReceived +from asyncssh.packet import Byte, String, UInt32, UInt64 +from asyncssh.public_key import generate_private_key + + +# pylint: disable=ungrouped-imports, unused-import try: import bcrypt @@ -41,6 +49,8 @@ except ImportError: # pragma: no cover bcrypt_available = False +nc_available = bool(shutil.which('nc')) + try: import uvloop uvloop_available = True @@ -53,14 +63,7 @@ except ImportError: # pragma: no cover x509_available = False -# pylint: enable=unused-import - -from asyncssh.gss import gss_available -from asyncssh.logging import logger -from asyncssh.misc import ConnectionLost, SignalReceived -from asyncssh.packet import Byte, String, UInt32, UInt64 -from asyncssh.public_key import generate_private_key - +# pylint: enable=ungrouped-imports, unused-import # pylint: disable=no-member