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

通过测试排除了一些问题

Browse Source
This commit is contained in:
wandering1
2026-07-19 04:32:18 +08:00
parent 1669cd216d
commit 4eb728583b
12 changed files with 1379 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

229
test/ch395f_socket_test.py
View File

@@ -1,4 +1,5 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CH395F BSD Socket API 测试程序
用于测试 STM32F4 + CH395F 的网络功能
@@ -105,73 +106,77 @@ def tcp_server_test(args):
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.settimeout(30)
server_socket.settimeout(args.timeout)
try:
server_socket.bind(('0.0.0.0', args.port))
server_socket.listen(1)
log_success(f"服务器启动，监听端口 {args.port}")
log_info("等待 CH395F TCP Client 连接...")
client_socket, client_addr = server_socket.accept()
log_success(f"连接来自: {client_addr[0]}:{client_addr[1]}")
client_socket.settimeout(5)
# 测试接收数据
log_info("-" * 40)
log_info("测试 1: 接收数据")
for i in range(args.count):
for conn_idx in range(3):
try:
data = client_socket.recv(args.size)
if data:
log_success(f"收到数据 ({len(data)} bytes): {data[:50]}...")
else:
log_error("收到空数据")
break
server_socket.settimeout(args.timeout)
log_info(f"等待第 {conn_idx+1} 个连接（超时 {args.timeout}s）...")
client_socket, client_addr = server_socket.accept()
log_success(f"连接#{conn_idx+1} 来自: {client_addr[0]}:{client_addr[1]}")
client_socket.settimeout(5)
# Test 1: 接收数据
log_info("-" * 40)
log_info(f"[连接{conn_idx+1}] 测试 1: 接收数据")
try:
data = client_socket.recv(args.size)
if data:
log_success(f"收到数据 ({len(data)} bytes): {data[:50]}...")
except socket.timeout:
log_error("接收超时")
# Test 2: PING echo
log_info(f"[连接{conn_idx+1}] 测试 2: 双向通信 (PING)")
ping_ok = 0
for i in range(5):
try:
test_data = f"PING_{i}".encode()
client_socket.send(test_data)
data = client_socket.recv(args.size)
if data and data == test_data:
log_success(f"回显匹配: {test_data.decode()}")
ping_ok += 1
elif data:
log_error(f"回显不匹配! 收到: {data[:20]}")
break
except socket.timeout:
log_error("通信超时")
break
except Exception as e:
log_error(f"通信错误: {e}")
break
log_info(f"PING 匹配: {ping_ok}/5")
# Test 3: 大包发送
log_info(f"[连接{conn_idx+1}] 测试 3: 大包发送")
for i in range(args.count):
try:
test_data = generate_data(min(100, args.size))
client_socket.send(test_data)
time.sleep(0.1)
except Exception as e:
log_error(f"发送错误: {e}")
break
log_success(f"发送 {args.count} 包完成")
client_socket.close()
log_success(f"连接#{conn_idx+1} 测试完成")
time.sleep(0.5)
except socket.timeout:
log_error("接收超时")
log_error(f"等待连接#{conn_idx+1} 超时")
break
except Exception as e:
log_error(f"接收错误: {e}")
log_error(f"连接#{conn_idx+1} 错误: {e}")
break
# 测试发送数据
log_info("-" * 40)
log_info("测试 2: 发送数据")
for i in range(args.count):
try:
test_data = generate_data(min(100, args.size))
client_socket.send(test_data)
log_success(f"发送数据 ({len(test_data)} bytes)")
time.sleep(0.1)
except Exception as e:
log_error(f"发送错误: {e}")
break
# 测试双向通信
log_info("-" * 40)
log_info("测试 3: 双向通信")
for i in range(min(5, args.count)):
try:
# 发送
test_data = f"PING_{i}".encode()
client_socket.send(test_data)
log_info(f"发送: {test_data.decode()}")
# 接收回显
data = client_socket.recv(args.size)
if data:
log_info(f"收到: {data.decode()}")
if data == test_data:
log_success("回显数据匹配!")
else:
log_error("回显数据不匹配!")
except Exception as e:
log_error(f"通信错误: {e}")
break
client_socket.close()
log_success("测试完成，连接已关闭")
log_success("全部连接测试完成")
except socket.timeout:
log_error("等待连接超时")
@@ -329,66 +334,83 @@ def udp_server_test(args):
def udp_client_test(args):
"""UDP 客户端测试（测试 CH395F UDP Server）"""
"""UDP Echo 测试（测试 CH395F UDP Server）"""
log_info("=" * 60)
log_info("UDP 客户端测试 - 测试 CH395F UDP Server 功能")
log_info("UDP Echo 测试")
log_info("=" * 60)
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
client_socket.settimeout(5)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
dest = (args.ip, args.port)
try:
dest_addr = (args.ip, args.port)
# 发送数据
log_info("-" * 40)
log_info("测试 1: 发送 UDP 数据")
for i in range(args.count):
# Phase A: HELLO 探针等待 STM32 就绪
log_info(f"Phase A: 等待 CH395F 就绪（超时 {args.timeout}s）")
sock.settimeout(2)
t0 = time.time()
ready = False
while time.time() - t0 < args.timeout:
try:
test_data = f"UDP_MSG_{i:04d}".encode()
client_socket.sendto(test_data, dest_addr)
log_success(f"发送数据到 {args.ip}:{args.port} ({len(test_data)} bytes)")
# 等待回显
data, addr = client_socket.recvfrom(args.size)
if data:
if data == test_data:
log_success(f"回显数据匹配! (from {addr[0]}:{addr[1]})")
else:
log_error("回显数据不匹配!")
else:
log_error("收到空数据")
sock.sendto(b"HELLO", dest)
data, addr = sock.recvfrom(64)
if data == b"HELLO":
log_success(f"就绪 ({time.time()-t0:.1f}s, from {addr[0]}:{addr[1]})")
ready = True
break
time.sleep(0.2)
except socket.timeout:
log_warn("接收超时")
pass
if not ready:
log_error(f"未就绪（{args.timeout}s 超时）")
return
sock.settimeout(5)
# Phase B: PING Echo
log_info("-" * 40)
log_info("Phase B: PING Echo（5 轮）")
ok = 0
for i in range(5):
msg = f"PING_{i}".encode()
try:
sock.sendto(msg, dest)
data, _ = sock.recvfrom(64)
if data == msg:
ok += 1
log_success(f"PING_{i} OK")
else:
log_error(f"PING_{i} mismatch: {data[:20]}")
break
except socket.timeout:
log_error(f"PING_{i} timeout")
break
time.sleep(0.02)
log_info(f"PING: {ok}/5")
# Phase C: 大包 Echo
log_info("-" * 40)
log_info(f"Phase C: 大包 Echo（{args.count} 轮）")
for i in range(args.count):
td = generate_data(min(100, args.size))
try:
sock.sendto(td, dest)
data, _ = sock.recvfrom(args.size)
if data == td:
log_success(f"#{i+1} OK ({len(data)}B)")
else:
log_error(f"#{i+1} mismatch")
break
time.sleep(0.02)
except socket.timeout:
log_warn(f"#{i+1} timeout")
except Exception as e:
log_error(f"发送/接收错误: {e}")
log_error(f"#{i+1}: {e}")
break
# 测试大数据
log_info("-" * 40)
log_info("测试 2: 大数据 UDP 传输")
try:
large_data = generate_data(min(512, args.size))
client_socket.sendto(large_data, dest_addr)
log_success(f"发送大数据 ({len(large_data)} bytes)")
data, addr = client_socket.recvfrom(args.size)
if data and data == large_data:
log_success("大数据传输完整!")
else:
log_error("大数据传输失败")
except Exception as e:
log_error(f"大数据传输错误: {e}")
log_success("测试完成")
log_success("Phase 3 完成")
except Exception as e:
log_error(f"客户端错误: {e}")
log_error(f"错误: {e}")
finally:
client_socket.close()
sock.close()
# ==================== 回显测试 ====================
@@ -401,13 +423,13 @@ def tcp_echo_server_test(args):
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.settimeout(60)
server_socket.settimeout(args.timeout)
try:
server_socket.bind(('0.0.0.0', args.port))
server_socket.listen(1)
log_success(f"回显服务器启动，端口 {args.port}")
log_info("等待连接...")
log_info(f"等待连接（超时 {args.timeout}s）...")
client_socket, client_addr = server_socket.accept()
log_success(f"连接来自: {client_addr[0]}:{client_addr[1]}")
@@ -656,6 +678,7 @@ def main():
parser.add_argument('--port', type=int, default=8080, help='端口号')
parser.add_argument('--count', type=int, default=10, help='测试次数')
parser.add_argument('--size', type=int, default=1024, help='数据包大小')
parser.add_argument('--timeout', type=int, default=180, help='接受连接超时（秒，默认180）')
args = parser.parse_args()

156
test/dhcp_server.py Normal file
View File

@@ -0,0 +1,156 @@
#!/usr/bin/env python3
"""
轻量 DHCP Server — 用于 CH395F Phase 6 测试
用法：以管理员身份运行
python dhcp_server.py --iface 192.168.1.2
"""
import socket
import struct
import argparse
import time
DHCP_SERVER_PORT = 67
DHCP_CLIENT_PORT = 68
# DHCP 消息类型
DHCP_DISCOVER = 1
DHCP_OFFER = 2
DHCP_REQUEST = 3
DHCP_ACK = 5
# DHCP 选项
OPT_MESSAGE_TYPE = 53
OPT_SERVER_ID = 54
OPT_REQUESTED_IP = 50
OPT_SUBNET_MASK = 1
OPT_ROUTER = 3
OPT_DNS = 6
OPT_LEASE_TIME = 51
OPT_END = 255
SERVER_IP = "192.168.1.2"
OFFER_IP = "192.168.1.100"
SUBNET_MASK = "255.255.255.0"
ROUTER = "192.168.1.1"
LEASE_TIME = 3600 # 1 小时
def ip_to_bytes(ip):
return bytes(map(int, ip.split('.')))
def build_dhcp_offer(discover_pkt, server_ip, offer_ip):
"""构建 DHCP OFFER 包"""
transaction_id = discover_pkt[232:236] # Bootp header 前 4 字节
client_mac = discover_pkt[28:34]
# BOOTP 头
op = b'\x02' # Boot reply
htype = b'\x01' # Ethernet
hlen = b'\x06' # MAC len
hops = b'\x00'
xid = transaction_id
secs = b'\x00\x00'
flags = b'\x00\x00'
ciaddr = b'\x00\x00\x00\x00'
yiaddr = ip_to_bytes(offer_ip) # Your IP
siaddr = ip_to_bytes(server_ip) # Server IP
giaddr = b'\x00\x00\x00\x00'
chaddr = client_mac + b'\x00' * 10 # Client MAC (padded to 16)
sname = b'\x00' * 64
file = b'\x00' * 128
magic = b'\x63\x82\x53\x63' # DHCP magic cookie
# DHCP 选项
options = b''
options += struct.pack('BBB', OPT_MESSAGE_TYPE, 1, DHCP_OFFER)
options += struct.pack('BB', OPT_SERVER_ID, 4) + ip_to_bytes(server_ip)
options += struct.pack('BB', OPT_REQUESTED_IP, 4) + ip_to_bytes(offer_ip)
options += struct.pack('BB', OPT_SUBNET_MASK, 4) + ip_to_bytes(SUBNET_MASK)
options += struct.pack('BB', OPT_ROUTER, 4) + ip_to_bytes(ROUTER)
options += struct.pack('BB', OPT_DNS, 4) + ip_to_bytes(ROUTER)
options += bytes([OPT_LEASE_TIME, 4]) + struct.pack('!I', LEASE_TIME)
options += bytes([OPT_END])
pkt = op + htype + hlen + hops + xid + secs + flags + ciaddr + yiaddr + siaddr + giaddr + chaddr + sname + file + magic + options
return pkt
def build_dhcp_ack(request_pkt, server_ip, offer_ip):
"""构建 DHCP ACK 包"""
transaction_id = request_pkt[232:236]
client_mac = request_pkt[28:34]
op = b'\x02'
htype = b'\x01'
hlen = b'\x06'
hops = b'\x00'
xid = transaction_id
secs = b'\x00\x00'
flags = b'\x00\x00'
ciaddr = b'\x00\x00\x00\x00'
yiaddr = ip_to_bytes(offer_ip)
siaddr = ip_to_bytes(server_ip)
giaddr = b'\x00\x00\x00\x00'
chaddr = client_mac + b'\x00' * 10
sname = b'\x00' * 64
file = b'\x00' * 128
magic = b'\x63\x82\x53\x63'
options = b''
options += struct.pack('BBB', OPT_MESSAGE_TYPE, 1, DHCP_ACK)
options += struct.pack('BB', OPT_SERVER_ID, 4) + ip_to_bytes(server_ip)
options += struct.pack('BB', OPT_SUBNET_MASK, 4) + ip_to_bytes(SUBNET_MASK)
options += struct.pack('BB', OPT_ROUTER, 4) + ip_to_bytes(ROUTER)
options += struct.pack('BB', OPT_DNS, 4) + ip_to_bytes(ROUTER)
options += bytes([OPT_LEASE_TIME, 4]) + struct.pack('!I', LEASE_TIME)
options += bytes([OPT_END])
return op + htype + hlen + hops + xid + secs + flags + ciaddr + yiaddr + siaddr + giaddr + chaddr + sname + file + magic + options
def main():
parser = argparse.ArgumentParser(description="轻量 DHCP Server for CH395F 测试")
parser.add_argument("--iface", default=SERVER_IP, help="网卡 IP 地址")
args = parser.parse_args()
offer_ip = OFFER_IP
server_ip = args.iface
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
sock.bind(('0.0.0.0', DHCP_SERVER_PORT))
sock.settimeout(60)
print(f"DHCP Server 启动, 网卡 IP={server_ip}, 分配 IP={offer_ip}")
print("等待 CH395F DHCP DISCOVER...")
try:
# 第一阶段：DISCOVER → OFFER
data, addr = sock.recvfrom(4096)
msg_type = data[232 + 4 + 1] if len(data) > 250 else 0
client_mac = ':'.join(f'{b:02X}' for b in data[28:34])
print(f"收到 DHCP DISCOVER from {client_mac}")
offer = build_dhcp_offer(data, server_ip, offer_ip)
sock.sendto(offer, ('255.255.255.255', DHCP_CLIENT_PORT))
print("发送 DHCP OFFER")
# 第二阶段：REQUEST → ACK
sock.settimeout(10)
try:
data, addr = sock.recvfrom(4096)
client_mac = ':'.join(f'{b:02X}' for b in data[28:34])
print(f"收到 DHCP REQUEST from {client_mac}")
ack = build_dhcp_ack(data, server_ip, offer_ip)
sock.sendto(ack, ('255.255.255.255', DHCP_CLIENT_PORT))
print(f"发送 DHCP ACK → {offer_ip}")
print("DHCP 分配完成！")
except socket.timeout:
print("等待 DHCP REQUEST 超时")
except socket.timeout:
print("等待 DHCP DISCOVER 超时（60s）")
finally:
sock.close()
if __name__ == '__main__':
main()