mirror of https://github.com/huashengdun/webssh
Refactoring
parent
049baad909
commit
fc30ead69e
|
@ -65,7 +65,7 @@ class Server(paramiko.ServerInterface):
|
||||||
self.key_verified = False
|
self.key_verified = False
|
||||||
|
|
||||||
def get_cmd2enc(self, encodings):
|
def get_cmd2enc(self, encodings):
|
||||||
while len(encodings) < 2:
|
while len(encodings) < len(self.commands):
|
||||||
encodings.append(random.choice(self.encodings))
|
encodings.append(random.choice(self.encodings))
|
||||||
return dict(zip(self.commands, encodings[0:2]))
|
return dict(zip(self.commands, encodings[0:2]))
|
||||||
|
|
||||||
|
|
|
@ -7,7 +7,7 @@ import tornado.gen
|
||||||
from tornado.testing import AsyncHTTPTestCase
|
from tornado.testing import AsyncHTTPTestCase
|
||||||
from tornado.httpclient import HTTPError
|
from tornado.httpclient import HTTPError
|
||||||
from tornado.options import options
|
from tornado.options import options
|
||||||
from tests.sshserver import run_ssh_server, banner
|
from tests.sshserver import run_ssh_server, banner, Server
|
||||||
from tests.utils import encode_multipart_formdata, read_file, make_tests_data_path # noqa
|
from tests.utils import encode_multipart_formdata, read_file, make_tests_data_path # noqa
|
||||||
from webssh import handler
|
from webssh import handler
|
||||||
from webssh.main import make_app, make_handlers
|
from webssh.main import make_app, make_handlers
|
||||||
|
@ -25,6 +25,7 @@ except ImportError:
|
||||||
|
|
||||||
handler.DELAY = 0.1
|
handler.DELAY = 0.1
|
||||||
swallow_http_errors = handler.swallow_http_errors
|
swallow_http_errors = handler.swallow_http_errors
|
||||||
|
server_encodings = {e.strip() for e in Server.encodings}
|
||||||
|
|
||||||
|
|
||||||
class TestAppBase(AsyncHTTPTestCase):
|
class TestAppBase(AsyncHTTPTestCase):
|
||||||
|
@ -775,7 +776,7 @@ class TestAppWithBadEncoding(OtherTestBase):
|
||||||
response = yield self.async_post('/', self.body)
|
response = yield self.async_post('/', self.body)
|
||||||
dic = json.loads(to_str(response.body))
|
dic = json.loads(to_str(response.body))
|
||||||
self.assert_status_none(dic)
|
self.assert_status_none(dic)
|
||||||
self.assertIn(dic['encoding'], ['UTF-8', 'GBK'])
|
self.assertIn(dic['encoding'], server_encodings)
|
||||||
|
|
||||||
|
|
||||||
class TestAppWithUnknownEncoding(OtherTestBase):
|
class TestAppWithUnknownEncoding(OtherTestBase):
|
||||||
|
|
|
@ -392,7 +392,7 @@ class IndexHandler(MixinHandler, tornado.web.RequestHandler):
|
||||||
|
|
||||||
def parse_encoding(self, data):
|
def parse_encoding(self, data):
|
||||||
try:
|
try:
|
||||||
encoding = to_str(data, 'ascii')
|
encoding = to_str(data.strip(), 'ascii')
|
||||||
except UnicodeDecodeError:
|
except UnicodeDecodeError:
|
||||||
return
|
return
|
||||||
|
|
||||||
|
@ -407,8 +407,8 @@ class IndexHandler(MixinHandler, tornado.web.RequestHandler):
|
||||||
|
|
||||||
for command in commands:
|
for command in commands:
|
||||||
_, stdout, _ = ssh.exec_command(command, get_pty=True)
|
_, stdout, _ = ssh.exec_command(command, get_pty=True)
|
||||||
data = stdout.read().strip()
|
data = stdout.read()
|
||||||
logging.debug('encoding: {}'.format(data))
|
logging.debug('{!r} => {!r}'.format(command, data))
|
||||||
result = self.parse_encoding(data)
|
result = self.parse_encoding(data)
|
||||||
if result:
|
if result:
|
||||||
return result
|
return result
|
||||||
|
|
|
@ -54,9 +54,9 @@ def is_valid_port(port):
|
||||||
return 0 < port < 65536
|
return 0 < port < 65536
|
||||||
|
|
||||||
|
|
||||||
def is_valid_encoding(encoding, ustr=u'test'):
|
def is_valid_encoding(encoding):
|
||||||
try:
|
try:
|
||||||
ustr.encode(encoding)
|
u'test'.encode(encoding)
|
||||||
except LookupError:
|
except LookupError:
|
||||||
return False
|
return False
|
||||||
return True
|
return True
|
||||||
|
|
Loading…
Reference in New Issue