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