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# 27import logging 28import os 29import pytest 30 31from testenv import Env, CurlClient 32 33 34log = logging.getLogger(__name__) 35 36 37class TestInfo: 38 39 @pytest.fixture(autouse=True, scope='class') 40 def _class_scope(self, env, httpd, nghttpx): 41 if env.have_h3(): 42 nghttpx.start_if_needed() 43 httpd.clear_extra_configs() 44 httpd.reload() 45 46 @pytest.fixture(autouse=True, scope='class') 47 def _class_scope(self, env, httpd): 48 indir = httpd.docs_dir 49 env.make_data_file(indir=indir, fname="data-10k", fsize=10*1024) 50 env.make_data_file(indir=indir, fname="data-100k", fsize=100*1024) 51 env.make_data_file(indir=indir, fname="data-1m", fsize=1024*1024) 52 53 # download plain file 54 @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3']) 55 def test_16_01_info_download(self, env: Env, httpd, nghttpx, repeat, proto): 56 if proto == 'h3' and not env.have_h3(): 57 pytest.skip("h3 not supported") 58 count = 2 59 curl = CurlClient(env=env) 60 url = f'https://{env.authority_for(env.domain1, proto)}/data.json?[0-{count-1}]' 61 r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True) 62 r.check_stats(count=count, http_status=200, exitcode=0, 63 remote_port=env.port_for(alpn_proto=proto), 64 remote_ip='127.0.0.1') 65 for idx, s in enumerate(r.stats): 66 self.check_stat(idx, s, r, dl_size=30, ul_size=0) 67 68 # download plain file with a 302 redirect 69 @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3']) 70 def test_16_02_info_302_download(self, env: Env, httpd, nghttpx, repeat, proto): 71 if proto == 'h3' and not env.have_h3(): 72 pytest.skip("h3 not supported") 73 count = 2 74 curl = CurlClient(env=env) 75 url = f'https://{env.authority_for(env.domain1, proto)}/data.json.302?[0-{count-1}]' 76 r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True, extra_args=[ 77 '--location' 78 ]) 79 r.check_stats(count=count, http_status=200, exitcode=0, 80 remote_port=env.port_for(alpn_proto=proto), 81 remote_ip='127.0.0.1') 82 for idx, s in enumerate(r.stats): 83 self.check_stat(idx, s, r, dl_size=30, ul_size=0) 84 85 @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3']) 86 def test_16_03_info_upload(self, env: Env, httpd, nghttpx, proto, repeat): 87 if proto == 'h3' and not env.have_h3(): 88 pytest.skip("h3 not supported") 89 count = 2 90 fdata = os.path.join(env.gen_dir, 'data-100k') 91 fsize = 100 * 1024 92 curl = CurlClient(env=env) 93 url = f'https://{env.authority_for(env.domain1, proto)}/curltest/echo?id=[0-{count-1}]' 94 r = curl.http_upload(urls=[url], data=f'@{fdata}', alpn_proto=proto, 95 with_headers=True, extra_args=[ 96 '--trace-config', 'http/2,http/3' 97 ]) 98 r.check_response(count=count, http_status=200) 99 r.check_stats(count=count, http_status=200, exitcode=0, 100 remote_port=env.port_for(alpn_proto=proto), 101 remote_ip='127.0.0.1') 102 for idx, s in enumerate(r.stats): 103 self.check_stat(idx, s, r, dl_size=fsize, ul_size=fsize) 104 105 # download plain file via http: ('time_appconnect' is 0) 106 @pytest.mark.parametrize("proto", ['http/1.1']) 107 def test_16_04_info_http_download(self, env: Env, httpd, nghttpx, repeat, proto): 108 count = 2 109 curl = CurlClient(env=env) 110 url = f'http://{env.domain1}:{env.http_port}/data.json?[0-{count-1}]' 111 r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True) 112 r.check_stats(count=count, http_status=200, exitcode=0, 113 remote_port=env.http_port, remote_ip='127.0.0.1') 114 for idx, s in enumerate(r.stats): 115 self.check_stat(idx, s, r, dl_size=30, ul_size=0) 116 117 def check_stat(self, idx, s, r, dl_size=None, ul_size=None): 118 self.check_stat_times(s, idx) 119 # we always send something 120 self.check_stat_positive(s, idx, 'size_request') 121 # we always receive response headers 122 self.check_stat_positive(s, idx, 'size_header') 123 if ul_size is not None: 124 assert s['size_upload'] == ul_size, f'stat #{idx}\n{r.dump_logs()}' # the file we sent 125 assert s['size_request'] >= s['size_upload'], \ 126 f'stat #{idx}, "size_request" smaller than "size_upload", {s}\n{r.dump_logs()}' 127 if dl_size is not None: 128 assert s['size_download'] == dl_size, f'stat #{idx}\n{r.dump_logs()}' # the file we received 129 130 def check_stat_positive(self, s, idx, key): 131 assert key in s, f'stat #{idx} "{key}" missing: {s}' 132 assert s[key] > 0, f'stat #{idx} "{key}" not positive: {s}' 133 134 def check_stat_zero(self, s, key): 135 assert key in s, f'stat "{key}" missing: {s}' 136 assert s[key] == 0, f'stat "{key}" not zero: {s}' 137 138 def check_stat_times(self, s, idx): 139 # check timings reported on a transfer for consistency 140 url = s['url_effective'] 141 # all stat keys which reporting timings 142 all_keys = { 143 'time_appconnect', 'time_connect', 'time_redirect', 144 'time_pretransfer', 'time_starttransfer', 'time_total' 145 } 146 # stat keys where we expect a positive value 147 pos_keys = {'time_pretransfer', 'time_starttransfer', 'time_total'} 148 if s['num_connects'] > 0: 149 pos_keys.add('time_connect') 150 if url.startswith('https:'): 151 pos_keys.add('time_appconnect') 152 if s['num_redirects'] > 0: 153 pos_keys.add('time_redirect') 154 zero_keys = all_keys - pos_keys 155 # assert all zeros are zeros and the others are positive 156 for key in zero_keys: 157 self.check_stat_zero(s, key) 158 for key in pos_keys: 159 self.check_stat_positive(s, idx, key) 160 # assert that all timers before "time_pretransfer" are less or equal 161 for key in ['time_appconnect', 'time_connect', 'time_namelookup']: 162 assert s[key] < s['time_pretransfer'], f'time "{key}" larger than' \ 163 f'"time_pretransfer": {s}' 164 # assert transfer start is after pretransfer 165 assert s['time_pretransfer'] <= s['time_starttransfer'], f'"time_pretransfer" '\ 166 f'greater than "time_starttransfer", {s}' 167 # assert that transfer start is before total 168 assert s['time_starttransfer'] <= s['time_total'], f'"time_starttransfer" '\ 169 f'greater than "time_total", {s}' 170
