mirror of https://github.com/huashengdun/webssh
Modified test_app.py sshserver.py
parent
20be29f989
commit
e88ecb2879
|
@ -77,14 +77,14 @@ class Server (paramiko.ServerInterface):
|
|||
return True
|
||||
|
||||
|
||||
def run_ssh_server(app):
|
||||
def run_ssh_server(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.listen(100)
|
||||
|
||||
while not app._tear_down:
|
||||
while running:
|
||||
client, addr = sock.accept()
|
||||
print('Got a connection!')
|
||||
|
||||
|
@ -120,4 +120,4 @@ def run_ssh_server(app):
|
|||
|
||||
|
||||
if __name__ == '__main__':
|
||||
run_ssh_server(False)
|
||||
run_ssh_server()
|
||||
|
|
|
@ -17,11 +17,10 @@ handler.DELAY = 0.1
|
|||
|
||||
class TestApp(AsyncHTTPTestCase):
|
||||
|
||||
_tear_down = False
|
||||
_is_running = False
|
||||
|
||||
def get_app(self):
|
||||
loop = self.io_loop
|
||||
self._tear_down = False
|
||||
options.debug = True
|
||||
options.policy = random.choice(['warning', 'autoadd'])
|
||||
options.hostFile = ''
|
||||
|
@ -31,7 +30,11 @@ class TestApp(AsyncHTTPTestCase):
|
|||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls._tear_down = True
|
||||
cls._is_running = True
|
||||
|
||||
@classmethod
|
||||
def __bool__(cls):
|
||||
return cls._is_running
|
||||
|
||||
def test_app_with_invalid_form(self):
|
||||
response = self.fetch('/')
|
||||
|
|
Loading…
Reference in New Issue