commit - 5a5c67ceb207c7a537c1b91c54824f2dae4321e7
commit + 842f3e21dc7d86e330729e04f758d990afd0cfae
blob - 5161889d1f48614e12ca7e9bde67b381be7cdf2d
blob + d22895cbcdc7e44fbc1c8c9c1673a3397ba7b965
--- dulwich/contrib/test_paramiko_vendor.py
+++ dulwich/contrib/test_paramiko_vendor.py
"""Tests for paramiko_vendor."""
import socket
-import paramiko
import threading
from dulwich.tests import TestCase
-from dulwich.contrib.paramiko_vendor import ParamikoSSHVendor
from io import StringIO
+from unittest import skipIf
+try:
+ import paramiko
+except ImportError:
+ paramiko = None
+else:
+ from dulwich.contrib.paramiko_vendor import ParamikoSSHVendor
+
USER = 'testuser'
PASSWORD = 'test'
SERVER_KEY = """\
-----END RSA PRIVATE KEY-----"""
-class Server(paramiko.ServerInterface):
- """http://docs.paramiko.org/en/2.4/api/server.html"""
- def __init__(self, commands, *args, **kwargs):
- super(Server, self).__init__(*args, **kwargs)
- self.commands = commands
+if paramiko is not None:
+ class Server(paramiko.ServerInterface):
+ """http://docs.paramiko.org/en/2.4/api/server.html"""
+ def __init__(self, commands, *args, **kwargs):
+ super(Server, self).__init__(*args, **kwargs)
+ self.commands = commands
- def check_channel_exec_request(self, channel, command):
- self.commands.append(command)
- return True
+ def check_channel_exec_request(self, channel, command):
+ self.commands.append(command)
+ return True
- def check_auth_password(self, username, password):
- if username == USER and password == PASSWORD:
- return paramiko.AUTH_SUCCESSFUL
- return paramiko.AUTH_FAILED
+ def check_auth_password(self, username, password):
+ if username == USER and password == PASSWORD:
+ return paramiko.AUTH_SUCCESSFUL
+ return paramiko.AUTH_FAILED
- def check_auth_publickey(self, username, key):
- pubkey = paramiko.RSAKey.from_private_key(StringIO(CLIENT_KEY))
- if username == USER and key == pubkey:
- return paramiko.AUTH_SUCCESSFUL
- return paramiko.AUTH_FAILED
+ def check_auth_publickey(self, username, key):
+ pubkey = paramiko.RSAKey.from_private_key(StringIO(CLIENT_KEY))
+ if username == USER and key == pubkey:
+ return paramiko.AUTH_SUCCESSFUL
+ return paramiko.AUTH_FAILED
- def check_channel_request(self, kind, chanid):
- if kind == "session":
- return paramiko.OPEN_SUCCEEDED
- return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
+ def check_channel_request(self, kind, chanid):
+ if kind == "session":
+ return paramiko.OPEN_SUCCEEDED
+ return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
- def get_allowed_auths(self, username):
- return "password,publickey"
+ def get_allowed_auths(self, username):
+ return "password,publickey"
+@skipIf(paramiko is None, "paramiko is not installed")
class ParamikoSSHVendorTests(TestCase):
def setUp(self):