/
main_test.py
82 lines (64 loc) · 2.25 KB
/
main_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
# Copyright 2018 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import socket
import subprocess
import pytest
import requests
from retrying import retry
import websocket
@pytest.fixture(scope='module')
def server():
"""Provides the address of a test HTTP/websocket server.
The test server is automatically created before
a test and destroyed at the end.
"""
# Ask the OS to allocate a port.
sock = socket.socket()
sock.bind(('127.0.0.1', 0))
port = sock.getsockname()[1]
# Free the port and pass it to a subprocess.
sock.close()
bind_to = '127.0.0.1:{}'.format(port)
server = subprocess.Popen(
['gunicorn', '-b', bind_to, '-k' 'flask_sockets.worker', 'main:app'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
# With btlr, there can be many processes are spawned and the
# server might be in a tight memory situation, so let's wait for 2
# mins.
# Wait until the server responds before proceeding.
@retry(wait_fixed=50, stop_max_delay=120000)
def check_server(url):
requests.get(url)
check_server('http://{}/'.format(bind_to))
yield bind_to
server.kill()
# Dump the logs for debugging
out, err = server.communicate()
print("gunicorn stdout: {}".format(out))
print("gunicorn stderr: {}".format(err))
def test_http(server):
result = requests.get('http://{}/'.format(server))
assert 'Python Websockets Chat' in result.text
def test_websocket(server):
url = 'ws://{}/chat'.format(server)
ws_one = websocket.WebSocket()
ws_one.connect(url)
ws_two = websocket.WebSocket()
ws_two.connect(url)
message = 'Hello, World'
ws_one.send(message)
assert ws_one.recv() == message
assert ws_two.recv() == message