test_integration.py

"""
Task 6: Integration Test Script
测试完整流程: list -> open -> start -> send -> stop -> query -> close
"""

import time
from main import _query_data_impl, state, SerialStateManager, HEADERS


def test_error_handling():
    """测试错误处理"""
    print("=" * 60)
    print("ERROR HANDLING TEST")
    print("=" * 60)

    # 确保状态是初始状态
    state.serial_conn = None
    state.is_capturing = False
    state.capture_thread = None
    with state.buffer_lock:
        state.data_buffer.clear()

    # 1. 未打开串口时的操作 - 使用MCP工具测试
    print("\n1. Testing operations without open port:")
    print("   (These will be tested via MCP tools)")

    # 2. 测试 query_data 内部实现
    print("\n2. Testing _query_data_impl (empty buffer):")
    result = _query_data_impl()
    print(f"   Result: {result[:100]}...")
    assert "data_rows: 0" in result, f"Expected data_rows: 0, got: {result}"
    print("   PASS: Empty buffer returns correct format")


def test_data_flow():
    """测试数据流程"""
    print("\n" + "=" * 60)
    print("DATA FLOW TEST")
    print("=" * 60)

    # 1. 清空缓冲区
    print("\n1. Clearing buffer:")
    with state.buffer_lock:
        state.data_buffer.clear()
    print("   Buffer cleared")

    # 2. 注入测试数据
    print("\n2. Injecting test data:")
    base_time = time.time()
    test_data = []
    for i in range(10):
        ts = base_time + i * 0.1
        time_str = time.strftime("%H:%M:%S", time.localtime(ts))
        ms = int((ts % 1) * 1000)
        formatted_time = f"{time_str}.{ms:03d}"
        test_data.append((ts, formatted_time, f"{1000 + i}, {i}, {i * 2}"))

    with state.buffer_lock:
        for item in test_data:
            state.data_buffer.append(item)
    print(f"   Injected {len(test_data)} entries")

    # 3. 查询所有数据
    print("\n3. Query all data:")
    result = _query_data_impl()
    lines = result.split("\n")
    print(f"   Total lines: {len(lines)}")
    assert "data_rows: 10" in result, "Should have 10 data rows"
    print("   PASS: All data retrieved")

    # 4. 降采样测试
    print("\n4. Downsample test (max_points=5):")
    result = _query_data_impl(max_points=5)
    # 解析YAML中的 data_rows
    import re

    match = re.search(r"data_rows:\s*(\d+)", result)
    if match:
        data_count = int(match.group(1))
    else:
        data_count = 0
    print(f"   Data rows: {data_count}")
    assert data_count == 5, f"Expected 5 rows, got {data_count}"
    print("   PASS: Downsample works correctly")

    # 5. 时间窗口测试
    print("\n5. Time window test:")
    mid_time = base_time + 0.5
    result = _query_data_impl(start_time=mid_time)
    assert "data_rows:" in result
    print(f"   Result contains data_rows info")
    print("   PASS: Time window filtering works")


def test_singleton():
    """测试单例模式"""
    print("\n" + "=" * 60)
    print("SINGLETON TEST")
    print("=" * 60)

    print("\n1. Testing SerialStateManager singleton:")
    state1 = SerialStateManager()
    state2 = SerialStateManager()
    assert state1 is state2, "Singleton should return same instance"
    print("   PASS: Singleton works correctly")

    print("\n2. Testing state attributes:")
    print(f"   serial_conn: {state.serial_conn}")
    print(f"   is_capturing: {state.is_capturing}")
    print(f"   buffer maxlen: {state.data_buffer.maxlen}")
    print("   PASS: All attributes accessible")


def test_format_compliance():
    """测试输出格式合规性"""
    print("\n" + "=" * 60)
    print("FORMAT COMPLIANCE TEST")
    print("=" * 60)

    # 清空并注入数据
    with state.buffer_lock:
        state.data_buffer.clear()

    base_time = time.time()
    with state.buffer_lock:
        state.data_buffer.append((base_time, "12:34:56.789", "1000, 10, 20"))

    result = _query_data_impl()
    print("\n1. Output format check:")
    print(f"   Result:\n{result}")

    # 检查格式
    assert "---" in result, "Missing YAML delimiter"
    assert "data_rows:" in result, "Missing data_rows field"
    assert "recv_time" in result, "Missing recv_time header"
    assert HEADERS[0] in result, f"Missing {HEADERS[0]} header"
    print("   PASS: Format compliant with specification")


def main():
    print("\n" + "=" * 60)
    print("TASK 6: ERROR HANDLING & INTEGRATION TEST")
    print("=" * 60)

    # 运行测试
    try:
        test_error_handling()
        test_data_flow()
        test_singleton()
        test_format_compliance()

        print("\n" + "=" * 60)
        print("ALL TESTS PASSED!")
        print("=" * 60)
        return True

    except AssertionError as e:
        print(f"\nTEST FAILED: {e}")
        return False
    except Exception as e:
        print(f"\nTEST FAILED with exception: {e}")
        import traceback

        traceback.print_exc()
        return False


if __name__ == "__main__":
    success = main()
    exit(0 if success else 1)