diff options
| -rw-r--r-- | tests/__init__.py | 37 | ||||
| -rw-r--r-- | tests/loop.py | 6 | ||||
| -rw-r--r-- | tests/stub_sftp.py | 1 | ||||
| -rw-r--r-- | tests/test_auth.py | 6 | ||||
| -rw-r--r-- | tests/test_buffered_pipe.py | 1 | ||||
| -rw-r--r-- | tests/test_client.py | 11 | ||||
| -rwxr-xr-x | tests/test_file.py | 8 | ||||
| -rw-r--r-- | tests/test_gssapi.py | 33 | ||||
| -rw-r--r-- | tests/test_hostkeys.py | 1 | ||||
| -rw-r--r-- | tests/test_kex.py | 3 | ||||
| -rw-r--r-- | tests/test_message.py | 1 | ||||
| -rw-r--r-- | tests/test_packetizer.py | 5 | ||||
| -rw-r--r-- | tests/test_pkey.py | 3 | ||||
| -rwxr-xr-x | tests/test_sftp.py | 16 | ||||
| -rw-r--r-- | tests/test_sftp_big.py | 4 | ||||
| -rw-r--r-- | tests/test_ssh_gss.py | 30 | ||||
| -rw-r--r-- | tests/test_transport.py | 10 | ||||
| -rw-r--r-- | tests/test_util.py | 1 | ||||
| -rw-r--r-- | tests/util.py | 22 |
19 files changed, 99 insertions, 100 deletions
diff --git a/tests/__init__.py b/tests/__init__.py index 8878f14d..be1d2daa 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,36 +1 @@ -# Copyright (C) 2017 Martin Packman <gzlist@googlemail.com> -# -# This file is part of paramiko. -# -# Paramiko is free software; you can redistribute it and/or modify it under the -# terms of the GNU Lesser General Public License as published by the Free -# Software Foundation; either version 2.1 of the License, or (at your option) -# any later version. -# -# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY -# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR -# A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more -# details. -# -# You should have received a copy of the GNU Lesser General Public License -# along with Paramiko; if not, write to the Free Software Foundation, Inc., -# 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA. - -"""Base classes and helpers for testing paramiko.""" - -import unittest - -from paramiko.py3compat import ( - builtins, - ) - - -def skipUnlessBuiltin(name): - """Skip decorated test if builtin name does not exist.""" - if getattr(builtins, name, None) is None: - skip = getattr(unittest, "skip", None) - if skip is None: - # Python 2.6 pseudo-skip - return lambda func: None - return skip("No builtin " + repr(name)) - return lambda func: func +# This file's just here so test modules can use explicit-relative imports. diff --git a/tests/loop.py b/tests/loop.py index e805ad96..6c432867 100644 --- a/tests/loop.py +++ b/tests/loop.py @@ -16,11 +16,9 @@ # along with Paramiko; if not, write to the Free Software Foundation, Inc., # 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA. -""" -... -""" +import socket +import threading -import threading, socket from paramiko.common import asbytes diff --git a/tests/stub_sftp.py b/tests/stub_sftp.py index 334af561..1b3d7387 100644 --- a/tests/stub_sftp.py +++ b/tests/stub_sftp.py @@ -22,6 +22,7 @@ A stub SFTP server for loopback SFTP testing. import os import sys + from paramiko import ( ServerInterface, SFTPServerInterface, SFTPServer, SFTPAttributes, SFTPHandle, SFTP_OK, AUTH_SUCCESSFUL, OPEN_SUCCEEDED, diff --git a/tests/test_auth.py b/tests/test_auth.py index ffd6bbc8..9ca48947 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -30,8 +30,10 @@ from paramiko import ( ) from paramiko import AUTH_FAILED, AUTH_PARTIALLY_SUCCESSFUL, AUTH_SUCCESSFUL from paramiko.py3compat import u -from tests.loop import LoopSocket -from tests.util import _support + +from .loop import LoopSocket +from .util import _support + _pwd = u('\u2022') diff --git a/tests/test_buffered_pipe.py b/tests/test_buffered_pipe.py index eeb4d0ad..03616c55 100644 --- a/tests/test_buffered_pipe.py +++ b/tests/test_buffered_pipe.py @@ -23,6 +23,7 @@ Some unit tests for BufferedPipe. import threading import time import unittest + from paramiko.buffered_pipe import BufferedPipe, PipeTimeout from paramiko import pipe from paramiko.py3compat import b diff --git a/tests/test_client.py b/tests/test_client.py index 937a10d0..b5fe2e06 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -23,21 +23,22 @@ Some unit tests for SSHClient. from __future__ import with_statement import gc +import os import platform import socket -from tempfile import mkstemp import threading +import time import unittest -import weakref import warnings -import os -import time -from tests.util import _support +import weakref +from tempfile import mkstemp import paramiko from paramiko.py3compat import PY2, b from paramiko.ssh_exception import SSHException +from .util import _support + FINGERPRINTS = { 'ssh-dss': b'\x44\x78\xf0\xb9\xa2\x3c\xc5\x18\x20\x09\xff\x75\x5b\xc1\xd2\x6c', diff --git a/tests/test_file.py b/tests/test_file.py index b33ecd51..3d2c94e6 100755 --- a/tests/test_file.py +++ b/tests/test_file.py @@ -27,7 +27,7 @@ from paramiko.common import linefeed_byte, crlf, cr_byte from paramiko.file import BufferedFile from paramiko.py3compat import BytesIO -from tests import skipUnlessBuiltin +from .util import needs_builtin class LoopbackFile (BufferedFile): @@ -198,13 +198,13 @@ class BufferedFileTest (unittest.TestCase): f.write(text) self.assertEqual(f.read(), text.encode("utf-8")) - @skipUnlessBuiltin('memoryview') + @needs_builtin('memoryview') def test_write_bytearray(self): with LoopbackFile('rb+') as f: f.write(bytearray(12)) self.assertEqual(f.read(), 12 * b"\0") - @skipUnlessBuiltin('buffer') + @needs_builtin('buffer') def test_write_buffer(self): data = 3 * b"pretend giant block of data\n" offsets = range(0, len(data), 8) @@ -213,7 +213,7 @@ class BufferedFileTest (unittest.TestCase): f.write(buffer(data, offset, 8)) self.assertEqual(f.read(), data) - @skipUnlessBuiltin('memoryview') + @needs_builtin('memoryview') def test_write_memoryview(self): data = 3 * b"pretend giant block of data\n" offsets = range(0, len(data), 8) diff --git a/tests/test_gssapi.py b/tests/test_gssapi.py index bc220108..d4b632be 100644 --- a/tests/test_gssapi.py +++ b/tests/test_gssapi.py @@ -25,14 +25,17 @@ Test the used APIs for GSS-API / SSPI authentication import unittest import socket +from .util import needs_gssapi + +@needs_gssapi class GSSAPITest(unittest.TestCase): - @staticmethod - def init(hostname=None, srv_mode=False): - global krb5_mech, targ_name, server_mode - krb5_mech = "1.2.840.113554.1.2.2" - targ_name = hostname - server_mode = srv_mode + def setup(): + # TODO: these vars should all come from os.environ or whatever the + # approved pytest method is for runtime-configuring test data. + self.krb5_mech = "1.2.840.113554.1.2.2" + self.targ_name = "hostname" + self.server_mode = False def test_1_pyasn1(self): """ @@ -40,9 +43,9 @@ class GSSAPITest(unittest.TestCase): """ from pyasn1.type.univ import ObjectIdentifier from pyasn1.codec.der import encoder, decoder - oid = encoder.encode(ObjectIdentifier(krb5_mech)) + oid = encoder.encode(ObjectIdentifier(self.krb5_mech)) mech, __ = decoder.decode(oid) - self.assertEquals(krb5_mech, mech.__str__()) + self.assertEquals(self.krb5_mech, mech.__str__()) def test_2_gssapi_sspi(self): """ @@ -61,7 +64,7 @@ class GSSAPITest(unittest.TestCase): mic_msg = b"G'day Mate!" if _API == "MIT": - if server_mode: + if self.server_mode: gss_flags = (gssapi.C_PROT_READY_FLAG, gssapi.C_INTEG_FLAG, gssapi.C_MUTUAL_FLAG, @@ -73,13 +76,13 @@ class GSSAPITest(unittest.TestCase): # Initialize a GSS-API context. ctx = gssapi.Context() ctx.flags = gss_flags - krb5_oid = gssapi.OID.mech_from_string(krb5_mech) - target_name = gssapi.Name("host@" + targ_name, + krb5_oid = gssapi.OID.mech_from_string(self.krb5_mech) + target_name = gssapi.Name("host@" + self.targ_name, gssapi.C_NT_HOSTBASED_SERVICE) gss_ctxt = gssapi.InitContext(peer_name=target_name, mech_type=krb5_oid, req_flags=ctx.flags) - if server_mode: + if self.server_mode: c_token = gss_ctxt.step(c_token) gss_ctxt_status = gss_ctxt.established self.assertEquals(False, gss_ctxt_status) @@ -99,7 +102,7 @@ class GSSAPITest(unittest.TestCase): # Build MIC mic_token = gss_ctxt.get_mic(mic_msg) - if server_mode: + if self.server_mode: # Check MIC status = gss_srv_ctxt.verify_mic(mic_msg, mic_token) self.assertEquals(0, status) @@ -110,11 +113,11 @@ class GSSAPITest(unittest.TestCase): sspicon.ISC_REQ_DELEGATE ) # Initialize a GSS-API context. - target_name = "host/" + socket.getfqdn(targ_name) + target_name = "host/" + socket.getfqdn(self.targ_name) gss_ctxt = sspi.ClientAuth("Kerberos", scflags=gss_flags, targetspn=target_name) - if server_mode: + if self.server_mode: error, token = gss_ctxt.authorize(c_token) c_token = token[0].Buffer self.assertEquals(0, error) diff --git a/tests/test_hostkeys.py b/tests/test_hostkeys.py index 2c7ceeb9..cd75f8ab 100644 --- a/tests/test_hostkeys.py +++ b/tests/test_hostkeys.py @@ -23,6 +23,7 @@ Some unit tests for HostKeys. from binascii import hexlify import os import unittest + import paramiko from paramiko.py3compat import decodebytes diff --git a/tests/test_kex.py b/tests/test_kex.py index 19804fbf..badab61e 100644 --- a/tests/test_kex.py +++ b/tests/test_kex.py @@ -24,6 +24,9 @@ from binascii import hexlify import os import unittest +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives.asymmetric import ec + import paramiko.util from paramiko.kex_group1 import KexGroup1 from paramiko.kex_gex import KexGex, KexGexSHA256 diff --git a/tests/test_message.py b/tests/test_message.py index f18cae90..645b0509 100644 --- a/tests/test_message.py +++ b/tests/test_message.py @@ -21,6 +21,7 @@ Some unit tests for ssh protocol message blocks. """ import unittest + from paramiko.message import Message from paramiko.common import byte_chr, zero_byte diff --git a/tests/test_packetizer.py b/tests/test_packetizer.py index 02173292..414b7e38 100644 --- a/tests/test_packetizer.py +++ b/tests/test_packetizer.py @@ -27,11 +27,12 @@ from hashlib import sha1 from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.ciphers import algorithms, Cipher, modes -from tests.loop import LoopSocket - from paramiko import Message, Packetizer, util from paramiko.common import byte_chr, zero_byte +from .loop import LoopSocket + + x55 = byte_chr(0x55) x1f = byte_chr(0x1f) diff --git a/tests/test_pkey.py b/tests/test_pkey.py index 6f5c25f0..f5da1fde 100644 --- a/tests/test_pkey.py +++ b/tests/test_pkey.py @@ -30,7 +30,8 @@ import base64 from paramiko import RSAKey, DSSKey, ECDSAKey, Message, util from paramiko.py3compat import StringIO, byte_chr, b, bytes, PY2 -from tests.util import _support +from .util import _support + # from openssh's ssh-keygen PUB_RSA = 'ssh-rsa AAAAB3NzaC1yc2EAAAABIwAAAIEA049W6geFpmsljTwfvI1UmKWWJPNFI74+vNKTk4dmzkQY2yAMs6FhlvhlI8ysU4oj71ZsRYMecHbBbxdN79+JRFVYTKaLqjwGENeTd+yv4q+V2PvZv3fLnzApI3l7EJCqhWwJUHJ1jAkZzqDx0tyOL4uoZpww3nmE0kb3y21tH4c=' diff --git a/tests/test_sftp.py b/tests/test_sftp.py index 624297c8..d3760bf7 100755 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -33,15 +33,17 @@ from binascii import hexlify from tempfile import mkstemp import paramiko +import paramiko.util from paramiko.py3compat import PY2, b, u, StringIO from paramiko.common import o777, o600, o666, o644 -from tests import skipUnlessBuiltin -from tests.stub_sftp import StubServer, StubSFTPServer -from tests.loop import LoopSocket -from tests.util import _support -import paramiko.util from paramiko.sftp_attr import SFTPAttributes +from .util import needs_builtin +from .stub_sftp import StubServer, StubSFTPServer +from .loop import LoopSocket +from .util import _support + + ARTICLE = ''' Insulin sensitivity and liver insulin receptor structure in ducks from two genera @@ -818,7 +820,7 @@ class SFTPTest (unittest.TestCase): sftp_attributes = SFTPAttributes() self.assertEqual(str(sftp_attributes), "?--------- 1 0 0 0 (unknown date) ?") - @skipUnlessBuiltin('buffer') + @needs_builtin('buffer') def test_write_buffer(self): """Test write() using a buffer instance.""" data = 3 * b'A potentially large block of data to chunk up.\n' @@ -832,7 +834,7 @@ class SFTPTest (unittest.TestCase): finally: sftp.remove('%s/write_buffer' % FOLDER) - @skipUnlessBuiltin('memoryview') + @needs_builtin('memoryview') def test_write_memoryview(self): """Test write() using a memoryview instance.""" data = 3 * b'A potentially large block of data to chunk up.\n' diff --git a/tests/test_sftp_big.py b/tests/test_sftp_big.py index cfad5682..580ba64e 100644 --- a/tests/test_sftp_big.py +++ b/tests/test_sftp_big.py @@ -31,7 +31,9 @@ import time import unittest from paramiko.common import o660 -from tests.test_sftp import get_sftp + +from .test_sftp import get_sftp + FOLDER = os.environ.get('TEST_FOLDER', 'temp-testing000') diff --git a/tests/test_ssh_gss.py b/tests/test_ssh_gss.py index 081d942b..f0645e0e 100644 --- a/tests/test_ssh_gss.py +++ b/tests/test_ssh_gss.py @@ -29,17 +29,20 @@ import unittest import paramiko -from tests.util import _support -from tests.test_client import FINGERPRINTS +from .util import _support, needs_gssapi +from .test_client import FINGERPRINTS -class NullServer (paramiko.ServerInterface): +class NullServer (paramiko.ServerInterface): def get_allowed_auths(self, username): return 'gssapi-with-mic,publickey' - def check_auth_gssapi_with_mic(self, username, - gss_authenticated=paramiko.AUTH_FAILED, - cc_file=None): + def check_auth_gssapi_with_mic( + self, + username, + gss_authenticated=paramiko.AUTH_FAILED, + cc_file=None, + ): if gss_authenticated == paramiko.AUTH_SUCCESSFUL: return paramiko.AUTH_SUCCESSFUL return paramiko.AUTH_FAILED @@ -66,18 +69,15 @@ class NullServer (paramiko.ServerInterface): return True +@needs_gssapi class GSSAuthTest(unittest.TestCase): - @staticmethod - def init(username, hostname): - global krb5_principal, targ_name - krb5_principal = username - targ_name = hostname - def setUp(self): - self.username = krb5_principal - self.hostname = socket.getfqdn(targ_name) + # TODO: username and targ_name should come from os.environ or whatever + # the approved pytest method is for runtime-configuring test data. + self.username = "krb5_principal" + self.hostname = socket.getfqdn("targ_name") self.sockl = socket.socket() - self.sockl.bind((targ_name, 0)) + self.sockl.bind(("targ_name", 0)) self.sockl.listen(1) self.addr, self.port = self.sockl.getsockname() self.event = threading.Event() diff --git a/tests/test_transport.py b/tests/test_transport.py index c3f67803..7fa67c43 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -43,9 +43,9 @@ from paramiko.common import ( ) from paramiko.py3compat import bytes from paramiko.message import Message -from tests import skipUnlessBuiltin -from tests.loop import LoopSocket -from tests.util import _support + +from .util import needs_builtin, _support +from .loop import LoopSocket LONG_BANNER = """\ @@ -892,7 +892,7 @@ class TransportTest(unittest.TestCase): expected = text.encode("utf-8") self.assertEqual(sfile.read(len(expected)), expected) - @skipUnlessBuiltin('buffer') + @needs_builtin('buffer') def test_channel_send_buffer(self): """ verify sending buffer instances to a channel @@ -915,7 +915,7 @@ class TransportTest(unittest.TestCase): chan.sendall(buffer(data)) self.assertEqual(sfile.read(len(data)), data) - @skipUnlessBuiltin('memoryview') + @needs_builtin('memoryview') def test_channel_send_memoryview(self): """ verify sending memoryview instances to a channel diff --git a/tests/test_util.py b/tests/test_util.py index 7880e156..1b6b5c92 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -30,6 +30,7 @@ import paramiko.util from paramiko.util import lookup_ssh_host_config as host_config, safe_string from paramiko.py3compat import StringIO, byte_ord, b + # Note some lines in this configuration have trailing spaces on purpose test_config_file = """\ Host * diff --git a/tests/util.py b/tests/util.py index bde9fa88..db1f077c 100644 --- a/tests/util.py +++ b/tests/util.py @@ -1,7 +1,23 @@ -import os +from os.path import dirname, realpath, join + +import pytest + +from paramiko.py3compat import builtins -root_path = os.path.dirname(os.path.realpath(__file__)) def _support(filename): - return os.path.join(root_path, filename) + return join(dirname(realpath(__file__)), filename) + + +# TODO: consider using pytest.importorskip('gssapi') instead? We presumably +# still need CLI configurability for the Kerberos parameters, though, so can't +# JUST key off presence of GSSAPI optional dependency... +# TODO: anyway, s/True/os.environ.get('RUN_GSSAPI', False)/ or something. +needs_gssapi = pytest.mark.skipif(True, reason="No GSSAPI to test") + +def needs_builtin(name): + """ + Skip decorated test if builtin name does not exist. + """ + return pytest.mark.skipif(not hasattr(builtins, name)) |