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 09:49:48 +08:00
parent e88ecb2879
commit de0a903de8
2 changed files with 16 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tests/sshserver.py
View File

@ -77,11 +77,11 @@ class Server (paramiko.ServerInterface):
return True
def run_ssh_server(running=True):
def run_ssh_server(port=2200, running=True):
# now connect
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('127.0.0.1', 2200))
sock.bind(('127.0.0.1', port))
sock.listen(100)
while running:

27
tests/test_app.py
View File

@ -18,6 +18,8 @@ handler.DELAY = 0.1
class TestApp(AsyncHTTPTestCase):
_is_running = False
sshserver_port = 2200
body = u'hostname=127.0.0.1&port={}&username=robey&password=foo'.format(sshserver_port) # noqa
def get_app(self):
loop = self.io_loop
@ -28,6 +30,14 @@ class TestApp(AsyncHTTPTestCase):
app = make_app(make_handlers(loop, options), get_app_settings(options))
return app
@classmethod
def setUpClass(cls):
t = threading.Thread(
target=run_ssh_server, args=(cls.sshserver_port, cls)
)
t.setDaemon(True)
t.start()
@classmethod
def tearDownClass(cls):
cls._is_running = True
@ -62,15 +72,13 @@ class TestApp(AsyncHTTPTestCase):
def test_app_with_wrong_credentials(self):
response = self.fetch('/')
self.assertEqual(response.code, 200)
body = u'hostname=127.0.0.1&port=2200&username=robey&password=foos'
response = self.fetch('/', method="POST", body=body)
response = self.fetch('/', method="POST", body=self.body + u's')
self.assertIn(b'Authentication failed.', response.body)
def test_app_with_correct_credentials(self):
response = self.fetch('/')
self.assertEqual(response.code, 200)
body = u'hostname=127.0.0.1&port=2200&username=robey&password=foo'
response = self.fetch('/', method="POST", body=body)
response = self.fetch('/', method="POST", body=self.body)
worker_id = json.loads(response.body.decode('utf-8'))['id']
self.assertIsNotNone(worker_id)
@ -81,8 +89,7 @@ class TestApp(AsyncHTTPTestCase):
response = yield client.fetch(url)
self.assertEqual(response.code, 200)
body = u'hostname=127.0.0.1&port=2200&username=robey&password=foo'
response = yield client.fetch(url, method="POST", body=body)
response = yield client.fetch(url, method="POST", body=self.body)
worker_id = json.loads(response.body.decode('utf-8'))['id']
self.assertIsNotNone(worker_id)
@ -101,8 +108,7 @@ class TestApp(AsyncHTTPTestCase):
response = yield client.fetch(url)
self.assertEqual(response.code, 200)
body = u'hostname=127.0.0.1&port=2200&username=robey&password=foo'
response = yield client.fetch(url, method="POST", body=body)
response = yield client.fetch(url, method="POST", body=self.body)
worker_id = json.loads(response.body.decode('utf-8'))['id']
self.assertIsNotNone(worker_id)
@ -112,8 +118,3 @@ class TestApp(AsyncHTTPTestCase):
msg = yield ws.read_message()
self.assertIn('Welcome!', msg)
ws.close()
t = threading.Thread(target=run_ssh_server, args=(TestApp,))
t.setDaemon(True)
t.start()