jinql
/
webssh
mirror of https://github.com/huashengdun/webssh
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Updated test_app.py and sshserver.py

pull/12/merge
Sheng 2018-04-27 14:34:06 +08:00
parent de0a903de8
commit 330145484d
2 changed files with 27 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
tests/sshserver.py
View File

@ -55,7 +55,7 @@ class Server (paramiko.ServerInterface):
return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
def check_auth_password(self, username, password): def check_auth_password(self, username, password):
if (username == 'robey') and (password == 'foo'): if (username in ['robey', 'bar']) and (password == 'foo'):
return paramiko.AUTH_SUCCESSFUL return paramiko.AUTH_SUCCESSFUL
return paramiko.AUTH_FAILED return paramiko.AUTH_FAILED
@ -104,7 +104,8 @@ def run_ssh_server(port=2200, running=True):
print('*** No channel.') print('*** No channel.')
continue continue
print('Authenticated!') username = t.get_username()
print('{} Authenticated!'.format(username))
server.event.wait(10) server.event.wait(10)
if not server.event.is_set(): if not server.event.is_set():
@ -112,6 +113,9 @@ def run_ssh_server(port=2200, running=True):
continue continue
chan.send('\r\n\r\nWelcome!\r\n\r\n') chan.send('\r\n\r\nWelcome!\r\n\r\n')
if username == 'bar':
print(chan.recv(1024))
chan.close()
try: try:
sock.close() sock.close()

22
tests/test_app.py
View File

@ -102,7 +102,7 @@ class TestApp(AsyncHTTPTestCase):
ws.close() ws.close()
@tornado.testing.gen_test @tornado.testing.gen_test
def test_app_with_correct_credentials_welcome(self): def test_app_with_correct_credentials_user_robey(self):
url = self.get_url('/') url = self.get_url('/')
client = self.get_http_client() client = self.get_http_client()
response = yield client.fetch(url) response = yield client.fetch(url)
@ -118,3 +118,23 @@ class TestApp(AsyncHTTPTestCase):
msg = yield ws.read_message() msg = yield ws.read_message()
self.assertIn('Welcome!', msg) self.assertIn('Welcome!', msg)
ws.close() ws.close()
@tornado.testing.gen_test
def test_app_with_correct_credentials_user_bar(self):
url = self.get_url('/')
client = self.get_http_client()
response = yield client.fetch(url)
self.assertEqual(response.code, 200)
body = self.body.replace('robey', 'bar')
response = yield client.fetch(url, method="POST", body=body)
worker_id = json.loads(response.body.decode('utf-8'))['id']
self.assertIsNotNone(worker_id)
url = url.replace('http', 'ws')
ws_url = url + 'ws?id=' + worker_id
ws = yield tornado.websocket.websocket_connect(ws_url)
msg = yield ws.read_message()
self.assertIn('Welcome!', msg)
yield ws.write_message('bye')
ws.close()