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

CH395F 硬件测试完毕

Browse Source
This commit is contained in:
wandering1
2026-08-24 16:16:18 +08:00
parent fc88bc8752
commit 4375873083
14 changed files with 2183 additions and 787 deletions
Download Patch File Download Diff File Expand all files Collapse all files

425
test/ch395f_socket_test.py
View File

@@ -8,19 +8,22 @@ CH395F BSD Socket API 测试程序
python ch395f_socket_test.py [模式] [选项]
模式:
tcp_server - 作为 TCP 服务器（测试 CH395F TCP Client）
tcp_server - TCP 回显服务器，多连接循环（测试 CH395F TCP Client）
tcp_client - 作为 TCP 客户端（测试 CH395F TCP Server）
udp_server - 作为 UDP 服务器（测试 CH395F UDP Client）
udp_client - 作为 UDP 客户端（测试 CH395F UDP Server）
tcp_echo - TCP 回显服务器（测试双向通信）
tcp_echo - TCP 回显服务器（单连接，测试双向通信）
udp_echo - UDP 回显服务器（测试双向通信）
stress - 压力测试（多连接并发测试）
tcp_bulk - 双向大块：先收满 bulk 字节（不回声，校验 LCG），再回放（阶段 5）
选项:
--ip 目标 IP 地址（默认: 192.168.1.100）
--port 端口号（默认: 8080）
--count 测试次数（默认: 10）
--size 数据包大小（默认: 1024）
--bulk tcp_bulk 模式双向字节数（默认: 262144）
--chunk tcp_bulk 模式块大小，须与固件 PHASE5_CHUNK_SIZE 一致（默认: 4096）
示例:
# 测试 CH395F TCP Server
@@ -37,10 +40,14 @@ CH395F BSD Socket API 测试程序
# 压力测试（多连接）
python ch395f_socket_test.py stress --ip 192.168.1.100 --port 8080 --count 7
# 阶段 5 双向大块（先收 256KB 再回放 256KB）
python ch395f_socket_test.py tcp_bulk --port 8081 --bulk 262144 --chunk 4096
"""
import socket
import sys
import os
import time
import argparse
import threading
@@ -49,6 +56,10 @@ import random
import string
from datetime import datetime
# TCP 回显服务器（tcp_server 模式）参数
INTER_ACCEPT_TIMEOUT = 60 # 首个连接后，等待下一个连接的超时秒数
ECHO_IDLE_TIMEOUT = 30 # 连接内无数据的空闲超时秒数
class Colors:
"""终端颜色"""
@@ -69,19 +80,19 @@ def log_msg(msg, color=Colors.END):
def log_success(msg):
log_msg(f"✓ {msg}", Colors.GREEN)
log_msg("[OK] %s" % msg, Colors.GREEN)
def log_error(msg):
log_msg(f"✗ {msg}", Colors.RED)
log_msg("[ERR] %s" % msg, Colors.RED)
def log_info(msg):
log_msg(f"ℹ {msg}", Colors.BLUE)
log_msg("[..] %s" % msg, Colors.BLUE)
def log_warn(msg):
log_msg(f"⚠ {msg}", Colors.YELLOW)
log_msg("[!!] %s" % msg, Colors.YELLOW)
def generate_data(size):
@@ -99,89 +110,195 @@ def verify_data(data, expected_len):
# ==================== TCP 测试函数 ====================
def tcp_server_test(args):
"""TCP 服务器测试（测试 CH395F TCP Client）"""
"""TCP 回显服务器测试（测试 CH395F TCP Client）
与固件 TC-NET-201/202 配套：MCU 连接后发送数据并等待原样回显，
校验通过后主动关闭。本函数循环接受多个连接，每个连接内将收到的
数据逐字回显，直到对端关闭或空闲超时。
首个连接等待 args.timeout 秒；之后每个连接间最多等待
INTER_ACCEPT_TIMEOUT 秒，无新连接则结束并汇总。
返回 False 表示失败（无连接或回显出错），main 据此返回非 0 退出码。
"""
log_info("=" * 60)
log_info("TCP 服务器测试 - 测试 CH395F TCP Client 功能")
log_info("TCP 回显服务器 - 测试 CH395F TCP Client 功能")
log_info("=" * 60)
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.settimeout(args.timeout)
conn_total = 0
echo_bytes = 0
conn_errors = 0
try:
server_socket.bind(('0.0.0.0', args.port))
server_socket.listen(1)
server_socket.listen(4)
log_success(f"服务器启动，监听端口 {args.port}")
for conn_idx in range(3):
conn_idx = 0
while True:
wait = args.timeout if conn_idx == 0 else min(args.timeout, INTER_ACCEPT_TIMEOUT)
server_socket.settimeout(wait)
try:
server_socket.settimeout(args.timeout)
log_info(f"等待第 {conn_idx+1} 个连接（超时 {args.timeout}s）...")
log_info(f"等待第 {conn_idx+1} 个连接（超时 {wait}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(f"等待连接#{conn_idx+1} 超时")
if conn_idx == 0:
log_error("等待连接超时")
else:
log_info(f"{wait}s 内无新连接，服务器退出")
break
conn_idx += 1
log_success(f"连接#{conn_idx} 来自: {client_addr[0]}:{client_addr[1]}")
# 回显循环：收到什么回什么（sendall 保证发完），对端关闭或空闲超时结束
client_socket.settimeout(ECHO_IDLE_TIMEOUT)
rx = 0
try:
while True:
data = client_socket.recv(4096)
if not data:
log_info(f"[连接{conn_idx}] 对端已关闭")
break
client_socket.sendall(data)
rx += len(data)
except socket.timeout:
log_warn(f"[连接{conn_idx}] 空闲超时，关闭连接")
except Exception as e:
log_error(f"连接#{conn_idx+1} 错误: {e}")
break
log_error(f"[连接{conn_idx}] 回显错误: {e}")
conn_errors += 1
finally:
client_socket.close()
log_success("全部连接测试完成")
if rx > 0:
log_success(f"连接#{conn_idx} 完成: 回显 {rx} 字节")
else:
log_warn(f"连接#{conn_idx} 未收到任何数据")
conn_errors += 1
echo_bytes += rx
conn_total += 1
log_info("-" * 40)
log_info(f"服务结束: 连接 {conn_total} 个, 回显 {echo_bytes} 字节, 异常 {conn_errors} 次")
if conn_total == 0:
log_error("无任何客户端连接")
return False
return conn_errors == 0
except socket.timeout:
log_error("等待连接超时")
except Exception as e:
log_error(f"服务器错误: {e}")
return False
finally:
server_socket.close()
# ==================== 阶段 5：双向大块传输 ====================
BULK_IDLE_TIMEOUT = 10 # bulk 接收阶段单次 recv 等待超时（秒）
def _lcg_block(idx, size):
"""生成与固件 phase5_gen_block 完全一致的 LCG 块数据（uint32 回绕）"""
seed = (0xC2D2E6F9 ^ (idx * 2654435761)) & 0xFFFFFFFF
out = bytearray(size)
for k in range(size):
seed = (seed * 1664525 + 1013904223) & 0xFFFFFFFF
out[k] = (seed >> 24) & 0xFF
return bytes(out)
def tcp_bulk_test(args):
"""双向大块传输测试（配合固件阶段 5 硬件层）
协议：接受连接 → 收满 bulk 字节（不回声，校验 LCG 模式）→
按相同 LCG 模式回放 bulk 字节 → 关闭。
MCU 侧先连续发送 256KB（本端只收），再连续接收 256KB（本端回放），
分别测得纯 TX / 纯 RX 吞吐。
"""
chunk = args.chunk
bulk = (args.bulk // chunk) * chunk # 按整块取整
blocks = bulk // chunk
log_info("=" * 60)
log_info("TCP Bulk 双向大块 - 配合 CH395F 阶段 5")
log_info(f"协议: 收满 {bulk} 字节(不回声) -> 回放 {bulk} 字节 (chunk={chunk}, blocks={blocks})")
log_info("=" * 60)
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
server_socket.bind(('0.0.0.0', args.port))
server_socket.listen(4)
log_success(f"服务器启动，监听端口 {args.port}")
server_socket.settimeout(args.timeout)
try:
client_socket, client_addr = server_socket.accept()
except socket.timeout:
log_error("等待连接超时")
return False
log_success(f"连接来自: {client_addr[0]}:{client_addr[1]}")
client_socket.settimeout(BULK_IDLE_TIMEOUT)
# 预生成期望/回放数据（LCG，与固件一致）
expected = bytearray(bulk)
for i in range(blocks):
expected[i * chunk:(i + 1) * chunk] = _lcg_block(i, chunk)
# ---- 阶段 A：收满 bulk 字节（不回声），校验 LCG 模式 ----
rx_buf = bytearray()
rx_t0 = time.time()
try:
while len(rx_buf) < bulk:
data = client_socket.recv(min(65536, bulk - len(rx_buf)))
if not data:
log_error(f"对端提前关闭，仅收到 {len(rx_buf)}/{bulk} 字节")
client_socket.close()
return False
rx_buf += data
except socket.timeout:
log_error(f"接收空闲超时，仅收到 {len(rx_buf)}/{bulk} 字节")
client_socket.close()
return False
rx_ms = int((time.time() - rx_t0) * 1000)
ok = (bytes(rx_buf) == bytes(expected))
if ok:
log_success(f"阶段A 收满 {len(rx_buf)} 字节，LCG 校验一致")
else:
bad = -1
for i in range(min(len(rx_buf), len(expected))):
if rx_buf[i] != expected[i]:
bad = i
break
if bad >= 0:
log_error(f"阶段A 数据不一致 @第 {bad} 字节 (期望 0x{expected[bad]:02X}, "
f"实际 0x{rx_buf[bad]:02X})")
else:
log_error(f"阶段A 数据不一致（长度 {len(rx_buf)} vs {len(expected)}）")
log_info(f"阶段A 接收 {len(rx_buf)} 字节，耗时 {rx_ms} ms "
f"({len(rx_buf) * 1000 // max(rx_ms, 1)} B/s)")
# ---- 阶段 B：按 LCG 模式回放 bulk 字节 ----
tx_t0 = time.time()
try:
for off in range(0, bulk, chunk):
client_socket.sendall(expected[off:off + chunk])
except Exception as e:
log_error(f"阶段B 回放错误: {e}")
ok = False
finally:
client_socket.close()
tx_ms = int((time.time() - tx_t0) * 1000)
if ok:
log_success(f"阶段B 回放 {bulk} 字节完成，耗时 {tx_ms} ms "
f"({bulk * 1000 // max(tx_ms, 1)} B/s)")
return ok
except Exception as e:
log_error(f"服务器错误: {e}")
return False
finally:
server_socket.close()
@@ -334,17 +451,24 @@ def udp_server_test(args):
def udp_client_test(args):
"""UDP Echo 测试（测试 CH395F UDP Server）"""
"""UDP Echo 测试（测试 CH395F UDP Server，对应指南阶段 3）
会话结构与固件 phase3_run() 约定：
- "HELLO" 探针：回显即就绪，不计入固件轮次
- TC-301：30 轮 64B 基础回显（--size 非默认值时覆盖单轮载荷大小）
- TC-302：1B/64B/200B 各 --count 轮变长合规（默认 10）
任一轮 mismatch/timeout 立即终止；结束输出汇总，失败退出码 1。
"""
log_info("=" * 60)
log_info("UDP Echo 测试")
log_info("UDP Echo 测试（TC-301 基础回显 + TC-302 变长合规）")
log_info("=" * 60)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
dest = (args.ip, args.port)
try:
# Phase A: HELLO 探针等待 STM32 就绪
log_info(f"Phase A: 等待 CH395F 就绪（超时 {args.timeout}s）")
# Phase A: HELLO 就绪探针（固件只回显不计数，不占用例预算）
log_info(f"Phase A: HELLO 就绪探针（超时 {args.timeout}s）")
sock.settimeout(2)
t0 = time.time()
ready = False
@@ -360,55 +484,70 @@ def udp_client_test(args):
pass
if not ready:
log_error(f"未就绪（{args.timeout}s 超时）")
return
sys.exit(1)
sock.settimeout(5)
# Phase B: PING Echo
# Phase B: TC-301 基础回显 30 轮 x 64B
size_b = 64 if args.size == 1024 else min(args.size, 200)
rounds_b = 30
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))
log_info(f"Phase B: TC-301 基础回显（{rounds_b} 轮 x {size_b}B）")
ok_b = 0
for i in range(rounds_b):
td = generate_data(size_b)
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")
data, _ = sock.recvfrom(size_b + 16)
if data != td:
log_error(f"[TC-301] #{i+1} mismatch: {len(data)}B != {len(td)}B")
break
time.sleep(0.02)
except socket.timeout:
log_warn(f"#{i+1} timeout")
except Exception as e:
log_error(f"#{i+1}: {e}")
log_error(f"[TC-301] #{i+1} echo timeout")
break
ok_b += 1
if ok_b % 10 == 0:
log_info(f"[TC-301] 进度 {ok_b}/{rounds_b}")
time.sleep(0.02)
log_success("Phase 3 完成")
# Phase C: TC-302 变长合规（仅 Phase B 全过后继续，否则后续必超时）
sizes_c = [1, 64, 200]
total_c = len(sizes_c) * args.count
ok_c = 0
if ok_b == rounds_b:
log_info("-" * 40)
log_info(f"Phase C: TC-302 变长合规（{sizes_c}B 各 {args.count} 轮）")
for sz in sizes_c:
ok_sz = 0
for i in range(args.count):
td = generate_data(sz)
try:
sock.sendto(td, dest)
data, _ = sock.recvfrom(sz + 16)
if data != td:
log_error(f"[TC-302] {sz}B #{i+1} mismatch: {len(data)}B != {sz}B")
break
except socket.timeout:
log_error(f"[TC-302] {sz}B #{i+1} echo timeout")
break
ok_sz += 1
time.sleep(0.02)
log_info(f"[TC-302] {sz}B: {ok_sz}/{args.count}")
ok_c += ok_sz
if ok_sz < args.count:
break
# 汇总与退出码
log_info("=" * 60)
if ok_b == rounds_b and ok_c == total_c:
log_success(f"PASS: TC-301 {ok_b}/{rounds_b}, TC-302 {ok_c}/{total_c}")
else:
log_error(f"FAIL: TC-301 {ok_b}/{rounds_b}, TC-302 {ok_c}/{total_c}")
sys.exit(1)
except Exception as e:
log_error(f"错误: {e}")
sys.exit(1)
finally:
sock.close()
@@ -647,6 +786,7 @@ DHCP_OFFER = 2
DHCP_REQUEST = 3
DHCP_ACK = 5
OPT_MESSAGE_TYPE = 53
DHCP_ANNOUNCE_PORT = 60001 # MCU Socket6 广播宣告端口（固件 phase4_run）
OPT_SERVER_ID = 54
OPT_SUBNET_MASK = 1
OPT_ROUTER = 3
@@ -675,6 +815,30 @@ def _build_dhcp_pkt(op, xid, client_mac, yiaddr, siaddr, options):
return op_b + htype + hlen + hops + xid + secs + flags + ciaddr + yiaddr + siaddr + giaddr + chaddr + sname + file + magic + options
def _announce_listener(results):
"""后台监听 MCU 的 4B IP 广播宣告（UDP :60001，TC-602 步1 的 PC 侧观测）"""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
s.bind(('0.0.0.0', DHCP_ANNOUNCE_PORT))
except OSError as e:
results['announce_err'] = f'绑定 {DHCP_ANNOUNCE_PORT} 失败: {e}'
return
s.settimeout(45)
try:
data, addr = s.recvfrom(16)
if len(data) == 4:
ip = '.'.join(str(b) for b in data)
log_success(f"收到 IP 宣告: {ip} (from {addr[0]}:{addr[1]})")
results['announce'] = ip
else:
results['announce_err'] = f'宣告长度异常 {len(data)}B（预期 4B）'
except socket.timeout:
results['announce_err'] = '45s 未收到宣告'
finally:
s.close()
def dhcp_server_test(args):
"""DHCP Server 模式（DISCOVER → OFFER → REQUEST → ACK）"""
log_info("=" * 60)
@@ -700,9 +864,14 @@ def dhcp_server_test(args):
sock.settimeout(60)
log_info("DHCP 服务器已启动，等待 CH395F DISCOVER...")
log_info("（请确保 MCU 端已启用 ENABLE_PHASE6_TESTS 并复位）")
log_info("（请确保 MCU 端已启用 ENABLE_PHASE4_TESTS 并复位）")
print()
# 宣告监听线程：MCU 在 ACK 后立即广播 4B IP 到 :60001
results = {}
listener = threading.Thread(target=_announce_listener, args=(results,), daemon=True)
listener.start()
try:
# Phase 1: DISCOVER → OFFER
data, addr = sock.recvfrom(4096)
@@ -799,6 +968,17 @@ def dhcp_server_test(args):
print()
# 汇总宣告监听结果（TC-602 步 1 PC 侧观测；join 延迟由固件 30s HELLO 窗覆盖）
listener.join(timeout=10)
if results.get('announce'):
if results['announce'] != offer_ip:
log_error(f"IP 宣告 {results['announce']} != 租约 {offer_ip}！"
"芯片固件可能回读了未提交的旧 IP 配置")
elif 'announce_err' in results:
log_warn(f"IP 宣告未观测到：{results['announce_err']}")
print()
# Phase 3: 连通性验证（MCU Phase 6 内置 HELLO 端口 = 60000）
hello_port = args.port if args.port != 8080 else 60000
log_info(f"验证 {offer_ip} 连通性（UDP HELLO → port {hello_port}）")
@@ -892,17 +1072,20 @@ def main():
%(prog)s tcp_server --port 8080
%(prog)s stress --ip 192.168.1.100 --port 8080 --count 5
%(prog)s dhcp_server --dhcp-iface 192.168.1.2 --ip 192.168.1.100
%(prog)s tcp_bulk --port 8081 --bulk 262144 --chunk 4096
"""
)
parser.add_argument('mode', choices=['tcp_server', 'tcp_client', 'udp_server', 'udp_client',
'tcp_echo', 'udp_echo', 'stress', 'packet_size',
'dhcp_server'],
'dhcp_server', 'tcp_bulk'],
help='测试模式')
parser.add_argument('--ip', default='192.168.1.100', help='目标 IP 地址 / DHCP 待分配 IP')
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('--bulk', type=int, default=262144, help='tcp_bulk 模式双向字节数')
parser.add_argument('--chunk', type=int, default=4096, help='tcp_bulk 模式块大小，须与固件 PHASE5_CHUNK_SIZE 一致')
parser.add_argument('--timeout', type=int, default=180, help='接受连接超时（秒，默认180）')
parser.add_argument('--dhcp-iface', default='192.168.1.2', help='DHCP 服务器网卡 IP')
parser.add_argument('--dhcp-mask', default='255.255.255.0', help='DHCP 分配的子网掩码')
@@ -910,15 +1093,24 @@ def main():
args = parser.parse_args()
# GBK 控制台容错：不可编码字符替换而非崩溃
for stream in (sys.stdout, sys.stderr):
if stream is not None and hasattr(stream, 'reconfigure'):
stream.reconfigure(errors='replace')
if sys.platform == 'win32':
os.system('') # 启用 Windows 控制台 VT 序列，避免 ANSI 颜色码乱码
print()
print(Colors.BOLD + "=" * 60 + Colors.END)
print(Colors.BOLD + " CH395F BSD Socket API 测试程序" + Colors.END)
print(Colors.BOLD + "=" * 60 + Colors.END)
print()
ret = None
try:
if args.mode == 'tcp_server':
tcp_server_test(args)
ret = tcp_server_test(args)
elif args.mode == 'tcp_client':
tcp_client_test(args)
elif args.mode == 'udp_server':
@@ -933,14 +1125,21 @@ def main():
stress_test(args)
elif args.mode == 'packet_size':
packet_size_test(args)
elif args.mode == 'tcp_bulk':
ret = tcp_bulk_test(args)
elif args.mode == 'dhcp_server':
dhcp_server_test(args)
except KeyboardInterrupt:
log_warn("用户中断")
sys.exit(130)
except Exception as e:
log_error(f"未预期的错误: {e}")
import traceback
traceback.print_exc()
sys.exit(2)
if ret is False:
sys.exit(1)
if __name__ == '__main__':