mirror of https://github.com/huashengdun/webssh
Added to_int to utils
parent
de0fda1ae1
commit
88405eddac
|
@ -1,7 +1,9 @@
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
from webssh.utils import (is_valid_ipv4_address, is_valid_ipv6_address,
|
from webssh.utils import (
|
||||||
is_valid_port, is_valid_hostname, to_str, to_bytes)
|
is_valid_ipv4_address, is_valid_ipv6_address, is_valid_port,
|
||||||
|
is_valid_hostname, to_str, to_bytes, to_int
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class TestUitls(unittest.TestCase):
|
class TestUitls(unittest.TestCase):
|
||||||
|
@ -18,6 +20,12 @@ class TestUitls(unittest.TestCase):
|
||||||
self.assertEqual(to_bytes(b), b)
|
self.assertEqual(to_bytes(b), b)
|
||||||
self.assertEqual(to_bytes(u), b)
|
self.assertEqual(to_bytes(u), b)
|
||||||
|
|
||||||
|
def test_to_int(self):
|
||||||
|
self.assertEqual(to_int(''), None)
|
||||||
|
self.assertEqual(to_int(None), None)
|
||||||
|
self.assertEqual(to_int('22'), 22)
|
||||||
|
self.assertEqual(to_int(' 22 '), 22)
|
||||||
|
|
||||||
def test_is_valid_ipv4_address(self):
|
def test_is_valid_ipv4_address(self):
|
||||||
self.assertFalse(is_valid_ipv4_address('127.0.0'))
|
self.assertFalse(is_valid_ipv4_address('127.0.0'))
|
||||||
self.assertFalse(is_valid_ipv4_address(b'127.0.0'))
|
self.assertFalse(is_valid_ipv4_address(b'127.0.0'))
|
||||||
|
|
|
@ -13,7 +13,7 @@ from tornado.ioloop import IOLoop
|
||||||
from webssh.settings import swallow_http_errors
|
from webssh.settings import swallow_http_errors
|
||||||
from webssh.utils import (
|
from webssh.utils import (
|
||||||
is_valid_ipv4_address, is_valid_ipv6_address, is_valid_port,
|
is_valid_ipv4_address, is_valid_ipv6_address, is_valid_port,
|
||||||
is_valid_hostname, to_bytes, to_str, UnicodeType
|
is_valid_hostname, to_bytes, to_str, to_int, UnicodeType
|
||||||
)
|
)
|
||||||
from webssh.worker import Worker, recycle_worker, workers
|
from webssh.worker import Worker, recycle_worker, workers
|
||||||
|
|
||||||
|
@ -52,12 +52,8 @@ class MixinHandler(object):
|
||||||
return # suppose this app doesn't run after an nginx server
|
return # suppose this app doesn't run after an nginx server
|
||||||
|
|
||||||
if is_valid_ipv4_address(ip) or is_valid_ipv6_address(ip):
|
if is_valid_ipv4_address(ip) or is_valid_ipv6_address(ip):
|
||||||
try:
|
port = to_int(port)
|
||||||
port = int(port)
|
if port and is_valid_port(port):
|
||||||
except (TypeError, ValueError):
|
|
||||||
pass
|
|
||||||
else:
|
|
||||||
if is_valid_port(port):
|
|
||||||
return (ip, port)
|
return (ip, port)
|
||||||
|
|
||||||
logging.warning('Bad nginx configuration.')
|
logging.warning('Bad nginx configuration.')
|
||||||
|
@ -154,14 +150,9 @@ class IndexHandler(MixinHandler, tornado.web.RequestHandler):
|
||||||
|
|
||||||
def get_port(self):
|
def get_port(self):
|
||||||
value = self.get_value('port')
|
value = self.get_value('port')
|
||||||
try:
|
port = to_int(value)
|
||||||
port = int(value)
|
if port and is_valid_port(port):
|
||||||
except ValueError:
|
|
||||||
pass
|
|
||||||
else:
|
|
||||||
if is_valid_port(port):
|
|
||||||
return port
|
return port
|
||||||
|
|
||||||
raise InvalidValueError('Invalid port: {}'.format(value))
|
raise InvalidValueError('Invalid port: {}'.format(value))
|
||||||
|
|
||||||
def lookup_hostname(self, hostname, port):
|
def lookup_hostname(self, hostname, port):
|
||||||
|
|
|
@ -23,6 +23,13 @@ def to_bytes(ustr, encoding='utf-8'):
|
||||||
return ustr
|
return ustr
|
||||||
|
|
||||||
|
|
||||||
|
def to_int(string):
|
||||||
|
try:
|
||||||
|
return int(string)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
def is_valid_ipv4_address(ipstr):
|
def is_valid_ipv4_address(ipstr):
|
||||||
ipstr = to_str(ipstr)
|
ipstr = to_str(ipstr)
|
||||||
try:
|
try:
|
||||||
|
|
Loading…
Reference in New Issue