9556aa08a7
This commit introduces a comprehensive suite of unit tests for the py_debug_graph module, which is responsible for building debug graphs in Python. The tests cover various scenarios including linked lists, cycles, scalar values, and the handling of local and global variables in the debugging context. Additionally, it verifies the behavior of the debug_graph function and ensures that the builtins.root patch works as intended. These tests aim to enhance the reliability and robustness of the debugging functionality.
567 lines
29 KiB
Python
567 lines
29 KiB
Python
"""Unit tests for scripts/dump_debug_graph.py.
|
|
|
|
Run with:
|
|
~/py314/bin/python -m pytest tests/ --cov=dump_debug_graph --cov-report=term-missing
|
|
|
|
The gdb/debugpy-facing functions (run_gdb, run_debugpy) are exercised here
|
|
against fake subprocesses/sockets rather than real gdb or debugpy, so the
|
|
suite runs fast and needs no toolchain installed. The real toolchain path
|
|
is still covered end-to-end by running scripts/dump_debug_graph.py itself
|
|
(see its own module docstring) -- these tests are for the parsing/
|
|
orchestration logic, not a replacement for that integration check.
|
|
"""
|
|
import json
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
import dump_debug_graph as ddg
|
|
|
|
# Real raw MI `value="..."` payloads, captured verbatim from an actual
|
|
# `gdb --interpreter=mi` / rust-gdb run against the built cpp/rust binaries
|
|
# (see git history of this file for how). Both go through
|
|
# parse_like_extension's fallback ("whole thing is itself a JSON string
|
|
# literal") branch, since gdb reports a `char*`/`&str` return value the
|
|
# same way regardless of the source language.
|
|
CPP_RAW_MI_VALUE = '\\"{\\\\\\"kind\\\\\\": {\\\\\\"nodeTable\\\\\\": true}, \\\\\\"nodes\\\\\\": [{\\\\\\"id\\\\\\": \\\\\\"0x55555556b020\\\\\\", \\\\\\"fields\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"data\\\\\\", \\\\\\"value\\\\\\": \\\\\\"10\\\\\\", \\\\\\"isPointer\\\\\\": false, \\\\\\"typeHint\\\\\\": \\\\\\"int\\\\\\"}, {\\\\\\"name\\\\\\": \\\\\\"next\\\\\\", \\\\\\"value\\\\\\": \\\\\\"0x0\\\\\\", \\\\\\"isPointer\\\\\\": true, \\\\\\"typeHint\\\\\\": \\\\\\"Node<int> *\\\\\\"}], \\\\\\"type\\\\\\": \\\\\\"Node<int>\\\\\\"}, {\\\\\\"id\\\\\\": \\\\\\"0x55555556b040\\\\\\", \\\\\\"fields\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"data\\\\\\", \\\\\\"value\\\\\\": \\\\\\"20\\\\\\", \\\\\\"isPointer\\\\\\": false, \\\\\\"typeHint\\\\\\": \\\\\\"int\\\\\\"}, {\\\\\\"name\\\\\\": \\\\\\"next\\\\\\", \\\\\\"value\\\\\\": \\\\\\"0x55555556b020\\\\\\", \\\\\\"isPointer\\\\\\": true, \\\\\\"typeHint\\\\\\": \\\\\\"Node<int> *\\\\\\"}], \\\\\\"type\\\\\\": \\\\\\"Node<int>\\\\\\"}, {\\\\\\"id\\\\\\": \\\\\\"0x55555556b060\\\\\\", \\\\\\"fields\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"data\\\\\\", \\\\\\"value\\\\\\": \\\\\\"30\\\\\\", \\\\\\"isPointer\\\\\\": false, \\\\\\"typeHint\\\\\\": \\\\\\"int\\\\\\"}, {\\\\\\"name\\\\\\": \\\\\\"next\\\\\\", \\\\\\"value\\\\\\": \\\\\\"0x55555556b040\\\\\\", \\\\\\"isPointer\\\\\\": true, \\\\\\"typeHint\\\\\\": \\\\\\"Node<int> *\\\\\\"}], \\\\\\"type\\\\\\": \\\\\\"Node<int>\\\\\\"}, {\\\\\\"id\\\\\\": \\\\\\"0x7fffffffd478\\\\\\", \\\\\\"fields\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"head_\\\\\\", \\\\\\"value\\\\\\": \\\\\\"0x55555556b060\\\\\\", \\\\\\"isPointer\\\\\\": true, \\\\\\"typeHint\\\\\\": \\\\\\"Node<int> *\\\\\\"}], \\\\\\"type\\\\\\": \\\\\\"LinkedList<int>\\\\\\"}, {\\\\\\"id\\\\\\": \\\\\\"0x555555556008\\\\\\", \\\\\\"fields\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"value\\\\\\", \\\\\\"value\\\\\\": \\\\\\"hello world\\\\\\", \\\\\\"isPointer\\\\\\": false, \\\\\\"typeHint\\\\\\": \\\\\\"char\\\\\\"}], \\\\\\"type\\\\\\": \\\\\\"char *\\\\\\"}], \\\\\\"edges\\\\\\": [{\\\\\\"from\\\\\\": \\\\\\"0x7fffffffd478\\\\\\", \\\\\\"to\\\\\\": \\\\\\"0x55555556b060\\\\\\", \\\\\\"label\\\\\\": \\\\\\"head_\\\\\\"}, {\\\\\\"from\\\\\\": \\\\\\"0x55555556b060\\\\\\", \\\\\\"to\\\\\\": \\\\\\"0x55555556b040\\\\\\", \\\\\\"label\\\\\\": \\\\\\"next\\\\\\"}, {\\\\\\"from\\\\\\": \\\\\\"0x55555556b040\\\\\\", \\\\\\"to\\\\\\": \\\\\\"0x55555556b020\\\\\\", \\\\\\"label\\\\\\": \\\\\\"next\\\\\\"}], \\\\\\"roots\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"top\\\\\\", \\\\\\"value\\\\\\": \\\\\\"0x55555556b060\\\\\\", \\\\\\"kind\\\\\\": \\\\\\"local\\\\\\", \\\\\\"type\\\\\\": \\\\\\"const int *\\\\\\"}, {\\\\\\"name\\\\\\": \\\\\\"count\\\\\\", \\\\\\"value\\\\\\": \\\\\\"3\\\\\\", \\\\\\"kind\\\\\\": \\\\\\"local\\\\\\", \\\\\\"type\\\\\\": \\\\\\"int\\\\\\"}, {\\\\\\"name\\\\\\": \\\\\\"message1\\\\\\", \\\\\\"value\\\\\\": \\\\\\"0x555555556008\\\\\\", \\\\\\"kind\\\\\\": \\\\\\"local\\\\\\", \\\\\\"type\\\\\\": \\\\\\"char *\\\\\\"}]}\\"'
|
|
|
|
RUST_RAW_MI_VALUE = '\\"{\\\\\\"kind\\\\\\": {\\\\\\"nodeTable\\\\\\": true}, \\\\\\"nodes\\\\\\": [{\\\\\\"id\\\\\\": \\\\\\"0x5555555add50\\\\\\", \\\\\\"fields\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"data\\\\\\", \\\\\\"value\\\\\\": \\\\\\"10\\\\\\", \\\\\\"isPointer\\\\\\": false, \\\\\\"typeHint\\\\\\": \\\\\\"i32\\\\\\"}, {\\\\\\"name\\\\\\": \\\\\\"next\\\\\\", \\\\\\"value\\\\\\": \\\\\\"core::option::Option<alloc::boxed::Box<list_example::Node<i32>, alloc::alloc::Global>>::None\\\\\\", \\\\\\"isPointer\\\\\\": false, \\\\\\"typeHint\\\\\\": \\\\\\"core::option::Option<alloc::boxed::Box<list_example::Node<i32>, alloc::alloc::Global>>\\\\\\"}], \\\\\\"type\\\\\\": \\\\\\"list_example::Node<i32>\\\\\\"}, {\\\\\\"id\\\\\\": \\\\\\"0x5555555add70\\\\\\", \\\\\\"fields\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"data\\\\\\", \\\\\\"value\\\\\\": \\\\\\"20\\\\\\", \\\\\\"isPointer\\\\\\": false, \\\\\\"typeHint\\\\\\": \\\\\\"i32\\\\\\"}, {\\\\\\"name\\\\\\": \\\\\\"next\\\\\\", \\\\\\"value\\\\\\": \\\\\\"0x5555555add50\\\\\\", \\\\\\"isPointer\\\\\\": true, \\\\\\"typeHint\\\\\\": \\\\\\"core::option::Option<alloc::boxed::Box<list_example::Node<i32>, alloc::alloc::Global>>\\\\\\"}], \\\\\\"type\\\\\\": \\\\\\"list_example::Node<i32>\\\\\\"}, {\\\\\\"id\\\\\\": \\\\\\"0x5555555add90\\\\\\", \\\\\\"fields\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"data\\\\\\", \\\\\\"value\\\\\\": \\\\\\"30\\\\\\", \\\\\\"isPointer\\\\\\": false, \\\\\\"typeHint\\\\\\": \\\\\\"i32\\\\\\"}, {\\\\\\"name\\\\\\": \\\\\\"next\\\\\\", \\\\\\"value\\\\\\": \\\\\\"0x5555555add70\\\\\\", \\\\\\"isPointer\\\\\\": true, \\\\\\"typeHint\\\\\\": \\\\\\"core::option::Option<alloc::boxed::Box<list_example::Node<i32>, alloc::alloc::Global>>\\\\\\"}], \\\\\\"type\\\\\\": \\\\\\"list_example::Node<i32>\\\\\\"}, {\\\\\\"id\\\\\\": \\\\\\"0x7fffffffd0a0\\\\\\", \\\\\\"fields\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"head\\\\\\", \\\\\\"value\\\\\\": \\\\\\"0x5555555add90\\\\\\", \\\\\\"isPointer\\\\\\": true, \\\\\\"typeHint\\\\\\": \\\\\\"core::option::Option<alloc::boxed::Box<list_example::Node<i32>, alloc::alloc::Global>>\\\\\\"}], \\\\\\"type\\\\\\": \\\\\\"list_example::LinkedList<i32>\\\\\\"}, {\\\\\\"id\\\\\\": \\\\\\"0x5555555add98\\\\\\", \\\\\\"fields\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"value\\\\\\", \\\\\\"value\\\\\\": \\\\\\"30\\\\\\", \\\\\\"isPointer\\\\\\": false, \\\\\\"typeHint\\\\\\": \\\\\\"i32\\\\\\"}], \\\\\\"type\\\\\\": \\\\\\"*mut i32\\\\\\"}, {\\\\\\"id\\\\\\": \\\\\\"0x7fffffffd0b0\\\\\\", \\\\\\"fields\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"value\\\\\\", \\\\\\"value\\\\\\": \\\\\\"\\\\\\\\\\\\\\"hello world\\\\\\\\\\\\\\"\\\\\\", \\\\\\"isPointer\\\\\\": false, \\\\\\"typeHint\\\\\\": \\\\\\"&str\\\\\\"}], \\\\\\"type\\\\\\": \\\\\\"&str\\\\\\"}], \\\\\\"edges\\\\\\": [{\\\\\\"from\\\\\\": \\\\\\"0x7fffffffd0a0\\\\\\", \\\\\\"to\\\\\\": \\\\\\"0x5555555add90\\\\\\", \\\\\\"label\\\\\\": \\\\\\"head\\\\\\"}, {\\\\\\"from\\\\\\": \\\\\\"0x5555555add90\\\\\\", \\\\\\"to\\\\\\": \\\\\\"0x5555555add70\\\\\\", \\\\\\"label\\\\\\": \\\\\\"next\\\\\\"}, {\\\\\\"from\\\\\\": \\\\\\"0x5555555add70\\\\\\", \\\\\\"to\\\\\\": \\\\\\"0x5555555add50\\\\\\", \\\\\\"label\\\\\\": \\\\\\"next\\\\\\"}], \\\\\\"roots\\\\\\": [{\\\\\\"name\\\\\\": \\\\\\"top\\\\\\", \\\\\\"value\\\\\\": \\\\\\"0x5555555add98\\\\\\", \\\\\\"kind\\\\\\": \\\\\\"local\\\\\\", \\\\\\"type\\\\\\": \\\\\\"*mut i32\\\\\\"}, {\\\\\\"name\\\\\\": \\\\\\"message1\\\\\\", \\\\\\"value\\\\\\": \\\\\\"0x7fffffffd0b0\\\\\\", \\\\\\"kind\\\\\\": \\\\\\"local\\\\\\", \\\\\\"type\\\\\\": \\\\\\"&str\\\\\\"}, {\\\\\\"name\\\\\\": \\\\\\"count\\\\\\", \\\\\\"value\\\\\\": \\\\\\"3\\\\\\", \\\\\\"kind\\\\\\": \\\\\\"local\\\\\\", \\\\\\"type\\\\\\": \\\\\\"i32\\\\\\"}]}\\"'
|
|
|
|
# Real raw `body.result` string, captured verbatim from an actual debugpy
|
|
# `evaluate` response: debugpy's repr() of the JSON string debug_graph()
|
|
# returned.
|
|
PY_RAW_EVALUATE_RESULT = '\'{"kind": {"nodeTable": true}, "nodes": [{"id": "0x7c61539b30e0", "fields": [{"name": "data", "value": "10", "isPointer": false, "typeHint": "int"}, {"name": "next", "value": "None", "isPointer": false, "typeHint": "NoneType"}], "type": "Node"}, {"id": "0x7c61539c87d0", "fields": [{"name": "data", "value": "20", "isPointer": false, "typeHint": "int"}, {"name": "next", "value": "0x7c61539b30e0", "isPointer": true, "typeHint": "Node"}], "type": "Node"}, {"id": "0x7c61539c9450", "fields": [{"name": "data", "value": "30", "isPointer": false, "typeHint": "int"}, {"name": "next", "value": "0x7c61539c87d0", "isPointer": true, "typeHint": "Node"}], "type": "Node"}, {"id": "0x7c61539b2f90", "fields": [{"name": "_head", "value": "0x7c61539c9450", "isPointer": true, "typeHint": "Node"}], "type": "LinkedList"}], "edges": [{"from": "0x7c61539b2f90", "to": "0x7c61539c9450", "label": "_head"}, {"from": "0x7c61539c9450", "to": "0x7c61539c87d0", "label": "next"}, {"from": "0x7c61539c87d0", "to": "0x7c61539b30e0", "label": "next"}], "roots": [{"name": "count", "value": "3", "kind": "local", "type": "int"}, {"name": "message1", "value": "hello world", "kind": "local", "type": "str"}, {"name": "top", "value": "30", "kind": "local", "type": "int"}, {"name": "Optional", "value": "typing.Optional", "kind": "global", "type": "_SpecialForm"}, {"name": "T", "value": "~T", "kind": "global", "type": "TypeVar"}]}\''
|
|
|
|
|
|
# --- is_enclosed_with -------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("s,ch,expected", [
|
|
('"hello"', '"', True),
|
|
("'hello'", "'", True),
|
|
('"hello"', "'", False),
|
|
('"', '"', False), # single char: startswith and endswith the same char, but len < 2
|
|
("", '"', False),
|
|
('"a', '"', False),
|
|
])
|
|
def test_is_enclosed_with(s, ch, expected):
|
|
assert ddg.is_enclosed_with(s, ch) is expected
|
|
|
|
|
|
# --- parse_like_extension ----------------------------------------------
|
|
|
|
def test_parse_like_extension_cpp_real_capture():
|
|
parsed = ddg.parse_like_extension(CPP_RAW_MI_VALUE)
|
|
assert parsed["ok"] is True
|
|
data = parsed["data"]
|
|
assert data["kind"] == {"nodeTable": True}
|
|
node_shaped = [n for n in data["nodes"] if [f["name"] for f in n["fields"]] == ["data", "next"]]
|
|
assert len(node_shaped) == 3
|
|
assert {r["name"] for r in data["roots"]} == {"top", "count", "message1"}
|
|
|
|
|
|
def test_parse_like_extension_rust_real_capture():
|
|
parsed = ddg.parse_like_extension(RUST_RAW_MI_VALUE)
|
|
assert parsed["ok"] is True
|
|
data = parsed["data"]
|
|
roots_by_name = {r["name"]: r for r in data["roots"]}
|
|
assert roots_by_name["message1"]["type"] == "&str"
|
|
# the &str node itself resolved the fat pointer to the real text
|
|
str_node = next(n for n in data["nodes"] if n["type"] == "&str")
|
|
assert str_node["fields"][0]["value"] == '"hello world"'
|
|
|
|
|
|
def test_parse_like_extension_simple_quoted_json():
|
|
"""The `is_enclosed_with` fast path: a plain string value that, once
|
|
MI-unescaped, is itself already valid JSON wrapped in one layer of
|
|
quotes (no double string-literal nesting)."""
|
|
payload = json.dumps({"kind": {"nodeTable": True}, "nodes": [], "edges": [], "roots": []})
|
|
mi_value = '\\"%s\\"' % payload.replace('"', '\\\\\\"')
|
|
parsed = ddg.parse_like_extension(mi_value)
|
|
assert parsed["ok"] is True
|
|
assert parsed["data"]["nodes"] == []
|
|
|
|
|
|
def test_parse_like_extension_malformed_reports_error_and_raw():
|
|
parsed = ddg.parse_like_extension("not valid json at all")
|
|
assert parsed["ok"] is False
|
|
assert "error" in parsed
|
|
assert parsed["raw"] == "not valid json at all"
|
|
|
|
|
|
# --- parse_python_result -------------------------------------------------
|
|
|
|
def test_parse_python_result_real_capture():
|
|
parsed = ddg.parse_python_result(PY_RAW_EVALUATE_RESULT)
|
|
assert parsed["ok"] is True
|
|
data = parsed["data"]
|
|
roots_by_name = {r["name"]: r for r in data["roots"]}
|
|
assert roots_by_name["count"]["value"] == "3"
|
|
assert roots_by_name["top"]["kind"] == "local"
|
|
assert roots_by_name["Optional"]["kind"] == "global"
|
|
|
|
|
|
def test_parse_python_result_malformed_reports_error_and_raw():
|
|
parsed = ddg.parse_python_result("this is not a repr'd string")
|
|
assert parsed["ok"] is False
|
|
assert "error" in parsed
|
|
assert parsed["raw"] == "this is not a repr'd string"
|
|
|
|
|
|
def test_parse_python_result_valid_repr_but_not_json():
|
|
parsed = ddg.parse_python_result(repr("not json"))
|
|
assert parsed["ok"] is False
|
|
assert parsed["raw"] == repr("not json")
|
|
|
|
|
|
# --- verify_top_delves_into_list / _py ----------------------------------
|
|
|
|
def _node(id_, *field_names):
|
|
return {"id": id_, "fields": [{"name": n} for n in field_names]}
|
|
|
|
|
|
def test_verify_top_delves_into_list_success():
|
|
data = {
|
|
"nodes": [_node("0x1", "data", "next"), _node("0x2", "data", "next")],
|
|
"edges": [{"label": "next"}, {"label": "next"}],
|
|
"roots": [{"name": "list", "kind": "local"}],
|
|
}
|
|
assert ddg.verify_top_delves_into_list(data) is True
|
|
|
|
|
|
def test_verify_top_delves_into_list_fails_without_node_shaped_node():
|
|
data = {"nodes": [_node("0x1", "value")], "edges": [], "roots": [{"name": "list", "kind": "local"}]}
|
|
assert ddg.verify_top_delves_into_list(data) is False
|
|
|
|
|
|
def test_verify_top_delves_into_list_fails_with_too_few_next_edges():
|
|
data = {
|
|
"nodes": [_node("0x1", "data", "next")],
|
|
"edges": [{"label": "next"}],
|
|
"roots": [{"name": "list", "kind": "local"}],
|
|
}
|
|
assert ddg.verify_top_delves_into_list(data) is False
|
|
|
|
|
|
def test_verify_top_delves_into_list_fails_without_list_root():
|
|
data = {
|
|
"nodes": [_node("0x1", "data", "next")],
|
|
"edges": [{"label": "next"}, {"label": "next"}],
|
|
"roots": [{"name": "something_else", "kind": "local"}],
|
|
}
|
|
assert ddg.verify_top_delves_into_list(data) is False
|
|
|
|
|
|
def test_verify_top_delves_into_list_py_success():
|
|
data = {
|
|
"nodes": [_node("0x1", "data", "next"), _node("0x2", "data", "next")],
|
|
"edges": [{"label": "next"}, {"label": "next"}],
|
|
"roots": [{"name": "lst", "kind": "local"}],
|
|
}
|
|
assert ddg.verify_top_delves_into_list_py(data) is True
|
|
|
|
|
|
def test_verify_top_delves_into_list_py_fails_without_lst_root():
|
|
data = {
|
|
"nodes": [_node("0x1", "data", "next")],
|
|
"edges": [{"label": "next"}, {"label": "next"}],
|
|
"roots": [{"name": "list", "kind": "local"}], # note: cpp/rust name, not python's
|
|
}
|
|
assert ddg.verify_top_delves_into_list_py(data) is False
|
|
|
|
|
|
def test_verify_top_delves_into_list_py_fails_without_node_shaped_node():
|
|
data = {"nodes": [_node("0x1", "value")], "edges": [], "roots": [{"name": "lst", "kind": "local"}]}
|
|
assert ddg.verify_top_delves_into_list_py(data) is False
|
|
|
|
|
|
def test_verify_top_delves_into_list_py_fails_with_too_few_next_edges():
|
|
data = {
|
|
"nodes": [_node("0x1", "data", "next")],
|
|
"edges": [{"label": "next"}],
|
|
"roots": [{"name": "lst", "kind": "local"}],
|
|
}
|
|
assert ddg.verify_top_delves_into_list_py(data) is False
|
|
|
|
|
|
# --- _write_dump ----------------------------------------------------------
|
|
|
|
def test_write_dump_writes_indented_json(tmp_path):
|
|
ddg._write_dump("mylabel", tmp_path, {"nodes": [], "roots": []})
|
|
out_path = tmp_path / "debug_mylabel.json"
|
|
assert out_path.exists()
|
|
assert json.loads(out_path.read_text()) == {"nodes": [], "roots": []}
|
|
|
|
|
|
def test_write_dump_reports_parse_error_in_output(tmp_path, capsys):
|
|
ddg._write_dump("bad", tmp_path, {"PARSE_ERROR": "boom"})
|
|
captured = capsys.readouterr()
|
|
assert "PARSE_ERROR" in captured.out
|
|
|
|
|
|
# --- find_debugpy_python --------------------------------------------------
|
|
|
|
def test_find_debugpy_python_prefers_env_override(monkeypatch):
|
|
monkeypatch.setenv("DUMP_DEBUG_GRAPH_PYTHON", "/fake/python")
|
|
seen = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
seen.append(cmd[0])
|
|
return subprocess.CompletedProcess(cmd, 0)
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
result = ddg.find_debugpy_python()
|
|
assert result == "/fake/python"
|
|
assert seen[0] == "/fake/python"
|
|
|
|
|
|
def test_find_debugpy_python_falls_through_candidates(monkeypatch):
|
|
monkeypatch.delenv("DUMP_DEBUG_GRAPH_PYTHON", raising=False)
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
if cmd[0] != "/only/this/one/works":
|
|
raise subprocess.CalledProcessError(1, cmd)
|
|
return subprocess.CompletedProcess(cmd, 0)
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
monkeypatch.setattr(ddg.shutil, "which", lambda name: "/only/this/one/works")
|
|
result = ddg.find_debugpy_python()
|
|
assert result == "/only/this/one/works"
|
|
|
|
|
|
def test_find_debugpy_python_raises_when_none_found(monkeypatch):
|
|
monkeypatch.delenv("DUMP_DEBUG_GRAPH_PYTHON", raising=False)
|
|
monkeypatch.setattr(subprocess, "run", lambda cmd, **kw: (_ for _ in ()).throw(FileNotFoundError()))
|
|
monkeypatch.setattr(ddg.shutil, "which", lambda name: None)
|
|
with pytest.raises(RuntimeError):
|
|
ddg.find_debugpy_python()
|
|
|
|
|
|
def test_find_debugpy_python_skips_falsy_candidates(monkeypatch):
|
|
"""sys.executable can (rarely) be empty; the candidate loop must skip
|
|
it rather than passing "" to subprocess.run."""
|
|
monkeypatch.delenv("DUMP_DEBUG_GRAPH_PYTHON", raising=False)
|
|
monkeypatch.setattr(ddg.sys, "executable", "")
|
|
monkeypatch.setattr(ddg.shutil, "which", lambda name: "/found/python3")
|
|
seen = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
seen.append(cmd[0])
|
|
if cmd[0] != "/found/python3":
|
|
raise subprocess.CalledProcessError(1, cmd)
|
|
return subprocess.CompletedProcess(cmd, 0)
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
result = ddg.find_debugpy_python()
|
|
assert result == "/found/python3"
|
|
assert "" not in seen
|
|
|
|
|
|
# --- _DapClient wire protocol ----------------------------------------------
|
|
|
|
class _FakeSocket:
|
|
"""Feeds back pre-queued bytes to _DapClient.recv(), split arbitrarily
|
|
to prove _read_message() reassembles a frame from partial reads."""
|
|
|
|
def __init__(self, chunks):
|
|
self._chunks = list(chunks)
|
|
self.sent = b""
|
|
|
|
def sendall(self, data):
|
|
self.sent += data
|
|
|
|
def recv(self, bufsize):
|
|
if not self._chunks:
|
|
return b""
|
|
return self._chunks.pop(0)
|
|
|
|
|
|
def _frame(msg):
|
|
body = json.dumps(msg).encode("utf-8")
|
|
return ("Content-Length: %d\r\n\r\n" % len(body)).encode("ascii") + body
|
|
|
|
|
|
def test_dap_client_send_frames_content_length_correctly():
|
|
sock = _FakeSocket([])
|
|
client = ddg._DapClient(sock)
|
|
seq = client.send("initialize", {"foo": "bar"})
|
|
assert seq == 1
|
|
header, _, rest = sock.sent.partition(b"\r\n\r\n")
|
|
length = int(header.split(b":")[1].strip())
|
|
assert len(rest) == length
|
|
msg = json.loads(rest.decode())
|
|
assert msg == {"seq": 1, "type": "request", "command": "initialize", "arguments": {"foo": "bar"}}
|
|
|
|
|
|
def test_dap_client_reassembles_message_split_across_recv_calls():
|
|
whole = _frame({"seq": 1, "type": "response", "command": "initialize", "success": True})
|
|
# split the single frame into three arbitrary chunks to exercise the
|
|
# partial-read loop in both the header and body phases
|
|
chunks = [whole[:10], whole[10:40], whole[40:]]
|
|
sock = _FakeSocket(chunks)
|
|
client = ddg._DapClient(sock)
|
|
msg = client.wait_for(lambda m: m.get("command") == "initialize")
|
|
assert msg["success"] is True
|
|
|
|
|
|
def test_dap_client_wait_for_skips_non_matching_messages():
|
|
chunks = [
|
|
_frame({"seq": 1, "type": "event", "event": "output"}),
|
|
_frame({"seq": 2, "type": "response", "command": "evaluate", "request_seq": 5}),
|
|
]
|
|
sock = _FakeSocket(chunks)
|
|
client = ddg._DapClient(sock)
|
|
msg = client.wait_for(lambda m: m.get("command") == "evaluate")
|
|
assert msg["request_seq"] == 5
|
|
|
|
|
|
def test_dap_client_read_message_raises_eof_on_closed_socket():
|
|
sock = _FakeSocket([]) # recv() immediately returns b""
|
|
client = ddg._DapClient(sock)
|
|
with pytest.raises(EOFError):
|
|
client._read_message()
|
|
|
|
|
|
def test_dap_client_read_message_raises_eof_when_body_truncated():
|
|
"""Header parses fine (so length is known), but the socket closes
|
|
before the full body arrives -- the second (body) read loop's EOF."""
|
|
whole = _frame({"seq": 1, "type": "response", "command": "initialize"})
|
|
header_end = whole.index(b"\r\n\r\n") + 4
|
|
truncated_body = whole[header_end:header_end + 3] # a few body bytes, not all
|
|
sock = _FakeSocket([whole[:header_end] + truncated_body])
|
|
client = ddg._DapClient(sock)
|
|
with pytest.raises(EOFError):
|
|
client._read_message()
|
|
|
|
|
|
def test_dap_client_wait_for_times_out(monkeypatch):
|
|
sock = _FakeSocket([_frame({"seq": 1, "type": "event", "event": "output"})] * 5)
|
|
client = ddg._DapClient(sock)
|
|
with pytest.raises(TimeoutError):
|
|
client.wait_for(lambda m: False, timeout=0)
|
|
|
|
|
|
# --- run_gdb ---------------------------------------------------------------
|
|
|
|
def test_run_gdb_extracts_last_done_value(monkeypatch):
|
|
mi_output = (
|
|
'(gdb)\n'
|
|
'^done\n'
|
|
'^done,value="first"\n'
|
|
'^done,value="second"\n'
|
|
).encode()
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
assert cmd[0] == "gdb"
|
|
return subprocess.CompletedProcess(cmd, 0, stdout=mi_output)
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
result = ddg.run_gdb("gdb", "/bin/true", "main.cpp", 10)
|
|
assert result == "second"
|
|
|
|
|
|
def test_run_gdb_raises_when_no_done_value(monkeypatch):
|
|
mi_output = b'^error,msg="No symbol table is loaded"\n'
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
return subprocess.CompletedProcess(cmd, 0, stdout=mi_output)
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
with pytest.raises(RuntimeError, match="No symbol table"):
|
|
ddg.run_gdb("gdb", "/bin/true", "main.cpp", 10)
|
|
|
|
|
|
# --- run_debugpy (fully mocked DAP session) --------------------------------
|
|
|
|
class _FakeDapServerSocket:
|
|
"""Plays the debugpy side of the handshake in-memory: parses each
|
|
framed request dump_debug_graph.run_debugpy sends and queues back
|
|
exactly the response/event sequence a real debugpy adapter sends for
|
|
a --wait-for-client launch that immediately hits one breakpoint."""
|
|
|
|
def __init__(self, evaluate_result):
|
|
self._evaluate_result = evaluate_result
|
|
self._out = b""
|
|
self._server_seq = 100
|
|
self._pending_attach_seq = None
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def _emit(self, msg):
|
|
self._server_seq += 1
|
|
msg = dict(msg, seq=self._server_seq)
|
|
self._out += _frame(msg)
|
|
|
|
def sendall(self, data):
|
|
header, _, rest = data.partition(b"\r\n\r\n")
|
|
length = int(header.split(b":")[1].strip())
|
|
req = json.loads(rest[:length].decode())
|
|
cmd, seq = req["command"], req["seq"]
|
|
|
|
if cmd == "initialize":
|
|
self._emit({"type": "response", "command": cmd, "request_seq": seq, "success": True})
|
|
self._emit({"type": "event", "event": "initialized"})
|
|
elif cmd == "attach":
|
|
self._pending_attach_seq = seq # debugpy defers this until after configurationDone
|
|
elif cmd == "setBreakpoints":
|
|
self._emit({"type": "response", "command": cmd, "request_seq": seq, "success": True})
|
|
elif cmd == "configurationDone":
|
|
self._emit({"type": "response", "command": cmd, "request_seq": seq, "success": True})
|
|
self._emit({"type": "response", "command": "attach", "request_seq": self._pending_attach_seq, "success": True})
|
|
self._emit({"type": "event", "event": "stopped", "body": {"threadId": 1, "reason": "breakpoint"}})
|
|
elif cmd == "stackTrace":
|
|
self._emit({
|
|
"type": "response", "command": cmd, "request_seq": seq, "success": True,
|
|
"body": {"stackFrames": [{"id": 7, "name": "main"}]},
|
|
})
|
|
elif cmd == "evaluate":
|
|
self._emit({
|
|
"type": "response", "command": cmd, "request_seq": seq, "success": True,
|
|
"body": {"result": self._evaluate_result},
|
|
})
|
|
elif cmd == "continue":
|
|
self._emit({"type": "response", "command": cmd, "request_seq": seq, "success": True})
|
|
else:
|
|
raise AssertionError("unexpected DAP command in test: %r" % cmd)
|
|
|
|
def recv(self, bufsize):
|
|
chunk, self._out = self._out[:bufsize], self._out[bufsize:]
|
|
return chunk
|
|
|
|
|
|
class _FakeProcess:
|
|
def __init__(self):
|
|
self.killed = False
|
|
|
|
def wait(self, timeout=None):
|
|
return 0
|
|
|
|
def kill(self):
|
|
self.killed = True
|
|
|
|
|
|
def test_run_debugpy_drives_full_handshake_and_returns_evaluate_result(monkeypatch):
|
|
fake_socket = _FakeDapServerSocket(evaluate_result="'{\"ok\": true}'")
|
|
fake_proc = _FakeProcess()
|
|
|
|
monkeypatch.setattr(ddg.subprocess, "Popen", lambda *a, **kw: fake_proc)
|
|
monkeypatch.setattr(ddg.socket, "create_connection", lambda addr, timeout=None: fake_socket)
|
|
|
|
result = ddg.run_debugpy("python3", "/tmp/main.py", 54, "debug_graph(lst)")
|
|
assert result == "'{\"ok\": true}'"
|
|
|
|
|
|
def test_run_debugpy_raises_when_listen_socket_never_opens(monkeypatch):
|
|
monkeypatch.setattr(ddg.subprocess, "Popen", lambda *a, **kw: _FakeProcess())
|
|
monkeypatch.setattr(ddg.time, "sleep", lambda s: None) # don't actually wait out the 50 retries
|
|
|
|
def never_connects(addr, timeout=None):
|
|
raise OSError("connection refused")
|
|
|
|
monkeypatch.setattr(ddg.socket, "create_connection", never_connects)
|
|
with pytest.raises(RuntimeError, match="never opened its listen socket"):
|
|
ddg.run_debugpy("python3", "/tmp/main.py", 54, "debug_graph(lst)")
|
|
|
|
|
|
def test_run_debugpy_kills_process_if_wait_times_out_on_cleanup(monkeypatch):
|
|
fake_socket = _FakeDapServerSocket(evaluate_result="'{}'")
|
|
|
|
class _SlowProcess(_FakeProcess):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.wait_calls = 0
|
|
|
|
def wait(self, timeout=None):
|
|
self.wait_calls += 1
|
|
if self.wait_calls == 1:
|
|
raise subprocess.TimeoutExpired(cmd="debugpy", timeout=timeout)
|
|
return 0
|
|
|
|
fake_proc = _SlowProcess()
|
|
monkeypatch.setattr(ddg.subprocess, "Popen", lambda *a, **kw: fake_proc)
|
|
monkeypatch.setattr(ddg.socket, "create_connection", lambda addr, timeout=None: fake_socket)
|
|
|
|
result = ddg.run_debugpy("python3", "/tmp/main.py", 54, "debug_graph(lst)")
|
|
assert result == "'{}'"
|
|
assert fake_proc.killed is True
|
|
assert fake_proc.wait_calls == 2
|
|
|
|
|
|
def test_run_debugpy_raises_when_evaluate_fails(monkeypatch):
|
|
class _FailingEvalSocket(_FakeDapServerSocket):
|
|
def sendall(self, data):
|
|
header, _, rest = data.partition(b"\r\n\r\n")
|
|
length = int(header.split(b":")[1].strip())
|
|
req = json.loads(rest[:length].decode())
|
|
if req["command"] == "evaluate":
|
|
self._emit({
|
|
"type": "response", "command": "evaluate", "request_seq": req["seq"],
|
|
"success": False, "message": "bad expression",
|
|
})
|
|
else:
|
|
super().sendall(data)
|
|
|
|
fake_socket = _FailingEvalSocket(evaluate_result="unused")
|
|
monkeypatch.setattr(ddg.subprocess, "Popen", lambda *a, **kw: _FakeProcess())
|
|
monkeypatch.setattr(ddg.socket, "create_connection", lambda addr, timeout=None: fake_socket)
|
|
|
|
with pytest.raises(RuntimeError, match="bad expression"):
|
|
ddg.run_debugpy("python3", "/tmp/main.py", 54, "bad expr(")
|
|
|
|
|
|
# --- dump / dump_python orchestration ---------------------------------------
|
|
|
|
def test_dump_success_path_writes_verified_result(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(ddg, "run_gdb", lambda *a, **kw: CPP_RAW_MI_VALUE)
|
|
ok = ddg.dump("cpp", "gdb", "/bin/true", "main.cpp", 70, tmp_path, verify=lambda data: True)
|
|
assert ok is True
|
|
assert json.loads((tmp_path / "debug_cpp.json").read_text())["kind"] == {"nodeTable": True}
|
|
|
|
|
|
def test_dump_reports_failure_when_verify_rejects(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(ddg, "run_gdb", lambda *a, **kw: CPP_RAW_MI_VALUE)
|
|
ok = ddg.dump("cpp", "gdb", "/bin/true", "main.cpp", 70, tmp_path, verify=lambda data: False)
|
|
assert ok is False
|
|
|
|
|
|
def test_dump_reports_failure_when_gdb_invocation_raises(tmp_path, monkeypatch):
|
|
def fake_run_gdb(*a, **kw):
|
|
raise RuntimeError("gdb exploded")
|
|
|
|
monkeypatch.setattr(ddg, "run_gdb", fake_run_gdb)
|
|
ok = ddg.dump("cpp", "gdb", "/bin/true", "main.cpp", 70, tmp_path)
|
|
assert ok is False
|
|
written = json.loads((tmp_path / "debug_cpp.json").read_text())
|
|
assert written["PARSE_ERROR"] == "gdb invocation failed"
|
|
assert "gdb exploded" in written["detail"]
|
|
|
|
|
|
def test_dump_reports_failure_when_result_unparseable(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(ddg, "run_gdb", lambda *a, **kw: "not json at all")
|
|
ok = ddg.dump("cpp", "gdb", "/bin/true", "main.cpp", 70, tmp_path)
|
|
assert ok is False
|
|
written = json.loads((tmp_path / "debug_cpp.json").read_text())
|
|
assert "PARSE_ERROR" in written
|
|
|
|
|
|
def test_dump_python_success_path_writes_verified_result(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(ddg, "run_debugpy", lambda *a, **kw: PY_RAW_EVALUATE_RESULT)
|
|
ok = ddg.dump_python("python", "python3", "/tmp/main.py", 54, tmp_path, verify=lambda data: True)
|
|
assert ok is True
|
|
assert json.loads((tmp_path / "debug_python.json").read_text())["kind"] == {"nodeTable": True}
|
|
|
|
|
|
def test_dump_python_reports_failure_when_debugpy_invocation_raises(tmp_path, monkeypatch):
|
|
def fake_run_debugpy(*a, **kw):
|
|
raise RuntimeError("debugpy exploded")
|
|
|
|
monkeypatch.setattr(ddg, "run_debugpy", fake_run_debugpy)
|
|
ok = ddg.dump_python("python", "python3", "/tmp/main.py", 54, tmp_path)
|
|
assert ok is False
|
|
written = json.loads((tmp_path / "debug_python.json").read_text())
|
|
assert written["PARSE_ERROR"] == "debugpy invocation failed"
|
|
assert "debugpy exploded" in written["detail"]
|