chuanqi/cq2_sdk/tests/unit/test_recorder.py

"""Tests for the recording infrastructure (Recorder + RecordedSession)."""
import asyncio
import json
import pytest
from pathlib import Path

from cq2_sdk.testing import Recorder, RecordedSession
from cq2_sdk.transport.tcp import TcpClient
from cq2_sdk.transport.framing import WIRE_HEADER_SIZE, decode_header, encode_frame


async def _scripted_echo_server(reader, writer, scripted_sc):
    """Echo back SC frames from a scripted list, in order."""
    try:
        idx = 0
        while True:
            try:
                hdr_raw = await reader.readexactly(WIRE_HEADER_SIZE)
            except asyncio.IncompleteReadError:
                break
            header = decode_header(hdr_raw)
            body = await reader.readexactly(header.body_len) if header.body_len else b""
            # Echo the CS frame back as an SC
            sc = scripted_sc[idx] if idx < len(scripted_sc) else None
            idx += 1
            if sc is None:
                break
            sc_msg_id, sc_recog, sc_body = sc
            resp = encode_frame(sc_msg_id, sc_recog, 0, 0, 0, sc_body)
            writer.write(resp)
            await writer.drain()
    finally:
        writer.close()


@pytest.fixture
async def echo_server():
    scripted_sc = [
        (529, 1, b"lk/uid"),
        (530, 999, b"ok"),
        (520, 0, b'{"list":[{"chr":1,"name":"a"}]}'),
    ]
    state = {"server": None}

    async def handler(reader, writer):
        await _scripted_echo_server(reader, writer, scripted_sc)

    server = await asyncio.start_server(handler, "127.0.0.1", 0)
    state["server"] = server
    port = server.sockets[0].getsockname()[1]
    yield port
    server.close()
    await server.wait_closed()


@pytest.mark.asyncio
async def test_recorder_logs_send_and_recv(echo_server, tmp_path):
    tcp = TcpClient("127.0.0.1", echo_server)
    await tcp.connect()
    out = tmp_path / "session.jsonl"
    recorder = Recorder(tcp, out)
    recorder.install()
    try:
        await tcp.send(2001, recog=900000, param1=0, param2=0, param3=0, body=b"attach1")
        hdr, body = await tcp.recv(timeout=1.0)
        assert hdr.msg_id == 529
    finally:
        recorder.close()
        await tcp.close()

    # Verify file
    lines = out.read_text(encoding="utf-8").strip().splitlines()
    assert len(lines) == 2

    send_event = json.loads(lines[0])
    assert send_event["dir"] == "cs"
    assert send_event["msg_id"] == 2001
    assert send_event["recog"] == 900000
    assert send_event["body"] == b"attach1".hex()

    recv_event = json.loads(lines[1])
    assert recv_event["dir"] == "sc"
    assert recv_event["msg_id"] == 529
    assert recv_event["body"] == b"lk/uid".hex()


@pytest.mark.asyncio
async def test_recorder_handles_empty_body(echo_server, tmp_path):
    tcp = TcpClient("127.0.0.1", echo_server)
    await tcp.connect()
    out = tmp_path / "session.jsonl"
    recorder = Recorder(tcp, out)
    recorder.install()
    try:
        await tcp.send(60000, recog=0, param1=0, param2=0, param3=0, body=b"")
    finally:
        recorder.close()
        await tcp.close()

    line = out.read_text(encoding="utf-8").strip()
    event = json.loads(line)
    assert event["body"] == ""


def test_recorded_session_reads_events(tmp_path):
    p = tmp_path / "session.jsonl"
    events = [
        {"dir": "cs", "msg_id": 2001, "recog": 0, "p1": 0, "p2": 0, "p3": 0, "body": ""},
        {"dir": "sc", "msg_id": 529, "recog": 1, "p1": 0, "p2": 0, "p3": 0, "body": "ab"},
        {"dir": "cs", "msg_id": 100, "recog": 0, "p1": 0, "p2": 0, "p3": 0, "body": ""},
    ]
    p.write_text("\n".join(json.dumps(e) for e in events) + "\n", encoding="utf-8")

    session = RecordedSession(p)
    assert len(session.events) == 3
    assert len(session.sent()) == 2
    assert len(session.received()) == 1
    assert session.received()[0]["msg_id"] == 529


def test_recorded_session_empty_file(tmp_path):
    p = tmp_path / "empty.jsonl"
    p.write_text("", encoding="utf-8")
    session = RecordedSession(p)
    assert session.events == []


def test_recorded_session_skips_blank_lines(tmp_path):
    p = tmp_path / "blanks.jsonl"
    content = (
        json.dumps({"dir": "cs", "msg_id": 1, "recog": 0, "p1": 0, "p2": 0, "p3": 0, "body": ""})
        + "\n\n"
        + json.dumps({"dir": "sc", "msg_id": 2, "recog": 0, "p1": 0, "p2": 0, "p3": 0, "body": ""})
        + "\n"
    )
    p.write_text(content, encoding="utf-8")
    session = RecordedSession(p)
    assert len(session.events) == 2