test_20_websockets.py (6743B)
1 #!/usr/bin/env python3 2 # -*- coding: utf-8 -*- 3 #*************************************************************************** 4 # _ _ ____ _ 5 # Project ___| | | | _ \| | 6 # / __| | | | |_) | | 7 # | (__| |_| | _ <| |___ 8 # \___|\___/|_| \_\_____| 9 # 10 # Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al. 11 # 12 # This software is licensed as described in the file COPYING, which 13 # you should have received as part of this distribution. The terms 14 # are also available at https://curl.se/docs/copyright.html. 15 # 16 # You may opt to use, copy, modify, merge, publish, distribute and/or sell 17 # copies of the Software, and permit persons to whom the Software is 18 # furnished to do so, under the terms of the COPYING file. 19 # 20 # This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY 21 # KIND, either express or implied. 22 # 23 # SPDX-License-Identifier: curl 24 # 25 ########################################################################### 26 # 27 import logging 28 import os 29 import shutil 30 import socket 31 import subprocess 32 import time 33 from datetime import datetime, timedelta 34 from typing import Dict 35 import pytest 36 37 from testenv import Env, CurlClient, LocalClient 38 from testenv.ports import alloc_ports_and_do 39 40 41 log = logging.getLogger(__name__) 42 43 44 @pytest.mark.skipif(condition=not Env.curl_has_protocol('ws'), 45 reason='curl lacks ws protocol support') 46 class TestWebsockets: 47 48 PORT_SPECS = { 49 'ws': socket.SOCK_STREAM, 50 } 51 52 def check_alive(self, env, port, timeout=Env.SERVER_TIMEOUT): 53 curl = CurlClient(env=env) 54 url = f'http://localhost:{port}/' 55 end = datetime.now() + timedelta(seconds=timeout) 56 while datetime.now() < end: 57 r = curl.http_download(urls=[url]) 58 if r.exit_code == 0: 59 return True 60 time.sleep(.1) 61 return False 62 63 def _mkpath(self, path): 64 if not os.path.exists(path): 65 return os.makedirs(path) 66 67 def _rmrf(self, path): 68 if os.path.exists(path): 69 return shutil.rmtree(path) 70 71 @pytest.fixture(autouse=True, scope='class') 72 def ws_echo(self, env): 73 self.run_dir = os.path.join(env.gen_dir, 'ws_echo_server') 74 err_file = os.path.join(self.run_dir, 'stderr') 75 self._rmrf(self.run_dir) 76 self._mkpath(self.run_dir) 77 self.cmd = os.path.join(env.project_dir, 78 'tests/http/testenv/ws_echo_server.py') 79 self.wsproc = None 80 self.cerr = None 81 82 def startup(ports: Dict[str, int]) -> bool: 83 wargs = [self.cmd, '--port', str(ports['ws'])] 84 log.info(f'start_ {wargs}') 85 self.wsproc = subprocess.Popen(args=wargs, 86 cwd=self.run_dir, 87 stderr=self.cerr, 88 stdout=self.cerr) 89 if self.check_alive(env, ports['ws']): 90 env.update_ports(ports) 91 return True 92 log.error(f'not alive {wargs}') 93 self.wsproc.terminate() 94 self.wsproc = None 95 return False 96 97 with open(err_file, 'w') as self.cerr: 98 assert alloc_ports_and_do(TestWebsockets.PORT_SPECS, startup, 99 env.gen_root, max_tries=3) 100 assert self.wsproc 101 yield 102 self.wsproc.terminate() 103 104 def test_20_01_basic(self, env: Env, ws_echo): 105 curl = CurlClient(env=env) 106 url = f'http://localhost:{env.ws_port}/' 107 r = curl.http_download(urls=[url]) 108 r.check_response(http_status=426) 109 110 def test_20_02_pingpong_small(self, env: Env, ws_echo): 111 payload = 125 * "x" 112 client = LocalClient(env=env, name='ws_pingpong') 113 if not client.exists(): 114 pytest.skip(f'example client not built: {client.name}') 115 url = f'ws://localhost:{env.ws_port}/' 116 r = client.run(args=[url, payload]) 117 r.check_exit_code(0) 118 119 # the python websocket server does not like 'large' control frames 120 def test_20_03_pingpong_too_large(self, env: Env, ws_echo): 121 payload = 127 * "x" 122 client = LocalClient(env=env, name='ws_pingpong') 123 if not client.exists(): 124 pytest.skip(f'example client not built: {client.name}') 125 url = f'ws://localhost:{env.ws_port}/' 126 r = client.run(args=[url, payload]) 127 r.check_exit_code(100) # CURLE_TOO_LARGE 128 129 def test_20_04_data_small(self, env: Env, ws_echo): 130 client = LocalClient(env=env, name='ws_data') 131 if not client.exists(): 132 pytest.skip(f'example client not built: {client.name}') 133 url = f'ws://localhost:{env.ws_port}/' 134 r = client.run(args=['-m', str(0), '-M', str(10), url]) 135 r.check_exit_code(0) 136 137 def test_20_05_data_med(self, env: Env, ws_echo): 138 client = LocalClient(env=env, name='ws_data') 139 if not client.exists(): 140 pytest.skip(f'example client not built: {client.name}') 141 url = f'ws://localhost:{env.ws_port}/' 142 r = client.run(args=['-m', str(120), '-M', str(130), url]) 143 r.check_exit_code(0) 144 145 def test_20_06_data_large(self, env: Env, ws_echo): 146 client = LocalClient(env=env, name='ws_data') 147 if not client.exists(): 148 pytest.skip(f'example client not built: {client.name}') 149 url = f'ws://localhost:{env.ws_port}/' 150 r = client.run(args=['-m', str(65535 - 5), '-M', str(65535 + 5), url]) 151 r.check_exit_code(0) 152 153 def test_20_07_data_large_small_recv(self, env: Env, ws_echo): 154 run_env = os.environ.copy() 155 run_env['CURL_WS_CHUNK_SIZE'] = '1024' 156 client = LocalClient(env=env, name='ws_data', run_env=run_env) 157 if not client.exists(): 158 pytest.skip(f'example client not built: {client.name}') 159 url = f'ws://localhost:{env.ws_port}/' 160 r = client.run(args=['-m', str(65535 - 5), '-M', str(65535 + 5), url]) 161 r.check_exit_code(0) 162 163 # Send large frames and simulate send blocking on 8192 bytes chunks 164 # Simlates error reported in #15865 165 def test_20_08_data_very_large(self, env: Env, ws_echo): 166 run_env = os.environ.copy() 167 run_env['CURL_WS_CHUNK_EAGAIN'] = '8192' 168 client = LocalClient(env=env, name='ws_data', run_env=run_env) 169 if not client.exists(): 170 pytest.skip(f'example client not built: {client.name}') 171 url = f'ws://localhost:{env.ws_port}/' 172 count = 10 173 large = 20000 174 r = client.run(args=['-c', str(count), '-m', str(large), url]) 175 r.check_exit_code(0)