ci: Added tests

Author: tomlin7

tests/__init__.py

@@ -0,0 +1 @@
+

tests/test_client.py

@@ -0,0 +1,121 @@
+
+import pytest
+from dap.client import DAPClient, ClientState
+from dap.events import InitializedEvent, StoppedEvent, TerminatedEvent
+from dap.protocol import Response
+import json
+
+def _make_response_bytes(seq, request_seq, command, success=True, body=None):
+    content = {
+        "type": "response",
+        "seq": seq,
+        "request_seq": request_seq,
+        "command": command,
+        "success": success,
+    }
+    if body:
+        content["body"] = body
+    
+    encoded = json.dumps(content).encode("utf-8")
+    header = f"Content-Length: {len(encoded)}\r\n\r\n".encode("utf-8")
+    return header + encoded
+
+def _make_event_bytes(seq, event_name, body=None):
+    content = {
+        "type": "event",
+        "seq": seq,
+        "event": event_name,
+    }
+    if body:
+        content["body"] = body
+        
+    encoded = json.dumps(content).encode("utf-8")
+    header = f"Content-Length: {len(encoded)}\r\n\r\n".encode("utf-8")
+    return header + encoded
+
+def test_full_lifecycle():
+    """Test a full debug session lifecycle."""
+    client = DAPClient(clientID="test", clientName="Test")
+    
+    # 1. Initialize
+    init_req_bytes = client.send()
+    # Expect seq 0
+    
+    # Adapter responds to initialize (seq=0)
+    response_bytes = _make_response_bytes(seq=1, request_seq=0, command="initialize", body={"supportsConfigurationDoneRequest": True})
+    events = list(client.recv(response_bytes))
+    
+    assert len(events) == 1
+    assert isinstance(events[0], InitializedEvent)
+    
+    # 2. Launch
+    client.launch(program="main.py")
+    client.send() # flush
+    # Expect seq 1
+    
+    # Adapter responds to launch (seq=1)
+    response_bytes = _make_response_bytes(seq=2, request_seq=1, command="launch")
+    events = list(client.recv(response_bytes))
+    
+    print(f"Launch response event: {events[0]}")
+    assert events[0].event == "response"
+    assert events[0].body["command"] == "launch"
+    
+    # 3. Configuration Done
+    client.configuration_done()
+    client.send() # flush
+    # Expect seq 2
+    
+    response_bytes = _make_response_bytes(seq=3, request_seq=2, command="configurationDone")
+    list(client.recv(response_bytes))
+    
+    # 4. Breakpoint Hit
+    stop_event_bytes = _make_event_bytes(seq=4, event_name="stopped", body={"reason": "breakpoint", "threadId": 1})
+    events = list(client.recv(stop_event_bytes))
+    assert events[0].event == "stopped"
+    
+    # 5. Continue
+    client.continue_execution(threadId=1)
+    client.send()
+    # Expect seq 3
+    
+    response_bytes = _make_response_bytes(seq=5, request_seq=3, command="continue")
+    list(client.recv(response_bytes))
+    
+    # 6. Disconnect
+    client.disconnect()
+    client.send()
+    # Expect seq 4
+    
+    print(f"Unanswered requests before disconnect response: {client._unanswered_requests}")
+    
+    response_bytes = _make_response_bytes(seq=6, request_seq=4, command="disconnect")
+    events = list(client.recv(response_bytes))
+    
+    print(f"Disconnect events: {events}")
+    
+    assert len(events) == 1
+    assert isinstance(events[0], TerminatedEvent) or events[0].event == "terminated"
+
+def test_unsolicited_events():
+    """Test parsing events that come without a request."""
+    client = DAPClient()
+    client.send()
+    
+    out_evt_bytes = _make_event_bytes(seq=1, event_name="output", body={"category": "stdout", "output": "Hello World\n"})
+    events = list(client.recv(out_evt_bytes))
+    
+    assert len(events) == 1
+    assert events[0].event == "output"
+    assert events[0].body["output"] == "Hello World\n"
+
+def test_request_methods():
+    """Test helper methods for sending requests."""
+    client = DAPClient()
+    client.send()
+    client._state = ClientState.NORMAL
+    
+    client.next(threadId=1)
+    req = client.send()
+    assert b'"command": "next"' in req
+    assert b'"threadId": 1' in req

tests/test_io.py

@@ -0,0 +1,73 @@
+
+import pytest
+from dap.client import DAPClient
+
+def _wrap(body: bytes) -> bytes:
+    return f"Content-Length: {len(body)}\r\n\r\n".encode("utf-8") + body
+
+def test_partial_messages():
+    """Test parsing logic when messages arrive in chunks."""
+    client = DAPClient()
+    
+    body = b'{"seq": 1, "type": "event", "event": "stopped"}'
+    msg = _wrap(body)
+    
+    # Split it
+    part1 = msg[:20]
+    part2 = msg[20:]
+    
+    # Feed part 1
+    events = list(client.recv(part1))
+    assert len(events) == 0
+    
+    # Feed part 2
+    events = list(client.recv(part2))
+    assert len(events) == 1
+    assert events[0].event == "stopped"
+
+def test_multiple_messages_in_one_chunk():
+    """Test parsing multiple messages arriving in a single buffer."""
+    client = DAPClient()
+    
+    body1 = b'{"seq": 1, "type": "event", "event": "stopped"}'
+    body2 = b'{"seq": 2, "type": "event", "event": "initialized"}'
+    
+    msg = _wrap(body1) + _wrap(body2)
+    
+    # Feed both at once
+    events = list(client.recv(msg))
+    
+    assert len(events) == 2
+    assert events[0].event == "stopped"
+    assert events[1].event == "initialized"
+
+def test_extra_headers():
+    """Test messages with extra headers like Content-Type are accepted."""
+    client = DAPClient()
+    
+    content = b'{"seq": 1, "type": "event", "event": "stopped"}'
+    # Manual construction ensuring length is correct
+    msg = (
+        f"Content-Length: {len(content)}\r\n"
+        "Content-Type: application/vscode-jsonrpc; charset=utf-8\r\n"
+        "\r\n"
+    ).encode("utf-8") + content
+    
+    events = list(client.recv(msg))
+    assert len(events) == 1
+    assert events[0].event == "stopped"
+
+def test_utf8_encoding():
+    """Test parsing utf-8 content with special characters."""
+    client = DAPClient()
+    
+    body = '{"text": "héllo world 🐛"}'
+    content = (
+        f'{{"seq": 1, "type": "event", "event": "output", "body": {body}}}'
+    ).encode("utf-8")
+    
+    msg = _wrap(content)
+    
+    events = list(client.recv(msg))
+    assert len(events) == 1
+    assert events[0].body["text"] == "héllo world 🐛"