Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
freebsd
GitHub Repository: freebsd/pkg
 Path: blob/main/external/curl/tests/http/test_04_stuttered.py
2066 views
1
#!/usr/bin/env python3

2
# -*- coding: utf-8 -*-

3
#***************************************************************************

4
#                                  _   _ ____  _

5
#  Project                     ___| | | |  _ \| |

6
#                             / __| | | | |_) | |

7
#                            | (__| |_| |  _ <| |___

8
#                             \___|\___/|_| \_\_____|

9
#

10
# Copyright (C) Daniel Stenberg, <[email protected]>, 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
from typing import Tuple, List, Dict

29
import pytest

30


31
from testenv import Env, CurlClient

32


33


34
log = logging.getLogger(__name__)

35


36


37
@pytest.mark.skipif(condition=Env().slow_network, reason="not suitable for slow network tests")

38
@pytest.mark.skipif(condition=Env().ci_run, reason="not suitable for CI runs")

39
class TestStuttered:

40


41
    # download 1 file, check that delayed response works in general

42
    @pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])

43
    def test_04_01_download_1(self, env: Env, httpd, nghttpx, proto):

44
        if proto == 'h3' and not env.have_h3():

45
            pytest.skip("h3 not supported")

46
        count = 1

47
        curl = CurlClient(env=env)

48
        urln = f'https://{env.authority_for(env.domain1, proto)}' \

49
            f'/curltest/tweak?id=[0-{count - 1}]'\

50
            '&chunks=100&chunk_size=100&chunk_delay=10ms'

51
        r = curl.http_download(urls=[urln], alpn_proto=proto)

52
        r.check_response(count=1, http_status=200)

53


54
    # download 50 files in 100 chunks a 100 bytes with 10ms delay between

55
    # prepend 100 file requests to warm up connection processing limits

56
    # (Apache2 increases # of parallel processed requests after successes)

57
    @pytest.mark.parametrize("proto", ['h2', 'h3'])

58
    def test_04_02_100_100_10(self, env: Env, httpd, nghttpx, proto):

59
        if proto == 'h3' and not env.have_h3():

60
            pytest.skip("h3 not supported")

61
        count = 50

62
        warmups = 100

63
        curl = CurlClient(env=env)

64
        url1 = f'https://{env.authority_for(env.domain1, proto)}/data.json?[0-{warmups-1}]'

65
        urln = f'https://{env.authority_for(env.domain1, proto)}' \

66
            f'/curltest/tweak?id=[0-{count-1}]'\

67
            '&chunks=100&chunk_size=100&chunk_delay=10ms'

68
        r = curl.http_download(urls=[url1, urln], alpn_proto=proto,

69
                               extra_args=['--parallel'])

70
        r.check_response(count=warmups+count, http_status=200)

71
        assert r.total_connects == 1

72
        t_avg, i_min, t_min, i_max, t_max = self.stats_spread(r.stats[warmups:], 'time_total')

73
        if t_max < (5 * t_min) and t_min < 2:

74
            log.warning(f'avg time of transfer: {t_avg} [{i_min}={t_min}, {i_max}={t_max}]')

75


76
    # download 50 files in 1000 chunks a 10 bytes with 1ms delay between

77
    # prepend 100 file requests to warm up connection processing limits

78
    # (Apache2 increases # of parallel processed requests after successes)

79
    @pytest.mark.parametrize("proto", ['h2', 'h3'])

80
    def test_04_03_1000_10_1(self, env: Env, httpd, nghttpx, proto):

81
        if proto == 'h3' and not env.have_h3():

82
            pytest.skip("h3 not supported")

83
        count = 50

84
        warmups = 100

85
        curl = CurlClient(env=env)

86
        url1 = f'https://{env.authority_for(env.domain1, proto)}/data.json?[0-{warmups-1}]'

87
        urln = f'https://{env.authority_for(env.domain1, proto)}' \

88
            f'/curltest/tweak?id=[0-{count - 1}]'\

89
            '&chunks=1000&chunk_size=10&chunk_delay=100us'

90
        r = curl.http_download(urls=[url1, urln], alpn_proto=proto,

91
                               extra_args=['--parallel'])

92
        r.check_response(count=warmups+count, http_status=200)

93
        assert r.total_connects == 1

94
        t_avg, i_min, t_min, i_max, t_max = self.stats_spread(r.stats[warmups:], 'time_total')

95
        if t_max < (5 * t_min):

96
            log.warning(f'avg time of transfer: {t_avg} [{i_min}={t_min}, {i_max}={t_max}]')

97


98
    # download 50 files in 10000 chunks a 1 byte with 10us delay between

99
    # prepend 100 file requests to warm up connection processing limits

100
    # (Apache2 increases # of parallel processed requests after successes)

101
    @pytest.mark.parametrize("proto", ['h2', 'h3'])

102
    def test_04_04_1000_10_1(self, env: Env, httpd, nghttpx, proto):

103
        if proto == 'h3' and not env.have_h3():

104
            pytest.skip("h3 not supported")

105
        count = 50

106
        warmups = 100

107
        curl = CurlClient(env=env)

108
        url1 = f'https://{env.authority_for(env.domain1, proto)}/data.json?[0-{warmups-1}]'

109
        urln = f'https://{env.authority_for(env.domain1, proto)}' \

110
            f'/curltest/tweak?id=[0-{count - 1}]'\

111
            '&chunks=10000&chunk_size=1&chunk_delay=50us'

112
        r = curl.http_download(urls=[url1, urln], alpn_proto=proto,

113
                               extra_args=['--parallel'])

114
        r.check_response(count=warmups+count, http_status=200)

115
        assert r.total_connects == 1

116
        t_avg, i_min, t_min, i_max, t_max = self.stats_spread(r.stats[warmups:], 'time_total')

117
        if t_max < (5 * t_min):

118
            log.warning(f'avg time of transfer: {t_avg} [{i_min}={t_min}, {i_max}={t_max}]')

119


120
    def stats_spread(self, stats: List[Dict], key: str) -> Tuple[float, int, float, int, float]:

121
        stotals = 0.0

122
        s_min = 100.0

123
        i_min = -1

124
        s_max = 0.0

125
        i_max = -1

126
        for idx, s in enumerate(stats):

127
            val = float(s[key])

128
            stotals += val

129
            if val > s_max:

130
                s_max = val

131
                i_max = idx

132
            if val < s_min:

133
                s_min = val

134
                i_min = idx

135
        return stotals/len(stats), i_min, s_min, i_max, s_max

136


137