wandering1/STM32F4-Base
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FTP 测试通过

Browse Source
This commit is contained in:
wandering1
2026-08-28 12:22:10 +08:00
parent cf33171c8f
commit 615a9b13e9
10 changed files with 437 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
test/ftp_up_dl_test.py
View File

@@ -128,33 +128,67 @@ def ftp_transfer_data(host, pasv_ip, pasv_port, cmd, data_to_send=None, read_dat
def test_upload_download(host, port):
print(f"=== FTP Upload/Download Test: {host}:{port} ===\n")
checks = []
def check(label, ok, detail=""):
checks.append((label, ok, detail))
mark = "PASS" if ok else "FAIL"
print(f" [{mark}] {label}{(' - ' + detail) if detail else ''}")
def finish(sock):
passed = all(ok for _, ok, _ in checks)
print("\n" + "=" * 40)
print("FTP Upload/Download Summary:")
for label, ok, _ in checks:
print(f" [{'PASS' if ok else 'FAIL'}] {label}")
print("=" * 40)
if passed:
print(">>> TEST PASSED <<<")
else:
print(">>> TEST FAILED <<<")
print("=== Test Complete ===")
if sock:
try:
sock.close()
except Exception:
pass
return passed
# 1. Connect + Login
print("[1] Connect + Login")
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(TIMEOUT)
sock.connect((host, port))
try:
sock.connect((host, port))
except Exception as e:
print(f" CONNECT FAILED: {e}")
check("Connect", False, str(e))
return finish(sock)
recv_line(sock) # 220
if not send_cmd(sock, "USER anonymous", 230):
sock.close(); return
check("Login USER 230", False)
return finish(sock)
if not send_cmd(sock, "PASS test@test.com", 230):
sock.close(); return
check("Login PASS 230", False)
return finish(sock)
check("Login (USER/PASS 230)", True)
# 2. TYPE I (binary mode)
print("\n[2] Set binary mode")
send_cmd(sock, "TYPE I")
resp = send_cmd(sock, "TYPE I")
type_ok = bool(resp) and resp.startswith(b"200")
check("TYPE I 200", type_ok)
# 3. PASV + STOR (upload)
print("\n[3] Upload: STOR up_test.dat")
resp = send_cmd(sock, "PASV")
if not resp:
sock.close(); return
check("PASV", False, "no response")
return finish(sock)
pasv_ip, pasv_port = parse_pasv(resp)
print(f" PASV: {pasv_ip}:{pasv_port}")
# Connect data channel
data_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
data_sock.settimeout(10.0)
try:
@@ -162,12 +196,11 @@ def test_upload_download(host, port):
print(f" Data connected.")
except Exception as e:
print(f" Data connect FAILED: {e}")
sock.close(); return
check("PASV data connect", False, str(e))
return finish(sock)
check("PASV data connect", True, f"{pasv_ip}:{pasv_port}")
# Send STOR
send_cmd(sock, "STOR up_test.dat")
# Send file data
total = 0
while total < len(TEST_DATA):
try:
@@ -179,20 +212,20 @@ def test_upload_download(host, port):
print(f" Send error: {e}")
break
print(f" Uploaded {total}/{len(TEST_DATA)} bytes.")
data_sock.shutdown(socket.SHUT_WR)
data_sock.close()
# Read STOR response
resp = recv_line(sock)
if resp:
print(f" Response: {resp.decode().rstrip()}")
stor_ok = bool(resp) and resp.startswith(b"226") and total == len(TEST_DATA)
check("STOR 226 + bytes", stor_ok,
f"uploaded={total}/{len(TEST_DATA)} resp={resp.decode().rstrip() if resp else 'none'}")
# 4. PASV + LIST (verify)
print("\n[4] Verify: LIST")
resp = send_cmd(sock, "PASV")
if not resp:
sock.close(); return
check("LIST PASV", False, "no response")
return finish(sock)
pasv_ip, pasv_port = parse_pasv(resp)
data_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -213,15 +246,17 @@ def test_upload_download(host, port):
data_sock.close()
resp = recv_line(sock)
if resp:
print(f" Response: {resp.decode().rstrip()}")
list_ok = bool(data) and b"UP_TEST.DAT" in data.upper()
check("LIST shows UP_TEST.DAT", list_ok,
resp.decode().rstrip() if resp else "no resp")
# 5. PASV + RETR (download)
time.sleep(1.0)
print("\n[5] Download: RETR up_test.dat")
resp = send_cmd(sock, "PASV")
if not resp:
sock.close(); return
check("RETR PASV", False, "no response")
return finish(sock)
pasv_ip, pasv_port = parse_pasv(resp)
data_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -230,10 +265,11 @@ def test_upload_download(host, port):
data_sock.connect((pasv_ip, pasv_port))
except Exception as e:
print(f" Data connect FAILED: {e}")
sock.close(); return
check("RETR data connect", False, str(e))
return finish(sock)
check("RETR data connect", True, f"{pasv_ip}:{pasv_port}")
send_cmd(sock, "RETR up_test.dat")
dl_data = b""
while True:
try:
@@ -244,39 +280,46 @@ def test_upload_download(host, port):
except socket.timeout:
break
data_sock.close()
resp = recv_line(sock)
if resp:
print(f" Response: {resp.decode().rstrip()}")
retr_ok = bool(resp) and resp.startswith(b"226")
check("RETR 226", retr_ok, resp.decode().rstrip() if resp else "no resp")
# Compare
print(f"\n[6] Compare")
print(f" Uploaded: {len(TEST_DATA)} bytes")
print(f" Downloaded: {len(dl_data)} bytes")
if dl_data == TEST_DATA:
match = (dl_data == TEST_DATA)
if match:
print(f" *** MATCH! Upload/Download success. ***")
else:
print(f" *** MISMATCH! ***")
if dl_data:
print(f" First 80 bytes uploaded: {TEST_DATA[:80]!r}")
print(f" First 80 bytes downloaded: {dl_data[:80]!r}")
check("Upload/Download MATCH", match, f"up={len(TEST_DATA)} dl={len(dl_data)}")
# 7. DELE (delete test file)
print("\n[7] Cleanup: DELE up_test.dat")
send_cmd(sock, "DELE up_test.dat")
resp = send_cmd(sock, "DELE up_test.dat")
dele_ok = bool(resp) and resp.startswith(b"250")
check("DELE 250", dele_ok, resp.decode().rstrip() if resp else "no resp")
# 8. QUIT
print("\n[8] QUIT")
send_cmd(sock, "QUIT")
sock.close()
print("\n=== Test Complete ===")
resp = send_cmd(sock, "QUIT")
quit_ok = bool(resp) and resp.startswith(b"221")
check("QUIT 221", quit_ok, resp.decode().rstrip() if resp else "no resp")
return finish(sock)
def main():
parser = argparse.ArgumentParser(description="FTP Upload/Download Test")
parser.add_argument("--host", default=HOST)
parser.add_argument("--port", type=int, default=PORT)
args = parser.parse_args()
test_upload_download(args.host, args.port)
success = test_upload_download(args.host, args.port)
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()