| # Copyright 2026 The Fuchsia Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import asyncio |
| import io |
| import json |
| import unittest |
| |
| from pydap.client import DapClient, DapError |
| from pydap.dap_types import Source, SourceBreakpoint |
| from pydap.models import * |
| |
| |
| class MockWriter: |
| def __init__(self) -> None: |
| self.buffer = io.BytesIO() |
| |
| def write(self, data: bytes) -> None: |
| self.buffer.write(data) |
| |
| async def drain(self) -> None: |
| pass |
| |
| |
| class TestDapClient(unittest.IsolatedAsyncioTestCase): |
| async def test_read_message(self) -> None: |
| data = b'Content-Length: 26\r\n\r\n{"seq":1,"type":"request"}' |
| reader = asyncio.StreamReader() |
| reader.feed_data(data) |
| reader.feed_eof() |
| |
| client = DapClient() |
| msg = await client._read_message(reader) |
| self.assertIsNotNone(msg) |
| assert msg is not None |
| self.assertEqual(msg["seq"], 1) |
| |
| self.assertEqual(msg["type"], "request") |
| |
| async def test_write_message(self) -> None: |
| client = DapClient() |
| value = {"seq": 1, "type": "request"} |
| |
| writer = MockWriter() |
| await client._write_message(writer, value) # type: ignore |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| self.assertEqual(headers, b"Content-Length: 26") |
| self.assertEqual(json.loads(body.decode("utf-8")), value) |
| |
| async def test__send_request(self) -> None: |
| client = DapClient() |
| |
| writer = MockWriter() |
| |
| send_task = asyncio.create_task( |
| client._send_request(writer, "initialize") # type: ignore |
| ) |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "initialize", |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp["success"]) |
| self.assertEqual(resp["request_seq"], seq) |
| |
| async def test_initialize(self) -> None: |
| client = DapClient() |
| |
| writer = MockWriter() |
| args = InitializeArguments(adapter_id="test") |
| send_task = asyncio.create_task( |
| client.initialize(writer, args) # type: ignore |
| ) |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "initialize", |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| |
| async def test_disconnect(self) -> None: |
| client = DapClient() |
| |
| writer = MockWriter() |
| args = DisconnectArguments(terminate_debuggee=True) |
| send_task = asyncio.create_task( |
| client.disconnect(writer, args) # type: ignore |
| ) |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "disconnect", |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| |
| async def test_stack_trace(self) -> None: |
| client = DapClient() |
| |
| writer = MockWriter() |
| args = StackTraceArguments(thread_id=1) |
| send_task = asyncio.create_task( |
| client.stack_trace(writer, args) # type: ignore |
| ) |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "stackTrace", |
| "body": {"stackFrames": []}, |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertEqual(resp.body.stack_frames, []) |
| |
| async def test_continue_thread(self) -> None: |
| client = DapClient() |
| |
| writer = MockWriter() |
| args = ContinueArguments(thread_id=1) |
| send_task = asyncio.create_task( |
| client.continue_thread(writer, args) # type: ignore |
| ) |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "continue", |
| "body": {"allThreadsContinued": True}, |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertIsInstance(resp, ContinueResponse) |
| self.assertTrue(resp.success) |
| self.assertIsNotNone(resp.body) |
| self.assertTrue(resp.body.all_threads_continued) |
| |
| def test_continue_response_empty_body(self) -> None: |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": 1, |
| "success": True, |
| "command": "continue", |
| "body": {}, |
| } |
| resp = ContinueResponse.model_validate(response) |
| self.assertIsInstance(resp, ContinueResponse) |
| self.assertIsNotNone(resp.body) |
| self.assertTrue(resp.body.all_threads_continued) |
| |
| def test_continue_response_missing_body(self) -> None: |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": 1, |
| "success": True, |
| "command": "continue", |
| } |
| resp = ContinueResponse.model_validate(response) |
| self.assertIsInstance(resp, ContinueResponse) |
| self.assertIsNotNone(resp.body) |
| self.assertTrue(resp.body.all_threads_continued) |
| |
| async def test_pause_thread(self) -> None: |
| client = DapClient() |
| |
| writer = MockWriter() |
| args = PauseArguments(thread_id=1) |
| send_task = asyncio.create_task( |
| client.pause_thread(writer, args) # type: ignore |
| ) |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "pause", |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| |
| async def test_threads(self) -> None: |
| client = DapClient() |
| |
| writer = MockWriter() |
| send_task = asyncio.create_task(client.threads(writer)) # type: ignore |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "threads", |
| "body": { |
| "threads": [ |
| {"id": 1234, "name": "main"}, |
| {"id": 5678, "name": "worker"}, |
| ] |
| }, |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertEqual(len(resp.body.threads), 2) |
| self.assertEqual(resp.body.threads[0].id, 1234) |
| self.assertEqual(resp.body.threads[0].name, "main") |
| |
| async def test_attach(self) -> None: |
| client = DapClient() |
| |
| writer = MockWriter() |
| args = AttachRequestArguments( |
| restart=True, extra_fields={"process": "my_process"} |
| ) |
| send_task = asyncio.create_task(client.attach(writer, args)) # type: ignore |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "attach", |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertTrue(req_val["arguments"]["__restart"]) |
| self.assertEqual(req_val["arguments"]["process"], "my_process") |
| |
| async def test_launch(self) -> None: |
| client = DapClient() |
| writer = MockWriter() |
| args = LaunchArguments(process="my_process", launch_command="run") |
| send_task = asyncio.create_task(client.launch(writer, args)) # type: ignore |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "launch", |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(req_val["arguments"]["process"], "my_process") |
| self.assertEqual(req_val["arguments"]["launchCommand"], "run") |
| |
| async def test_evaluate(self) -> None: |
| client = DapClient() |
| writer = MockWriter() |
| args = EvaluateArguments( |
| expression="1 + 1", context="repl", frame_id=42 |
| ) |
| send_task = asyncio.create_task(client.evaluate(writer, args)) # type: ignore |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "evaluate", |
| "body": { |
| "result": "2", |
| "type": "int", |
| "variablesReference": 0, |
| }, |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(resp.body.result, "2") |
| self.assertEqual(resp.body.type, "int") |
| self.assertEqual(resp.body.variables_reference, 0) |
| self.assertEqual(req_val["arguments"]["expression"], "1 + 1") |
| self.assertEqual(req_val["arguments"]["context"], "repl") |
| self.assertEqual(req_val["arguments"]["frameId"], 42) |
| |
| async def test_scopes(self) -> None: |
| client = DapClient() |
| writer = MockWriter() |
| args = ScopesArguments(frame_id=42) |
| send_task = asyncio.create_task(client.scopes(writer, args)) # type: ignore |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "scopes", |
| "body": { |
| "scopes": [ |
| { |
| "name": "Locals", |
| "variablesReference": 100, |
| "expensive": False, |
| } |
| ] |
| }, |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(len(resp.body.scopes), 1) |
| self.assertEqual(resp.body.scopes[0].name, "Locals") |
| self.assertEqual(resp.body.scopes[0].variables_reference, 100) |
| self.assertFalse(resp.body.scopes[0].expensive) |
| self.assertEqual(req_val["arguments"]["frameId"], 42) |
| |
| async def test_variables(self) -> None: |
| client = DapClient() |
| writer = MockWriter() |
| args = VariablesArguments(variables_reference=100) |
| send_task = asyncio.create_task(client.variables(writer, args)) # type: ignore |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "variables", |
| "body": { |
| "variables": [ |
| { |
| "name": "foo", |
| "value": "bar", |
| "variablesReference": 0, |
| "type": "str", |
| } |
| ] |
| }, |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(len(resp.body.variables), 1) |
| self.assertEqual(resp.body.variables[0].name, "foo") |
| self.assertEqual(resp.body.variables[0].value, "bar") |
| self.assertEqual(resp.body.variables[0].variables_reference, 0) |
| self.assertEqual(resp.body.variables[0].type, "str") |
| self.assertEqual(req_val["arguments"]["variablesReference"], 100) |
| self.assertNotIn("start", req_val["arguments"]) |
| self.assertNotIn("count", req_val["arguments"]) |
| |
| async def test_variables_with_start_only(self) -> None: |
| client = DapClient() |
| writer = MockWriter() |
| args = VariablesArguments(variables_reference=100, start=5) |
| send_task = asyncio.create_task(client.variables(writer, args)) # type: ignore |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "variables", |
| "body": {"variables": []}, |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(req_val["arguments"]["variablesReference"], 100) |
| self.assertEqual(req_val["arguments"]["start"], 5) |
| self.assertNotIn("count", req_val["arguments"]) |
| |
| async def test_variables_with_count_only(self) -> None: |
| client = DapClient() |
| writer = MockWriter() |
| args = VariablesArguments(variables_reference=100, count=10) |
| send_task = asyncio.create_task(client.variables(writer, args)) # type: ignore |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "variables", |
| "body": {"variables": []}, |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(req_val["arguments"]["variablesReference"], 100) |
| self.assertNotIn("start", req_val["arguments"]) |
| self.assertEqual(req_val["arguments"]["count"], 10) |
| |
| async def test_variables_with_start_and_count(self) -> None: |
| client = DapClient() |
| writer = MockWriter() |
| args = VariablesArguments(variables_reference=100, start=5, count=10) |
| send_task = asyncio.create_task(client.variables(writer, args)) # type: ignore |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "variables", |
| "body": {"variables": []}, |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(req_val["arguments"]["variablesReference"], 100) |
| self.assertEqual(req_val["arguments"]["start"], 5) |
| self.assertEqual(req_val["arguments"]["count"], 10) |
| |
| async def test_set_breakpoints(self) -> None: |
| client = DapClient() |
| writer = MockWriter() |
| args = SetBreakpointsArguments( |
| source=Source(path="/path/to/file.rs"), |
| breakpoints=[SourceBreakpoint(line=12)], |
| ) |
| send_task = asyncio.create_task( |
| client.set_breakpoints(writer, args) # type: ignore |
| ) |
| |
| await asyncio.sleep(0.1) |
| |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "setBreakpoints", |
| "body": { |
| "breakpoints": [ |
| { |
| "id": 1, |
| "verified": True, |
| "source": {"path": "/path/to/file.rs"}, |
| "line": 12, |
| } |
| ] |
| }, |
| } |
| |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(len(resp.body.breakpoints), 1) |
| self.assertEqual(resp.body.breakpoints[0].id, 1) |
| self.assertTrue(resp.body.breakpoints[0].verified) |
| self.assertEqual(resp.body.breakpoints[0].line, 12) |
| self.assertEqual( |
| req_val["arguments"]["source"]["path"], "/path/to/file.rs" |
| ) |
| |
| async def test_error_response(self) -> None: |
| client = DapClient() |
| writer = MockWriter() |
| args = SetBreakpointsArguments(source=Source(path="relative.rs")) |
| |
| send_task = asyncio.create_task( |
| client.set_breakpoints(writer, args) # type: ignore |
| ) |
| await asyncio.sleep(0.1) |
| buffer_val = writer.buffer.getvalue() |
| headers, body = buffer_val.split(b"\r\n\r\n", 1) |
| req_val = json.loads(body.decode("utf-8")) |
| seq = req_val["seq"] |
| |
| response = { |
| "seq": 10, |
| "type": "response", |
| "request_seq": seq, |
| "success": False, |
| "command": "setBreakpoints", |
| "message": "SetBreakpointsRequest path must be absolute!", |
| } |
| if seq in client._pending_requests: |
| client._pending_requests[seq].set_result(response) |
| |
| with self.assertRaises(DapError) as cm: |
| await send_task |
| self.assertIn( |
| "SetBreakpointsRequest path must be absolute!", str(cm.exception) |
| ) |
