mirror of
https://github.com/sshuttle/sshuttle.git
synced 2024-11-22 16:03:57 +01:00
3bfb975ed9
* re-organized imports according to pep8 * fixed all remaining pep8 issues * moved common config into setup.cfg, additionally test `tests` * removed --select=X -- the errors selected where by default not in flake8's --ignore list so effectively had no effect * update .travis.yml to reflect changes in tox.ini * make travis just use tox in order to avoid code duplaction * replace py.test with pytest * fixed .travis.yml * try different pypy toxenv * hopefully fixed testenv for pypy * added pypy basepython, removed unused python2.6 * install dev package before testing (fixes missing coverage) * fixed empty exception pass blocks with noqa * Added dummy log message on empty try-except-pass blocks to make dodacy happy :( * Replaced Exception with BaseException
202 lines
6.2 KiB
Python
202 lines
6.2 KiB
Python
import sys
|
|
import io
|
|
import socket
|
|
from socket import AF_INET, AF_INET6
|
|
import errno
|
|
|
|
from mock import patch, call
|
|
import sshuttle.helpers
|
|
|
|
|
|
@patch('sshuttle.helpers.logprefix', new='prefix: ')
|
|
@patch('sshuttle.helpers.sys.stdout')
|
|
@patch('sshuttle.helpers.sys.stderr')
|
|
def test_log(mock_stderr, mock_stdout):
|
|
sshuttle.helpers.log("message")
|
|
sshuttle.helpers.log("abc")
|
|
sshuttle.helpers.log("message 1\n")
|
|
sshuttle.helpers.log("message 2\nline2\nline3\n")
|
|
sshuttle.helpers.log("message 3\nline2\nline3")
|
|
assert mock_stdout.mock_calls == [
|
|
call.flush(),
|
|
call.flush(),
|
|
call.flush(),
|
|
call.flush(),
|
|
call.flush(),
|
|
]
|
|
assert mock_stderr.mock_calls == [
|
|
call.write('prefix: message'),
|
|
call.flush(),
|
|
call.write('prefix: abc'),
|
|
call.flush(),
|
|
call.write('prefix: message 1\n'),
|
|
call.flush(),
|
|
call.write('prefix: message 2\n'),
|
|
call.write('---> line2\n'),
|
|
call.write('---> line3\n'),
|
|
call.flush(),
|
|
call.write('prefix: message 3\n'),
|
|
call.write('---> line2\n'),
|
|
call.write('---> line3\n'),
|
|
call.flush(),
|
|
]
|
|
|
|
|
|
@patch('sshuttle.helpers.logprefix', new='prefix: ')
|
|
@patch('sshuttle.helpers.verbose', new=1)
|
|
@patch('sshuttle.helpers.sys.stdout')
|
|
@patch('sshuttle.helpers.sys.stderr')
|
|
def test_debug1(mock_stderr, mock_stdout):
|
|
sshuttle.helpers.debug1("message")
|
|
assert mock_stdout.mock_calls == [
|
|
call.flush(),
|
|
]
|
|
assert mock_stderr.mock_calls == [
|
|
call.write('prefix: message'),
|
|
call.flush(),
|
|
]
|
|
|
|
|
|
@patch('sshuttle.helpers.logprefix', new='prefix: ')
|
|
@patch('sshuttle.helpers.verbose', new=0)
|
|
@patch('sshuttle.helpers.sys.stdout')
|
|
@patch('sshuttle.helpers.sys.stderr')
|
|
def test_debug1_nop(mock_stderr, mock_stdout):
|
|
sshuttle.helpers.debug1("message")
|
|
assert mock_stdout.mock_calls == []
|
|
assert mock_stderr.mock_calls == []
|
|
|
|
|
|
@patch('sshuttle.helpers.logprefix', new='prefix: ')
|
|
@patch('sshuttle.helpers.verbose', new=2)
|
|
@patch('sshuttle.helpers.sys.stdout')
|
|
@patch('sshuttle.helpers.sys.stderr')
|
|
def test_debug2(mock_stderr, mock_stdout):
|
|
sshuttle.helpers.debug2("message")
|
|
assert mock_stdout.mock_calls == [
|
|
call.flush(),
|
|
]
|
|
assert mock_stderr.mock_calls == [
|
|
call.write('prefix: message'),
|
|
call.flush(),
|
|
]
|
|
|
|
|
|
@patch('sshuttle.helpers.logprefix', new='prefix: ')
|
|
@patch('sshuttle.helpers.verbose', new=1)
|
|
@patch('sshuttle.helpers.sys.stdout')
|
|
@patch('sshuttle.helpers.sys.stderr')
|
|
def test_debug2_nop(mock_stderr, mock_stdout):
|
|
sshuttle.helpers.debug2("message")
|
|
assert mock_stdout.mock_calls == []
|
|
assert mock_stderr.mock_calls == []
|
|
|
|
|
|
@patch('sshuttle.helpers.logprefix', new='prefix: ')
|
|
@patch('sshuttle.helpers.verbose', new=3)
|
|
@patch('sshuttle.helpers.sys.stdout')
|
|
@patch('sshuttle.helpers.sys.stderr')
|
|
def test_debug3(mock_stderr, mock_stdout):
|
|
sshuttle.helpers.debug3("message")
|
|
assert mock_stdout.mock_calls == [
|
|
call.flush(),
|
|
]
|
|
assert mock_stderr.mock_calls == [
|
|
call.write('prefix: message'),
|
|
call.flush(),
|
|
]
|
|
|
|
|
|
@patch('sshuttle.helpers.logprefix', new='prefix: ')
|
|
@patch('sshuttle.helpers.verbose', new=2)
|
|
@patch('sshuttle.helpers.sys.stdout')
|
|
@patch('sshuttle.helpers.sys.stderr')
|
|
def test_debug3_nop(mock_stderr, mock_stdout):
|
|
sshuttle.helpers.debug3("message")
|
|
assert mock_stdout.mock_calls == []
|
|
assert mock_stderr.mock_calls == []
|
|
|
|
|
|
@patch('sshuttle.helpers.open', create=True)
|
|
def test_resolvconf_nameservers(mock_open):
|
|
mock_open.return_value = io.StringIO(u"""
|
|
# Generated by NetworkManager
|
|
search pri
|
|
nameserver 192.168.1.1
|
|
nameserver 192.168.2.1
|
|
nameserver 192.168.3.1
|
|
nameserver 192.168.4.1
|
|
nameserver 2404:6800:4004:80c::1
|
|
nameserver 2404:6800:4004:80c::2
|
|
nameserver 2404:6800:4004:80c::3
|
|
nameserver 2404:6800:4004:80c::4
|
|
""")
|
|
|
|
ns = sshuttle.helpers.resolvconf_nameservers()
|
|
assert ns == [
|
|
(AF_INET, u'192.168.1.1'), (AF_INET, u'192.168.2.1'),
|
|
(AF_INET, u'192.168.3.1'), (AF_INET, u'192.168.4.1'),
|
|
(AF_INET6, u'2404:6800:4004:80c::1'),
|
|
(AF_INET6, u'2404:6800:4004:80c::2'),
|
|
(AF_INET6, u'2404:6800:4004:80c::3'),
|
|
(AF_INET6, u'2404:6800:4004:80c::4')
|
|
]
|
|
|
|
|
|
@patch('sshuttle.helpers.open', create=True)
|
|
def test_resolvconf_random_nameserver(mock_open):
|
|
mock_open.return_value = io.StringIO(u"""
|
|
# Generated by NetworkManager
|
|
search pri
|
|
nameserver 192.168.1.1
|
|
nameserver 192.168.2.1
|
|
nameserver 192.168.3.1
|
|
nameserver 192.168.4.1
|
|
nameserver 2404:6800:4004:80c::1
|
|
nameserver 2404:6800:4004:80c::2
|
|
nameserver 2404:6800:4004:80c::3
|
|
nameserver 2404:6800:4004:80c::4
|
|
""")
|
|
ns = sshuttle.helpers.resolvconf_random_nameserver()
|
|
assert ns in [
|
|
(AF_INET, u'192.168.1.1'), (AF_INET, u'192.168.2.1'),
|
|
(AF_INET, u'192.168.3.1'), (AF_INET, u'192.168.4.1'),
|
|
(AF_INET6, u'2404:6800:4004:80c::1'),
|
|
(AF_INET6, u'2404:6800:4004:80c::2'),
|
|
(AF_INET6, u'2404:6800:4004:80c::3'),
|
|
(AF_INET6, u'2404:6800:4004:80c::4')
|
|
]
|
|
|
|
|
|
@patch('sshuttle.helpers.socket.socket.bind')
|
|
def test_islocal(mock_bind):
|
|
bind_error = socket.error(errno.EADDRNOTAVAIL)
|
|
mock_bind.side_effect = [None, bind_error, None, bind_error]
|
|
|
|
assert sshuttle.helpers.islocal("127.0.0.1", AF_INET)
|
|
assert not sshuttle.helpers.islocal("192.0.2.1", AF_INET)
|
|
assert sshuttle.helpers.islocal("::1", AF_INET6)
|
|
assert not sshuttle.helpers.islocal("2001:db8::1", AF_INET6)
|
|
|
|
|
|
def test_family_ip_tuple():
|
|
assert sshuttle.helpers.family_ip_tuple("127.0.0.1") \
|
|
== (AF_INET, "127.0.0.1")
|
|
assert sshuttle.helpers.family_ip_tuple("192.168.2.6") \
|
|
== (AF_INET, "192.168.2.6")
|
|
assert sshuttle.helpers.family_ip_tuple("::1") \
|
|
== (AF_INET6, "::1")
|
|
assert sshuttle.helpers.family_ip_tuple("2404:6800:4004:80c::1") \
|
|
== (AF_INET6, "2404:6800:4004:80c::1")
|
|
|
|
|
|
def test_family_to_string():
|
|
assert sshuttle.helpers.family_to_string(AF_INET) == "AF_INET"
|
|
assert sshuttle.helpers.family_to_string(AF_INET6) == "AF_INET6"
|
|
if sys.version_info < (3, 0):
|
|
expected = "1"
|
|
assert sshuttle.helpers.family_to_string(socket.AF_UNIX) == "1"
|
|
else:
|
|
expected = 'AddressFamily.AF_UNIX'
|
|
assert sshuttle.helpers.family_to_string(socket.AF_UNIX) == expected
|