How to use the sshtunnel.SSHTunnelForwarder._process_deprecated function in sshtunnel

To help you get started, we’ve selected a few sshtunnel examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github pahaz / sshtunnel / tests / test_forwarder.py View on Github external
""" 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())
github pahaz / sshtunnel / tests / test_forwarder.py View on Github external
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())
github pahaz / sshtunnel / tests / test_forwarder.py View on Github external
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())
github pahaz / sshtunnel / sshtunnel.py View on Github external
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