f'/curltest/tweak?status=400&delay=5ms&chunks=1&body_error=reset&id=[0-{count-1}]'
r = curl.http_upload(urls=[url], data=f'@{fdata}', alpn_proto=proto,
extra_args=['--parallel'])
- exp_exit = 92 if proto == 'h2' else 95
- r.check_stats(count=count, exitcode=exp_exit)
+ # depending on timing and protocol, we might get CURLE_PARTIAL_FILE or
+ # CURLE_HTTP3 or CURLE_HTTP2_STREAM
+ r.check_stats(count=count, exitcode=[18, 92, 95])
# PUT 100k
@pytest.mark.parametrize("proto", ['http/1.1', 'h2', 'h3'])
f'were made\n{self.dump_logs()}'
def check_stats(self, count: int, http_status: Optional[int] = None,
- exitcode: Optional[int] = None,
+ exitcode: Optional[Union[int, List[int]]] = None,
remote_port: Optional[int] = None,
remote_ip: Optional[str] = None):
if exitcode is None:
self.check_exit_code(0)
+ elif isinstance(exitcode, int):
+ exitcode = [exitcode]
assert len(self.stats) == count, \
f'stats count: expected {count}, got {len(self.stats)}\n{self.dump_logs()}'
if http_status is not None:
if exitcode is not None:
for idx, x in enumerate(self.stats):
if 'exitcode' in x:
- assert x['exitcode'] == exitcode, \
+ assert x['exitcode'] in exitcode, \
f'status #{idx} exitcode: expected {exitcode}, '\
f'got {x["exitcode"]}\n{self.dump_stat(x)}'
if remote_port is not None: