Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
""" Test processing deprecated API attributes """
kwargs = {'ssh_host': '10.0.0.1',
'ssh_address': '10.0.0.1',
'ssh_private_key': 'testrsa.key',
'raise_exception_if_any_forwarder_have_a_problem': True}
for item in kwargs:
self.assertEqual(kwargs[item],
sshtunnel.SSHTunnelForwarder._process_deprecated(
None,
item,
kwargs.copy()
))
# use both deprecated and not None new attribute should raise exception
for item in kwargs:
with self.assertRaises(ValueError):
sshtunnel.SSHTunnelForwarder._process_deprecated('some value',
item,
kwargs.copy())
# deprecated attribute not in deprecation list should raise exception
with self.assertRaises(ValueError):
sshtunnel.SSHTunnelForwarder._process_deprecated('some value',
'item',
kwargs.copy())
def test_process_deprecations(self):
""" Test processing deprecated API attributes """
kwargs = {'ssh_host': '10.0.0.1',
'ssh_address': '10.0.0.1',
'ssh_private_key': 'testrsa.key',
'raise_exception_if_any_forwarder_have_a_problem': True}
for item in kwargs:
self.assertEqual(kwargs[item],
sshtunnel.SSHTunnelForwarder._process_deprecated(
None,
item,
kwargs.copy()
))
# use both deprecated and not None new attribute should raise exception
for item in kwargs:
with self.assertRaises(ValueError):
sshtunnel.SSHTunnelForwarder._process_deprecated('some value',
item,
kwargs.copy())
# deprecated attribute not in deprecation list should raise exception
with self.assertRaises(ValueError):
sshtunnel.SSHTunnelForwarder._process_deprecated('some value',
'item',
kwargs.copy())
for item in kwargs:
self.assertEqual(kwargs[item],
sshtunnel.SSHTunnelForwarder._process_deprecated(
None,
item,
kwargs.copy()
))
# use both deprecated and not None new attribute should raise exception
for item in kwargs:
with self.assertRaises(ValueError):
sshtunnel.SSHTunnelForwarder._process_deprecated('some value',
item,
kwargs.copy())
# deprecated attribute not in deprecation list should raise exception
with self.assertRaises(ValueError):
sshtunnel.SSHTunnelForwarder._process_deprecated('some value',
'item',
kwargs.copy())
local_bind_address=('', LOCAL_PORT)) as server:
def do_something(port):
pass
print("LOCAL PORTS:", server.local_bind_port)
do_something(server.local_bind_port)
"""
# Attach a console handler to the logger or create one if not passed
kwargs['logger'] = create_logger(logger=kwargs.get('logger', None),
loglevel=kwargs.pop('debug_level', None))
ssh_address_or_host = kwargs.pop('ssh_address_or_host', None)
# Check if deprecated arguments ssh_address or ssh_host were used
for deprecated_argument in ['ssh_address', 'ssh_host']:
ssh_address_or_host = SSHTunnelForwarder._process_deprecated(
ssh_address_or_host,
deprecated_argument,
kwargs
)
ssh_port = kwargs.pop('ssh_port', 22)
skip_tunnel_checkup = kwargs.pop('skip_tunnel_checkup', True)
block_on_close = kwargs.pop('block_on_close', _DAEMON)
if not args:
if isinstance(ssh_address_or_host, tuple):
args = (ssh_address_or_host, )
else:
args = ((ssh_address_or_host, ssh_port), )
forwarder = SSHTunnelForwarder(*args, **kwargs)
forwarder.skip_tunnel_checkup = skip_tunnel_checkup
forwarder.daemon_forward_servers = not block_on_close