summaryrefslogtreecommitdiffhomepage
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/__init__.py33
-rw-r--r--tests/loop.py6
-rw-r--r--tests/stub_sftp.py1
-rw-r--r--tests/test_auth.py6
-rw-r--r--tests/test_buffered_pipe.py1
-rw-r--r--tests/test_client.py11
-rwxr-xr-xtests/test_file.py8
-rw-r--r--tests/test_gssapi.py33
-rw-r--r--tests/test_hostkeys.py1
-rw-r--r--tests/test_kex.py5
-rw-r--r--tests/test_message.py1
-rw-r--r--tests/test_packetizer.py5
-rw-r--r--tests/test_pkey.py3
-rwxr-xr-xtests/test_sftp.py16
-rw-r--r--tests/test_sftp_big.py4
-rw-r--r--tests/test_ssh_gss.py30
-rw-r--r--tests/test_transport.py12
-rw-r--r--tests/test_util.py1
-rw-r--r--tests/util.py23
19 files changed, 101 insertions, 99 deletions
diff --git a/tests/__init__.py b/tests/__init__.py
index ea571abe..be1d2daa 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -1,32 +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:
- return unittest.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 0d673091..19545865 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, SFTP_FAILURE, AUTH_SUCCESSFUL, OPEN_SUCCEEDED,
diff --git a/tests/test_auth.py b/tests/test_auth.py
index e1f06cc1..c619dc33 100644
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@@ -31,8 +31,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 230739b0..fd662604 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -23,22 +23,23 @@ 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.pkey import PublicBlob
from paramiko.common import PY2
from paramiko.ssh_exception import SSHException, AuthenticationException
+from .util import _support
+
requires_gss_auth = unittest.skipUnless(
paramiko.GSS_AUTH_AVAILABLE, "GSS auth not available"
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 b7f588f7..b5808e7e 100644
--- a/tests/test_kex.py
+++ b/tests/test_kex.py
@@ -24,14 +24,15 @@ from binascii import hexlify, unhexlify
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
from paramiko import Message
from paramiko.common import byte_chr
from paramiko.kex_ecdh_nist import KexNistp256
-from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives.asymmetric import ec
def dummy_urandom(n):
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 be0f9465..c745232b 100644
--- a/tests/test_pkey.py
+++ b/tests/test_pkey.py
@@ -30,7 +30,8 @@ import base64
from paramiko import RSAKey, DSSKey, ECDSAKey, Ed25519Key, 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 7264baeb..518efd83 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
@@ -849,7 +851,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'
@@ -863,7 +865,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 f4cf622f..46c26c0e 100644
--- a/tests/test_transport.py
+++ b/tests/test_transport.py
@@ -33,7 +33,7 @@ import unittest
from paramiko import (
Transport, SecurityOptions, ServerInterface, RSAKey, DSSKey, SSHException,
- ChannelException, Packetizer,
+ ChannelException, Packetizer,
)
from paramiko import AUTH_FAILED, AUTH_SUCCESSFUL
from paramiko import OPEN_SUCCEEDED, OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
@@ -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 = """\
@@ -881,7 +881,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
@@ -904,7 +904,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 cb326ff7..90473f43 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 55aedc98..db1f077c 100644
--- a/tests/util.py
+++ b/tests/util.py
@@ -1,6 +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))
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-23 22:38:32 +0000