# disregard RST packets it sent from its port to curl
assert len(r.tcpdump.stats_excluding(src_port=env.ftps_port)) == 0, f'Unexpected TCP RSTs packets'
+ def test_31_08_upload_ascii(self, env: Env, vsftpds: VsFTPD):
+ docname = 'upload-ascii'
+ line_length = 21
+ srcfile = os.path.join(env.gen_dir, docname)
+ dstfile = os.path.join(vsftpds.docs_dir, docname)
+ env.make_data_file(indir=env.gen_dir, fname=docname, fsize=100*1024,
+ line_length=line_length)
+ srcsize = os.path.getsize(srcfile)
+ self._rmf(dstfile)
+ count = 1
+ curl = CurlClient(env=env)
+ url = f'ftp://{env.ftp_domain}:{vsftpds.port}/'
+ r = curl.ftp_ssl_upload(urls=[url], fupload=f'{srcfile}', with_stats=True,
+ extra_args=['--use-ascii'])
+ r.check_stats(count=count, http_status=226)
+ # expect the uploaded file to be number of converted newlines larger
+ dstsize = os.path.getsize(dstfile)
+ newlines = len(open(srcfile).readlines())
+ assert (srcsize + newlines) == dstsize, \
+ f'expected source with {newlines} lines to be that much larger,'\
+ f'instead srcsize={srcsize}, upload size={dstsize}, diff={dstsize-srcsize}'
+
def check_downloads(self, client, srcfile: str, count: int,
complete: bool = True):
for i in range(count):
def authority_for(self, domain: str, alpn_proto: Optional[str] = None):
return f'{domain}:{self.port_for(alpn_proto=alpn_proto)}'
- def make_data_file(self, indir: str, fname: str, fsize: int) -> str:
+ def make_data_file(self, indir: str, fname: str, fsize: int,
+ line_length: int = 1024) -> str:
+ if line_length < 11:
+ raise 'line_length less than 11 not supported'
fpath = os.path.join(indir, fname)
s10 = "0123456789"
- s = (101 * s10) + s10[0:3]
+ s = round((line_length / 10) + 1) * s10
+ s = s[0:line_length-11]
with open(fpath, 'w') as fd:
- for i in range(int(fsize / 1024)):
+ for i in range(int(fsize / line_length)):
fd.write(f"{i:09d}-{s}\n")
- remain = int(fsize % 1024)
+ remain = int(fsize % line_length)
if remain != 0:
- i = int(fsize / 1024) + 1
- s = f"{i:09d}-{s}\n"
- fd.write(s[0:remain])
+ i = int(fsize / line_length) + 1
+ fd.write(f"{i:09d}-{s}"[0:remain-1] + "\n")
return fpath
def make_clients(self):