| # 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 typing import Any |
| |
| from pydap.client import DapClient, DapError |
| from pydap.dap_types import Source, SourceBreakpoint |
| from pydap.models import ( |
| AttachRequestArguments, |
| ContinueArguments, |
| ContinueResponse, |
| DisconnectArguments, |
| EvaluateArguments, |
| InitializeArguments, |
| LaunchArguments, |
| NextArguments, |
| PauseArguments, |
| ScopesArguments, |
| SetBreakpointsArguments, |
| StackTraceArguments, |
| StepInArguments, |
| StepOutArguments, |
| VariablesArguments, |
| ) |
| |
| |
| 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 FailingWriter: |
| def write(self, data: bytes) -> None: |
| raise OSError("Connection reset by peer") |
| |
| async def drain(self) -> None: |
| pass |
| |
| |
| def feed_dap_response( |
| reader: asyncio.StreamReader, response: dict[str, Any] |
| ) -> None: |
| body = json.dumps(response, separators=(",", ":")).encode("utf-8") |
| header = f"Content-Length: {len(body)}\r\n\r\n".encode("utf-8") |
| reader.feed_data(header + body) |
| |
| |
| class TestDapClient(unittest.IsolatedAsyncioTestCase): |
| def _start_client( |
| self, client: DapClient |
| ) -> tuple[asyncio.StreamReader, MockWriter]: |
| reader = asyncio.StreamReader() |
| writer = MockWriter() |
| event_queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue() |
| client.run(reader, writer, event_queue) |
| return reader, writer |
| |
| # Should not send request before running the client. |
| async def test_not_running_error(self) -> None: |
| client = DapClient() |
| with self.assertRaises(DapError) as cm: |
| client._send_request_future("initialize") |
| self.assertIn("DapClient is not running", str(cm.exception)) |
| |
| async def test_run(self) -> None: |
| client = DapClient() |
| reader = asyncio.StreamReader() |
| writer = MockWriter() |
| event_queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue() |
| |
| client.run(reader, writer, event_queue) |
| self.assertTrue(client.is_running) |
| |
| reader.feed_eof() |
| await client.close() |
| self.assertFalse(client.is_running) |
| |
| async def test_already_running_error(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| with self.assertRaises(DapError) as cm: |
| client.run(reader, writer, asyncio.Queue()) |
| self.assertIn("DapClient is already running", str(cm.exception)) |
| |
| 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) |
| |
| 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() |
| reader, writer = self._start_client(client) |
| |
| send_task = asyncio.create_task(client._send_request("initialize")) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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", |
| } |
| |
| feed_dap_response(reader, response) |
| |
| resp = await send_task |
| self.assertTrue(resp["success"]) |
| self.assertEqual(resp["request_seq"], seq) |
| |
| async def test_initialize(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| args = InitializeArguments(adapter_id="test") |
| send_task = asyncio.create_task(client.initialize(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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", |
| } |
| |
| feed_dap_response(reader, response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| |
| async def test_disconnect(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| args = DisconnectArguments(terminate_debuggee=True) |
| send_task = asyncio.create_task(client.disconnect(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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", |
| } |
| |
| feed_dap_response(reader, response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| |
| async def test_stack_trace(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| args = StackTraceArguments(thread_id=1) |
| send_task = asyncio.create_task(client.stack_trace(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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": []}, |
| } |
| |
| feed_dap_response(reader, response) |
| |
| resp = await send_task |
| self.assertEqual(resp.body.stack_frames, []) |
| |
| async def test_stack_trace_presentation_hint(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| args = StackTraceArguments(thread_id=1) |
| send_task = asyncio.create_task(client.stack_trace(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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": [ |
| { |
| "id": 1, |
| "name": "frame1", |
| "line": 10, |
| "column": 1, |
| "presentationHint": "subtle", |
| "source": { |
| "origin": "Rust panic", |
| }, |
| } |
| ] |
| }, |
| } |
| |
| feed_dap_response(reader, response) |
| |
| resp = await send_task |
| self.assertEqual(len(resp.body.stack_frames), 1) |
| frame = resp.body.stack_frames[0] |
| self.assertEqual(frame.presentation_hint, "subtle") |
| self.assertIsNotNone(frame.source) |
| assert frame.source is not None |
| self.assertEqual(frame.source.origin, "Rust panic") |
| |
| async def test_continue_thread(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| args = ContinueArguments(thread_id=1) |
| send_task = asyncio.create_task(client.continue_thread(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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}, |
| } |
| |
| feed_dap_response(reader, 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() |
| reader, writer = self._start_client(client) |
| |
| args = PauseArguments(thread_id=1) |
| send_task = asyncio.create_task(client.pause_thread(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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", |
| } |
| |
| feed_dap_response(reader, response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| |
| async def test_step_out(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| args = StepOutArguments( |
| thread_id=1, single_thread=True, granularity="line" |
| ) |
| send_task = asyncio.create_task(client.step_out(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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": "stepOut", |
| } |
| |
| feed_dap_response(reader, response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(req_val["arguments"]["threadId"], 1) |
| self.assertTrue(req_val["arguments"]["singleThread"]) |
| self.assertEqual(req_val["arguments"]["granularity"], "line") |
| |
| async def test_next(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| args = NextArguments( |
| thread_id=1, single_thread=True, granularity="line" |
| ) |
| send_task = asyncio.create_task(client.next(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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": "next", |
| } |
| |
| feed_dap_response(reader, response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(req_val["arguments"]["threadId"], 1) |
| self.assertTrue(req_val["arguments"]["singleThread"]) |
| self.assertEqual(req_val["arguments"]["granularity"], "line") |
| |
| async def test_next_minimal(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| args = NextArguments(thread_id=2) |
| send_task = asyncio.create_task(client.next(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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": "next", |
| } |
| |
| feed_dap_response(reader, response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(req_val["arguments"]["threadId"], 2) |
| self.assertNotIn("singleThread", req_val["arguments"]) |
| self.assertNotIn("granularity", req_val["arguments"]) |
| |
| async def test_step_in(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| args = StepInArguments( |
| thread_id=1, single_thread=True, target_id=0, granularity="line" |
| ) |
| send_task = asyncio.create_task(client.step_in(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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": "stepIn", |
| } |
| |
| feed_dap_response(reader, response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(req_val["arguments"]["threadId"], 1) |
| self.assertTrue(req_val["arguments"]["singleThread"]) |
| self.assertEqual(req_val["arguments"]["targetId"], 0) |
| self.assertEqual(req_val["arguments"]["granularity"], "line") |
| |
| async def test_step_in_minimal(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| args = StepInArguments(thread_id=2) |
| send_task = asyncio.create_task(client.step_in(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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": 11, |
| "type": "response", |
| "request_seq": seq, |
| "success": True, |
| "command": "stepIn", |
| } |
| |
| feed_dap_response(reader, response) |
| |
| resp = await send_task |
| self.assertTrue(resp.success) |
| self.assertEqual(req_val["arguments"]["threadId"], 2) |
| self.assertNotIn("singleThread", req_val["arguments"]) |
| self.assertNotIn("targetId", req_val["arguments"]) |
| self.assertNotIn("granularity", req_val["arguments"]) |
| |
| async def test_step_in_error(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| args = StepInArguments(thread_id=1) |
| send_task = asyncio.create_task(client.step_in(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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": 12, |
| "type": "response", |
| "request_seq": seq, |
| "success": False, |
| "command": "stepIn", |
| "message": "Thread not stopped", |
| } |
| |
| feed_dap_response(reader, response) |
| |
| with self.assertRaises(DapError) as ctx: |
| await send_task |
| self.assertIn("Thread not stopped", str(ctx.exception)) |
| |
| async def test_threads(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| send_task = asyncio.create_task(client.threads()) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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"}, |
| ] |
| }, |
| } |
| |
| feed_dap_response(reader, 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() |
| reader, writer = self._start_client(client) |
| |
| args = AttachRequestArguments( |
| restart=True, extra_fields={"process": "my_process"} |
| ) |
| send_task = asyncio.create_task(client.attach(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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", |
| } |
| |
| feed_dap_response(reader, 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() |
| reader, writer = self._start_client(client) |
| args = LaunchArguments(process="my_process", launch_command="run") |
| send_task = asyncio.create_task(client.launch(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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", |
| } |
| |
| feed_dap_response(reader, 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() |
| reader, writer = self._start_client(client) |
| args = EvaluateArguments( |
| expression="1 + 1", context="repl", frame_id=42 |
| ) |
| send_task = asyncio.create_task(client.evaluate(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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, |
| }, |
| } |
| |
| feed_dap_response(reader, 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() |
| reader, writer = self._start_client(client) |
| args = ScopesArguments(frame_id=42) |
| send_task = asyncio.create_task(client.scopes(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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, |
| } |
| ] |
| }, |
| } |
| |
| feed_dap_response(reader, 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() |
| reader, writer = self._start_client(client) |
| args = VariablesArguments(variables_reference=100) |
| send_task = asyncio.create_task(client.variables(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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", |
| } |
| ] |
| }, |
| } |
| |
| feed_dap_response(reader, 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() |
| reader, writer = self._start_client(client) |
| args = VariablesArguments(variables_reference=100, start=5) |
| send_task = asyncio.create_task(client.variables(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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": []}, |
| } |
| |
| feed_dap_response(reader, 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() |
| reader, writer = self._start_client(client) |
| args = VariablesArguments(variables_reference=100, count=10) |
| send_task = asyncio.create_task(client.variables(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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": []}, |
| } |
| |
| feed_dap_response(reader, 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() |
| reader, writer = self._start_client(client) |
| args = VariablesArguments(variables_reference=100, start=5, count=10) |
| send_task = asyncio.create_task(client.variables(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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": []}, |
| } |
| |
| feed_dap_response(reader, 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() |
| reader, writer = self._start_client(client) |
| args = SetBreakpointsArguments( |
| source=Source(path="/path/to/file.rs"), |
| breakpoints=[SourceBreakpoint(line=12)], |
| ) |
| send_task = asyncio.create_task(client.set_breakpoints(args)) |
| |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| 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, |
| } |
| ] |
| }, |
| } |
| |
| feed_dap_response(reader, 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() |
| reader, writer = self._start_client(client) |
| args = SetBreakpointsArguments(source=Source(path="relative.rs")) |
| |
| send_task = asyncio.create_task(client.set_breakpoints(args)) |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| 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!", |
| } |
| |
| feed_dap_response(reader, response) |
| |
| with self.assertRaises(DapError) as cm: |
| await send_task |
| self.assertIn( |
| "SetBreakpointsRequest path must be absolute!", str(cm.exception) |
| ) |
| |
| async def test_reader_failure_fails_pending_futures(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| seq, sent_fut, data_fut = client._send_request_future("initialize") |
| await asyncio.sleep(0) |
| await client._write_queue.join() |
| |
| # Feed corrupted header length to cause _read_message to raise DapError |
| reader.feed_data(b"Content-Length: invalid\r\n\r\n") |
| |
| with self.assertRaises(DapError): |
| await data_fut |
| |
| self.assertEqual(len(client._pending_requests), 0) |
| self.assertFalse(client.is_running) |
| |
| async def test_writer_failure_fails_pending_future(self) -> None: |
| client = DapClient() |
| reader = asyncio.StreamReader() |
| writer = FailingWriter() |
| event_queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue() |
| client.run(reader, writer, event_queue) |
| |
| seq, sent_fut, data_fut = client._send_request_future("initialize") |
| |
| with self.assertRaises(OSError) as cm_sent: |
| await sent_fut |
| self.assertIn("Connection reset by peer", str(cm_sent.exception)) |
| |
| with self.assertRaises(OSError) as cm_data: |
| await data_fut |
| self.assertIn("Connection reset by peer", str(cm_data.exception)) |
| |
| async def test_close_cancels_write_queue_futures(self) -> None: |
| client = DapClient() |
| reader, writer = self._start_client(client) |
| |
| seq, sent_fut, data_fut = client._send_request_future("initialize") |
| |
| await client.close() |
| |
| self.assertTrue(sent_fut.cancelled()) |
| self.assertTrue(data_fut.cancelled()) |
| self.assertFalse(client.is_running) |
| |
| async def test_caller_cancelled_data_fut_cancels_sent_fut(self) -> None: |
| client = DapClient() |
| reader = asyncio.StreamReader() |
| writer = MockWriter() |
| event_queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue() |
| |
| client.run(reader, writer, event_queue) |
| seq, sent_fut, data_fut = client._send_request_future("initialize") |
| # Cancel data_fut before client processes the write queue |
| data_fut.cancel() |
| |
| await client._write_queue.join() |
| |
| self.assertTrue(sent_fut.cancelled()) |
| # Verify nothing was written to writer |
| self.assertEqual(len(writer.buffer.getvalue()), 0) |
| await client.close() |
