serial_debug_mcp/test_query.py at main · luiox/serial_debug_mcp · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
"""测试 query_data 功能"""

import time
from main import state, _query_data_impl, HEADERS

# 模拟添加测试数据到缓冲区
print("Adding test data to buffer...")

base_time = time.time()
test_data = [
    (base_time + 0.0, "12:00:00.000", "1000, 10, 20"),
    (base_time + 0.1, "12:00:00.100", "1001, 11, 21"),
    (base_time + 0.2, "12:00:00.200", "1002, 12, 22"),
    (base_time + 0.3, "12:00:00.300", "1003, 13, 23"),
    (base_time + 0.4, "12:00:00.400", "1004, 14, 24"),
    (base_time + 0.5, "12:00:00.500", "1005, 15, 25"),
    (base_time + 0.6, "12:00:00.600", "1006, 16, 26"),
    (base_time + 0.7, "12:00:00.700", "1007, 17, 27"),
    (base_time + 0.8, "12:00:00.800", "1008, 18, 28"),
    (base_time + 0.9, "12:00:00.900", "1009, 19, 29"),
]

with state.buffer_lock:
    for item in test_data:
        state.data_buffer.append(item)

print(f"Added {len(test_data)} test data entries")
print(f"Buffer size: {len(state.data_buffer)}")

# 测试1: 查询所有数据
print("\n=== Test 1: Query all data ===")
result = _query_data_impl()
print(result)
print(f"\nLines in result: {len(result.split(chr(10)))}")

# 测试2: 时间窗口查询 (start_time)
print("\n=== Test 2: Time window (start_time) ===")
result = _query_data_impl(start_time=base_time + 0.3)
print(result)
lines = [
    l
    for l in result.split("\n")
    if l
    and not l.startswith("-")
    and not l.startswith("data")
    and not l.startswith("start")
    and not l.startswith("end")
]
print(f"Data rows: {len(lines) - 1}")  # -1 for header

# 测试3: 时间窗口查询 (end_time)
print("\n=== Test 3: Time window (end_time) ===")
result = _query_data_impl(end_time=base_time + 0.5)
print(result)
lines = [
    l
    for l in result.split("\n")
    if l
    and not l.startswith("-")
    and not l.startswith("data")
    and not l.startswith("start")
    and not l.startswith("end")
]
print(f"Data rows: {len(lines) - 1}")

# 测试4: 降采样 (max_points=3)
print("\n=== Test 4: Downsample (max_points=3) ===")
result = _query_data_impl(max_points=3)
print(result)
lines = [
    l
    for l in result.split("\n")
    if l
    and not l.startswith("-")
    and not l.startswith("data")
    and not l.startswith("start")
    and not l.startswith("end")
]
print(f"Data rows: {len(lines) - 1}")
assert len(lines) - 1 == 3, f"Expected 3 rows, got {len(lines) - 1}"
print("Downsample test PASSED!")

# 测试5: 空数据查询
print("\n=== Test 5: Empty data query ===")
with state.buffer_lock:
    state.data_buffer.clear()
result = _query_data_impl()
print(result)
assert "data_rows: 0" in result, "Empty data test FAILED!"
print("Empty data test PASSED!")

print("\n=== All tests PASSED! ===")