diff --git a/src/rdc/_skills/references/commands-quick-ref.md b/src/rdc/_skills/references/commands-quick-ref.md index 23a2392..b04dee3 100644 --- a/src/rdc/_skills/references/commands-quick-ref.md +++ b/src/rdc/_skills/references/commands-quick-ref.md @@ -142,6 +142,17 @@ Export buffer raw data. | `-o, --output` | Write to file | path | | | `--raw` | Force raw output even on TTY | flag | | +## `rdc callstacks` + +Resolve CPU callstack for an event. + +**Options:** + +| Flag | Help | Type | Default | +|------|------|------|---------| +| `--eid` | Event ID (defaults to current). | integer | | +| `--json` | Output as JSON. | flag | | + ## `rdc capture` Execute application and capture a frame. @@ -806,7 +817,7 @@ Search shader disassembly text for PATTERN (regex). ## `rdc section` -Extract named section contents. +Extract or write named section contents. **Arguments:** @@ -819,6 +830,7 @@ Extract named section contents. | Flag | Help | Type | Default | |------|------|------|---------| | `--json` | Output as JSON. | flag | | +| `--write` | Write file contents as this section (use '-' for stdin). | path | | ## `rdc sections` diff --git a/src/rdc/cli.py b/src/rdc/cli.py index 5fc0f53..9960d67 100644 --- a/src/rdc/cli.py +++ b/src/rdc/cli.py @@ -20,6 +20,7 @@ capture_trigger_cmd, ) from rdc.commands.capturefile import ( + callstacks_cmd, gpus_cmd, section_cmd, sections_cmd, @@ -150,6 +151,7 @@ def main() -> None: main.add_command(capture_copy_cmd, name="capture-copy") main.add_command(remote_group, name="remote") main.add_command(serve_cmd, name="serve") +main.add_command(callstacks_cmd, name="callstacks") if __name__ == "__main__": diff --git a/src/rdc/commands/_helpers.py b/src/rdc/commands/_helpers.py index 6850a1a..dcafd71 100644 --- a/src/rdc/commands/_helpers.py +++ b/src/rdc/commands/_helpers.py @@ -97,12 +97,13 @@ def require_session() -> tuple[str, int, str]: return session.host, session.port, session.token -def call(method: str, params: dict[str, Any]) -> dict[str, Any]: +def call(method: str, params: dict[str, Any], *, timeout: float = 30.0) -> dict[str, Any]: """Send a JSON-RPC request to the daemon and return the result. Args: method: The JSON-RPC method name. params: Request parameters. + timeout: Socket timeout in seconds. Returns: The result dict from the daemon response. @@ -113,7 +114,7 @@ def call(method: str, params: dict[str, Any]) -> dict[str, Any]: host, port, token = require_session() payload = _request(method, 1, {"_token": token, **params}).to_dict() try: - response = send_request(host, port, payload) + response = send_request(host, port, payload, timeout=timeout) except (OSError, ValueError) as exc: _emit_error(f"daemon unreachable: {exc}") if "error" in response: diff --git a/src/rdc/commands/capturefile.py b/src/rdc/commands/capturefile.py index c9165e5..fe21eec 100644 --- a/src/rdc/commands/capturefile.py +++ b/src/rdc/commands/capturefile.py @@ -1,4 +1,4 @@ -"""CaptureFile CLI commands: thumbnail, gpus, sections, section.""" +"""CaptureFile CLI commands: thumbnail, gpus, sections, section, callstacks.""" from __future__ import annotations @@ -61,10 +61,52 @@ def sections_cmd(use_json: bool) -> None: @click.command("section") @click.argument("name") @click.option("--json", "use_json", is_flag=True, help="Output as JSON.") -def section_cmd(name: str, use_json: bool) -> None: - """Extract named section contents.""" +@click.option( + "--write", + "write_file", + type=click.Path(path_type=Path), + default=None, + help="Write file contents as this section (use '-' for stdin).", +) +def section_cmd(name: str, use_json: bool, write_file: Path | None) -> None: + """Extract or write named section contents.""" + if write_file is not None: + if str(write_file) == "-": + data = click.get_binary_stream("stdin").read() + else: + try: + data = write_file.read_bytes() + except OSError as exc: + click.echo(f"error: {exc}", err=True) + raise SystemExit(1) from exc + encoded = base64.b64encode(data).decode() + result = call("section_write", {"name": name, "data": encoded}) + click.echo(f"section '{name}' written ({result['bytes']} bytes)", err=True) + if use_json: + click.echo(json.dumps(result)) + return result = call("capture_section_content", {"name": name}) if use_json: click.echo(json.dumps(result)) else: click.echo(result["contents"]) + + +@click.command("callstacks") +@click.option("--eid", type=int, default=None, help="Event ID (defaults to current).") +@click.option("--json", "use_json", is_flag=True, help="Output as JSON.") +def callstacks_cmd(eid: int | None, use_json: bool) -> None: + """Resolve CPU callstack for an event.""" + params: dict[str, int] = {} + if eid is not None: + params["eid"] = eid + result = call("callstack_resolve", params, timeout=60.0) + frames = result.get("frames", []) + if use_json: + click.echo(json.dumps(result)) + elif not frames: + click.echo(f"no frames at eid {result.get('eid', '?')}") + else: + click.echo("function\tfile\tline") + for f in frames: + click.echo(f"{f['function']}\t{f['file']}\t{f['line']}") diff --git a/src/rdc/handlers/capturefile.py b/src/rdc/handlers/capturefile.py index 8fda253..e931525 100644 --- a/src/rdc/handlers/capturefile.py +++ b/src/rdc/handlers/capturefile.py @@ -1,8 +1,9 @@ -"""CaptureFile handlers: thumbnail, gpus, sections, section content.""" +"""CaptureFile handlers: thumbnail, gpus, sections, section content, callstacks, write.""" from __future__ import annotations import base64 +import logging from typing import TYPE_CHECKING, Any from rdc.handlers._helpers import _error_response, _result_response @@ -11,6 +12,19 @@ if TYPE_CHECKING: from rdc.daemon_server import DaemonState +_log = logging.getLogger(__name__) + +_PROTECTED_SECTION_TYPES: frozenset[str] = frozenset( + { + "FrameCapture", + "ResolveDatabase", + "ExtendedThumbnail", + "EmbeddedLogfile", + "ResourceRenames", + "AMDRGPProfile", + } +) + def _handle_capture_thumbnail( request_id: int, params: dict[str, Any], state: DaemonState @@ -107,9 +121,134 @@ def _handle_capture_section_content( ), True +def _parse_resolve_string(s: str) -> dict[str, Any]: + """Parse a resolve string like ``'func file.c:42'`` into a frame dict.""" + parts = s.rsplit(" ", 1) + func = parts[0] if len(parts) > 1 else s + file_line = parts[1] if len(parts) > 1 else "" + file_name = "" + line = 0 + if ":" in file_line: + fl_parts = file_line.rsplit(":", 1) + file_name = fl_parts[0] + try: + line = int(fl_parts[1]) + except ValueError: + file_name = file_line + else: + file_name = file_line + return {"function": func, "file": file_name, "line": line} + + +def _handle_callstack_resolve( + request_id: int, params: dict[str, Any], state: DaemonState +) -> tuple[dict[str, Any], bool]: + """Resolve CPU callstack for an event ID.""" + if state.cap is None: + return _error_response(request_id, -32002, "no capture file open"), True + if not state.cap.HasCallstacks(): + return _error_response(request_id, -32002, "no callstacks recorded in this capture"), True + try: + ok = state.cap.InitResolver(interactive=False, progress=None) + except Exception as exc: # noqa: BLE001 + _log.debug("InitResolver failed: %s", exc) + return _error_response(request_id, -32002, "symbols not available"), True + if not ok: + return _error_response(request_id, -32002, "symbols not available"), True + + eid = params.get("eid", state.current_eid) + if not isinstance(eid, int) or eid < 0: + return _error_response(request_id, -32602, "invalid eid"), True + if eid > state.max_eid: + return _error_response( + request_id, -32602, f"eid {eid} out of range (max {state.max_eid})" + ), True + + callstack: list[int] = [] + ctrl = getattr(state.adapter, "controller", None) if state.adapter else None + get_cs = getattr(ctrl, "GetCallstack", None) if ctrl else None + if callable(get_cs): + try: + callstack = list(get_cs(eid)) + except Exception: # noqa: BLE001 + _log.debug("GetCallstack(%d) failed, falling back", eid) + + frames: list[dict[str, Any]] = [] + if callstack: + resolved = state.cap.GetResolve(callstack) + frames = [_parse_resolve_string(s) for s in resolved] + + return _result_response(request_id, {"eid": eid, "frames": frames}), True + + +def _handle_section_write( + request_id: int, params: dict[str, Any], state: DaemonState +) -> tuple[dict[str, Any], bool]: + """Write a named section to the capture file.""" + if state.cap is None: + return _error_response(request_id, -32002, "no capture file open"), True + + name = params.get("name") + if not name or not isinstance(name, str): + return _error_response(request_id, -32602, "missing or empty 'name'"), True + + data_b64 = params.get("data") + if data_b64 is None or not isinstance(data_b64, str): + return _error_response(request_id, -32602, "missing 'data' parameter"), True + + idx = state.cap.FindSectionByName(name) + if idx >= 0: + props = state.cap.GetSectionProperties(idx) + type_name = getattr(getattr(props, "type", None), "name", "") + if type_name in _PROTECTED_SECTION_TYPES: + return _error_response( + request_id, + -32602, + f"cannot overwrite built-in section '{name}'", + ), True + + try: + decoded = base64.b64decode(data_b64) + except Exception: # noqa: BLE001 + return _error_response(request_id, -32602, "invalid base64 data"), True + + rd = state.rd + sec_type = getattr(rd, "SectionType", None) if rd else None + unknown_type = getattr(sec_type, "Unknown", 0) if sec_type else 0 + sec_flags_cls = getattr(rd, "SectionFlags", None) if rd else None + no_flags = getattr(sec_flags_cls, "NoFlags", 0) if sec_flags_cls else 0 + sp_cls = getattr(rd, "SectionProperties", None) if rd else None + + if sp_cls is not None: + new_props = sp_cls() + new_props.name = name + new_props.type = unknown_type + new_props.flags = no_flags + else: + from types import SimpleNamespace + + new_props = SimpleNamespace( + name=name, + type=unknown_type, + version=1, + flags=no_flags, + compressedSize=0, + uncompressedSize=0, + ) + + try: + state.cap.WriteSection(new_props, decoded) + except Exception as exc: # noqa: BLE001 + _log.debug("WriteSection failed: %s", exc) + return _error_response(request_id, -32002, "section write failed"), True + return _result_response(request_id, {"name": name, "bytes": len(decoded)}), True + + HANDLERS: dict[str, Handler] = { "capture_thumbnail": _handle_capture_thumbnail, "capture_gpus": _handle_capture_gpus, "capture_sections": _handle_capture_sections, "capture_section_content": _handle_capture_section_content, + "callstack_resolve": _handle_callstack_resolve, + "section_write": _handle_section_write, } diff --git a/tests/e2e/test_capturefile.py b/tests/e2e/test_capturefile.py new file mode 100644 index 0000000..7458c63 --- /dev/null +++ b/tests/e2e/test_capturefile.py @@ -0,0 +1,93 @@ +"""E2E tests for capturefile commands (callstacks, section write). + +These are black-box tests that invoke the CLI via subprocess and require +a working renderdoc installation for capture open/replay. +""" + +from __future__ import annotations + +import base64 +import uuid + +import pytest +from e2e_helpers import VKCUBE, rdc, rdc_fail, rdc_json, rdc_ok + +pytestmark = pytest.mark.gpu + + +def _uid() -> str: + return uuid.uuid4().hex[:8] + + +class TestCallstacks: + """T-5C-36/37: callstacks on capture without callstack data.""" + + def test_callstacks_no_callstacks(self) -> None: + """``rdc callstacks`` exits 1 on fixture without callstack data.""" + name = f"e2e_cs_{_uid()}" + try: + rdc_ok("open", str(VKCUBE), session=name) + out = rdc_fail("callstacks", session=name, exit_code=1) + assert "no callstacks" in out.lower() + finally: + rdc("close", session=name) + + def test_callstacks_json_no_callstacks(self) -> None: + """``rdc callstacks --json`` exits 1, no partial JSON.""" + name = f"e2e_csj_{_uid()}" + try: + rdc_ok("open", str(VKCUBE), session=name) + out = rdc_fail("callstacks", "--json", session=name, exit_code=1) + assert "no callstacks" in out.lower() + finally: + rdc("close", session=name) + + +class TestSectionWrite: + """T-5C-33/34/35: section write and read-back.""" + + def test_write_and_readback(self, tmp_path) -> None: + """Write a custom section, then read it back.""" + name = f"e2e_sw_{_uid()}" + note = tmp_path / "notes.txt" + note.write_text("e2e-round-trip") + try: + rdc_ok("open", str(VKCUBE), session=name) + rdc_ok("section", "MyNotes", "--write", str(note), session=name) + out = rdc_ok("section", "MyNotes", session=name) + assert "e2e-round-trip" in out + finally: + rdc("close", session=name) + + def test_refuse_system_section(self, tmp_path) -> None: + """Refuse to overwrite real system section by internal name.""" + name = f"e2e_sws_{_uid()}" + note = tmp_path / "bad.txt" + note.write_bytes(b"bad") + try: + rdc_ok("open", str(VKCUBE), session=name) + out = rdc_fail( + "section", + "renderdoc/internal/framecapture", + "--write", + str(note), + session=name, + exit_code=1, + ) + assert "built-in" in out.lower() + finally: + rdc("close", session=name) + + def test_binary_roundtrip(self, tmp_path) -> None: + """Write binary section and verify base64 round-trip.""" + name = f"e2e_swb_{_uid()}" + binfile = tmp_path / "data.bin" + binfile.write_bytes(b"\xde\xad\xbe\xef") + try: + rdc_ok("open", str(VKCUBE), session=name) + rdc_ok("section", "BinSection", "--write", str(binfile), session=name) + data = rdc_json("section", "BinSection", session=name) + assert data["encoding"] == "base64" + assert base64.b64decode(data["contents"]) == b"\xde\xad\xbe\xef" + finally: + rdc("close", session=name) diff --git a/tests/mocks/mock_renderdoc.py b/tests/mocks/mock_renderdoc.py index 89c17bb..01705a7 100644 --- a/tests/mocks/mock_renderdoc.py +++ b/tests/mocks/mock_renderdoc.py @@ -390,6 +390,7 @@ class SectionType(IntEnum): ResourceRenames = 5 AMDRGPProfile = 6 ExtendedThumbnail = 7 + EmbeddedLogfile = 8 class SectionFlags(IntFlag): @@ -1569,6 +1570,10 @@ def ReplaceResource(self, original: Any, replacement: Any) -> None: def RemoveReplacement(self, original: Any) -> None: self._replacements.pop(int(original), None) + def GetCallstack(self, eid: int) -> list[int]: + """Mock GetCallstack -- returns instruction addresses for the event.""" + return [] + def FreeTargetResource(self, rid: Any) -> None: self._freed.add(int(rid)) @@ -1694,6 +1699,9 @@ def __init__(self) -> None: self._structured_data: StructuredFile = StructuredFile() self._path: str = "" self._shutdown_called: bool = False + self._has_callstacks: bool = False + self._resolver_ready: bool = False + self._written_sections: list[tuple[SectionProperties, bytes]] = [] def OpenFile(self, path: str, filetype: str, progress: Any) -> ResultCode: self._path = path @@ -1727,7 +1735,17 @@ def FindSectionByName(self, name: str) -> int: return 0 if name == "FrameCapture" else -1 def HasCallstacks(self) -> bool: - return False + return self._has_callstacks + + def InitResolver(self, interactive: bool = False, progress: Any = None) -> bool: + self._resolver_ready = True + return True + + def GetResolve(self, callstack: list[int]) -> list[str]: + return [f"mock_function mock_file.c:{42 + i}" for i in range(len(callstack))] + + def WriteSection(self, props: SectionProperties, contents: bytes) -> None: + self._written_sections.append((props, contents)) def RecordedMachineIdent(self) -> str: return "mock-machine-ident" diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index e0659d1..25bcd16 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -119,7 +119,7 @@ def patch_cli_session( session = type("S", (), {"host": host, "port": port, "token": token})() monkeypatch.setattr(mod, "load_session", lambda: session) - monkeypatch.setattr(mod, "send_request", lambda _h, _p, _payload: {"result": response}) + monkeypatch.setattr(mod, "send_request", lambda _h, _p, _payload, **_kw: {"result": response}) def assert_json_output(result: Any) -> dict[str, Any]: diff --git a/tests/unit/test_capturefile_commands.py b/tests/unit/test_capturefile_commands.py index 2b457cb..09abfe6 100644 --- a/tests/unit/test_capturefile_commands.py +++ b/tests/unit/test_capturefile_commands.py @@ -2,10 +2,14 @@ from __future__ import annotations +import base64 +from pathlib import Path + from click.testing import CliRunner from conftest import assert_json_output, patch_cli_session from rdc.commands.capturefile import ( + callstacks_cmd, gpus_cmd, section_cmd, sections_cmd, @@ -116,3 +120,223 @@ def test_section_cmd_json(monkeypatch) -> None: result = CliRunner().invoke(section_cmd, ["Notes", "--json"]) data = assert_json_output(result) assert data["encoding"] == "utf-8" + + +# --------------------------------------------------------------------------- +# section --write +# --------------------------------------------------------------------------- + + +def test_section_write_file(monkeypatch, tmp_path: Path) -> None: + """T-5C-26: write a text file to a section.""" + f = tmp_path / "notes.txt" + f.write_bytes(b"hello world") + patch_cli_session(monkeypatch, {"name": "MyNotes", "bytes": 11}) + result = CliRunner().invoke(section_cmd, ["MyNotes", "--write", str(f)]) + assert result.exit_code == 0 + assert "written" in result.output + + +def test_section_write_binary(monkeypatch, tmp_path: Path) -> None: + """T-5C-27: write binary file.""" + f = tmp_path / "data.bin" + f.write_bytes(b"\xff\xfe\x00\x01") + # Capture send_request args + calls: list[dict] = [] + import rdc.commands._helpers as mod + + session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() + monkeypatch.setattr(mod, "load_session", lambda: session) + + def _spy(_h, _p, payload, **kw): + calls.append(payload) + return {"result": {"name": "BinData", "bytes": 4}} + + monkeypatch.setattr(mod, "send_request", _spy) + result = CliRunner().invoke(section_cmd, ["BinData", "--write", str(f)]) + assert result.exit_code == 0 + sent_data = calls[0]["params"]["data"] + assert base64.b64decode(sent_data) == b"\xff\xfe\x00\x01" + + +def test_section_write_json(monkeypatch, tmp_path: Path) -> None: + """T-5C-28: --json on write emits JSON.""" + import json as _json + + f = tmp_path / "notes.txt" + f.write_bytes(b"hello") + patch_cli_session(monkeypatch, {"name": "MyNotes", "bytes": 5}) + result = CliRunner().invoke(section_cmd, ["MyNotes", "--write", str(f), "--json"]) + assert result.exit_code == 0 + # stdout has confirmation on stderr + JSON; extract JSON line + json_lines = [ln for ln in result.output.splitlines() if ln.startswith("{")] + assert json_lines, "no JSON line in output" + data = _json.loads(json_lines[0]) + assert data["name"] == "MyNotes" + assert data["bytes"] == 5 + + +def test_section_write_file_not_found(monkeypatch) -> None: + """T-5C-29: write file does not exist.""" + patch_cli_session(monkeypatch, {"name": "X", "bytes": 0}) + result = CliRunner().invoke(section_cmd, ["MyNotes", "--write", "/nonexistent/path.txt"]) + assert result.exit_code != 0 + + +def test_section_write_no_session(monkeypatch, tmp_path: Path) -> None: + """T-5C-30: no active session.""" + f = tmp_path / "x.txt" + f.write_bytes(b"x") + patch_cli_session(monkeypatch, None) + result = CliRunner().invoke( + section_cmd, + ["MyNotes", "--write", str(f)], + catch_exceptions=False, + ) + assert result.exit_code == 1 + assert "no active session" in result.output + + +def test_section_write_daemon_error(monkeypatch, tmp_path: Path) -> None: + """T-5C-31: daemon refuses write.""" + f = tmp_path / "x.txt" + f.write_bytes(b"x") + import rdc.commands._helpers as mod + + session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() + monkeypatch.setattr(mod, "load_session", lambda: session) + monkeypatch.setattr( + mod, + "send_request", + lambda _h, _p, _payload, **kw: { + "error": {"code": -32602, "message": "cannot overwrite built-in section 'FrameCapture'"} + }, + ) + result = CliRunner().invoke( + section_cmd, + ["FrameCapture", "--write", str(f)], + catch_exceptions=False, + ) + assert result.exit_code == 1 + + +def test_section_read_unchanged(monkeypatch) -> None: + """T-5C-32: read path unchanged without --write.""" + resp = {"name": "Notes", "contents": "hi", "encoding": "utf-8"} + patch_cli_session(monkeypatch, resp) + result = CliRunner().invoke(section_cmd, ["Notes"]) + assert result.exit_code == 0 + assert "hi" in result.output + + +def test_section_write_zero_byte(monkeypatch, tmp_path: Path) -> None: + """T-5C-46: zero-byte file.""" + f = tmp_path / "empty.txt" + f.write_bytes(b"") + calls: list[dict] = [] + import rdc.commands._helpers as mod + + session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() + monkeypatch.setattr(mod, "load_session", lambda: session) + + def _spy(_h, _p, payload, **kw): + calls.append(payload) + return {"result": {"name": "EmptySection", "bytes": 0}} + + monkeypatch.setattr(mod, "send_request", _spy) + result = CliRunner().invoke(section_cmd, ["EmptySection", "--write", str(f)]) + assert result.exit_code == 0 + assert calls[0]["params"]["data"] == "" + + +# --------------------------------------------------------------------------- +# callstacks +# --------------------------------------------------------------------------- + + +def test_callstacks_tsv(monkeypatch) -> None: + """T-5C-20: default TSV output.""" + resp = { + "eid": 0, + "frames": [ + {"function": "main", "file": "app.c", "line": 10}, + {"function": "draw", "file": "render.c", "line": 55}, + ], + } + patch_cli_session(monkeypatch, resp) + result = CliRunner().invoke(callstacks_cmd, []) + assert result.exit_code == 0 + assert "main" in result.output + assert "app.c" in result.output + assert "10" in result.output + assert "function\tfile\tline" in result.output + + +def test_callstacks_json(monkeypatch) -> None: + """T-5C-21: --json flag.""" + resp = { + "eid": 0, + "frames": [{"function": "main", "file": "app.c", "line": 10}], + } + patch_cli_session(monkeypatch, resp) + result = CliRunner().invoke(callstacks_cmd, ["--json"]) + data = assert_json_output(result) + assert data["frames"][0]["function"] == "main" + + +def test_callstacks_eid(monkeypatch) -> None: + """T-5C-22: --eid flag passes eid to daemon.""" + calls: list[dict] = [] + import rdc.commands._helpers as mod + + session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() + monkeypatch.setattr(mod, "load_session", lambda: session) + + def _spy(_h, _p, payload, **kw): + calls.append(payload) + return {"result": {"eid": 42, "frames": []}} + + monkeypatch.setattr(mod, "send_request", _spy) + result = CliRunner().invoke(callstacks_cmd, ["--eid", "42"]) + assert result.exit_code == 0 + assert calls[0]["params"]["eid"] == 42 + + +def test_callstacks_empty(monkeypatch) -> None: + """T-5C-23: empty frames list.""" + patch_cli_session(monkeypatch, {"eid": 0, "frames": []}) + result = CliRunner().invoke(callstacks_cmd, []) + assert result.exit_code == 0 + assert "no frames" in result.output + + +def test_callstacks_no_session(monkeypatch) -> None: + """T-5C-24: no active session.""" + patch_cli_session(monkeypatch, None) + result = CliRunner().invoke(callstacks_cmd, [], catch_exceptions=False) + assert result.exit_code == 1 + assert "no active session" in result.output + + +def test_callstacks_daemon_error(monkeypatch) -> None: + """T-5C-25: daemon returns error.""" + import rdc.commands._helpers as mod + + session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() + monkeypatch.setattr(mod, "load_session", lambda: session) + monkeypatch.setattr( + mod, + "send_request", + lambda _h, _p, _payload, **kw: { + "error": {"code": -32002, "message": "no callstacks in capture"} + }, + ) + result = CliRunner().invoke(callstacks_cmd, [], catch_exceptions=False) + assert result.exit_code == 1 + + +def test_callstacks_eid_non_integer(monkeypatch) -> None: + """T-5C-44: non-integer eid value.""" + patch_cli_session(monkeypatch, {"eid": 0, "frames": []}) + result = CliRunner().invoke(callstacks_cmd, ["--eid", "abc"]) + assert result.exit_code == 2 diff --git a/tests/unit/test_capturefile_handlers.py b/tests/unit/test_capturefile_handlers.py index 8e133d1..4a80202 100644 --- a/tests/unit/test_capturefile_handlers.py +++ b/tests/unit/test_capturefile_handlers.py @@ -151,3 +151,352 @@ def test_section_missing_name(tmp_path: Path) -> None: state = _make_state(tmp_path) resp = _handle("capture_section_content", {}, state) assert resp["error"]["code"] == -32602 + + +# --------------------------------------------------------------------------- +# callstack_resolve +# --------------------------------------------------------------------------- + + +def _make_callstack_state( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + *, + has_callstacks: bool = True, + callstack_addrs: list[int] | None = None, + resolve_strings: list[str] | None = None, +) -> Any: + """Build a state with callstack support configured.""" + state = _make_state(tmp_path) + state.cap._has_callstacks = has_callstacks + if callstack_addrs is not None: + assert state.adapter is not None + state.adapter.controller.GetCallstack = lambda eid: callstack_addrs # type: ignore[union-attr] + if resolve_strings is not None: + monkeypatch.setattr( + state.cap, + "GetResolve", + lambda cs: resolve_strings, + ) + return state + + +def test_callstack_resolve_no_cap(tmp_path: Path) -> None: + """T-5C-07: no capture open.""" + state = make_daemon_state(tmp_path=tmp_path, rd=mock_rd) + resp = _handle("callstack_resolve", {}, state) + assert resp["error"]["code"] == -32002 + + +def test_callstack_resolve_no_callstacks(tmp_path: Path) -> None: + """T-5C-03: capture has no callstacks.""" + state = _make_state(tmp_path) + resp = _handle("callstack_resolve", {}, state) + assert resp["error"]["code"] == -32002 + assert "no callstacks" in resp["error"]["message"].lower() + + +def test_callstack_resolve_init_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-04: InitResolver returns falsy.""" + state = _make_state(tmp_path) + state.cap._has_callstacks = True + monkeypatch.setattr(state.cap, "InitResolver", lambda **_kw: False) + resp = _handle("callstack_resolve", {}, state) + assert resp["error"]["code"] == -32002 + assert "symbols" in resp["error"]["message"].lower() + + +def test_callstack_resolve_init_raises( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-04 variant: InitResolver raises.""" + state = _make_state(tmp_path) + state.cap._has_callstacks = True + + def _raise(**_kw: Any) -> bool: + raise RuntimeError("no PDB") + + monkeypatch.setattr(state.cap, "InitResolver", _raise) + resp = _handle("callstack_resolve", {}, state) + assert resp["error"]["code"] == -32002 + assert "symbols" in resp["error"]["message"].lower() + + +def test_callstack_resolve_default_eid( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-01: defaults to current_eid, resolve with mock frames.""" + state = _make_callstack_state( + tmp_path, + monkeypatch, + callstack_addrs=[0x1000], + resolve_strings=["main app.c:10"], + ) + resp = _handle("callstack_resolve", {}, state) + r = resp["result"] + assert r["eid"] == 0 + assert len(r["frames"]) == 1 + assert r["frames"][0]["function"] == "main" + assert r["frames"][0]["file"] == "app.c" + assert r["frames"][0]["line"] == 10 + + +def test_callstack_resolve_specific_eid( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-02: specific EID with two frames.""" + state = _make_callstack_state( + tmp_path, + monkeypatch, + callstack_addrs=[0x1000, 0x2000], + resolve_strings=["draw render.c:55", "main app.c:10"], + ) + resp = _handle("callstack_resolve", {"eid": 42}, state) + r = resp["result"] + assert r["eid"] == 42 + assert len(r["frames"]) == 2 + assert r["frames"][0]["function"] == "draw" + + +def test_callstack_resolve_eid_out_of_range( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-05: EID exceeds max_eid.""" + state = _make_callstack_state(tmp_path, monkeypatch) + resp = _handle("callstack_resolve", {"eid": 9999}, state) + assert resp["error"]["code"] == -32602 + + +def test_callstack_resolve_eid_negative( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-06: negative EID.""" + state = _make_callstack_state(tmp_path, monkeypatch) + resp = _handle("callstack_resolve", {"eid": -1}, state) + assert resp["error"]["code"] == -32602 + + +def test_callstack_resolve_empty_frames( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-08: GetResolve returns empty list.""" + state = _make_callstack_state( + tmp_path, + monkeypatch, + callstack_addrs=[], + ) + resp = _handle("callstack_resolve", {}, state) + r = resp["result"] + assert r["frames"] == [] + + +def test_callstack_resolve_multi_frame( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-09: five-frame callstack order preserved.""" + resolve = [f"func{i} file{i}.c:{i * 10}" for i in range(5)] + state = _make_callstack_state( + tmp_path, + monkeypatch, + callstack_addrs=[i for i in range(5)], + resolve_strings=resolve, + ) + resp = _handle("callstack_resolve", {}, state) + frames = resp["result"]["frames"] + assert len(frames) == 5 + for i, f in enumerate(frames): + assert f["function"] == f"func{i}" + assert f["file"] == f"file{i}.c" + assert f["line"] == i * 10 + + +def test_callstack_resolve_eid_0( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-38: EID 0 boundary.""" + state = _make_callstack_state( + tmp_path, + monkeypatch, + callstack_addrs=[0x1], + resolve_strings=["f file.c:1"], + ) + resp = _handle("callstack_resolve", {"eid": 0}, state) + assert resp["result"]["eid"] == 0 + + +def test_callstack_resolve_max_eid( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-39: max_eid boundary (valid).""" + state = _make_callstack_state( + tmp_path, + monkeypatch, + callstack_addrs=[0x1], + resolve_strings=["f file.c:1"], + ) + resp = _handle("callstack_resolve", {"eid": 100}, state) + assert "result" in resp + + +def test_callstack_resolve_max_eid_plus_one( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-40: max_eid + 1 boundary (invalid).""" + state = _make_callstack_state(tmp_path, monkeypatch) + resp = _handle("callstack_resolve", {"eid": 101}, state) + assert resp["error"]["code"] == -32602 + + +# --------------------------------------------------------------------------- +# section_write +# --------------------------------------------------------------------------- + + +def test_section_write_new( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-10: write a new custom section.""" + state = _make_state(tmp_path) + monkeypatch.setattr(state.cap, "FindSectionByName", lambda n: -1) + data_b64 = base64.b64encode(b"hello").decode() + resp = _handle("section_write", {"name": "MyNotes", "data": data_b64}, state) + r = resp["result"] + assert r["name"] == "MyNotes" + assert r["bytes"] == 5 + assert len(state.cap._written_sections) == 1 + assert state.cap._written_sections[0][1] == b"hello" + + +def test_section_write_binary( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-11: write binary content.""" + state = _make_state(tmp_path) + monkeypatch.setattr(state.cap, "FindSectionByName", lambda n: -1) + data_b64 = base64.b64encode(b"\xff\xfe").decode() + resp = _handle("section_write", {"name": "BinData", "data": data_b64}, state) + assert resp["result"]["bytes"] == 2 + assert state.cap._written_sections[0][1] == b"\xff\xfe" + + +def test_section_write_overwrite_user( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-12: overwrite non-system section.""" + state = _make_state(tmp_path) + monkeypatch.setattr(state.cap, "FindSectionByName", lambda n: 0) + monkeypatch.setattr( + state.cap, + "GetSectionProperties", + lambda idx: mock_rd.SectionProperties(name="Notes", type=mock_rd.SectionType.Notes), + ) + data_b64 = base64.b64encode(b"updated").decode() + resp = _handle("section_write", {"name": "Notes", "data": data_b64}, state) + assert resp["result"]["name"] == "Notes" + + +def test_section_write_reject_system( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-13: refuse to overwrite FrameCapture.""" + state = _make_state(tmp_path) + monkeypatch.setattr(state.cap, "FindSectionByName", lambda n: 0) + monkeypatch.setattr( + state.cap, + "GetSectionProperties", + lambda idx: mock_rd.SectionProperties( + name="FrameCapture", + type=mock_rd.SectionType.FrameCapture, + ), + ) + data_b64 = base64.b64encode(b"bad").decode() + resp = _handle("section_write", {"name": "FrameCapture", "data": data_b64}, state) + assert resp["error"]["code"] == -32602 + assert "built-in" in resp["error"]["message"].lower() + assert len(state.cap._written_sections) == 0 + + +def test_section_write_missing_name(tmp_path: Path) -> None: + """T-5C-14: missing name parameter.""" + state = _make_state(tmp_path) + data_b64 = base64.b64encode(b"x").decode() + resp = _handle("section_write", {"data": data_b64}, state) + assert resp["error"]["code"] == -32602 + + +def test_section_write_missing_data(tmp_path: Path) -> None: + """T-5C-15: missing data parameter.""" + state = _make_state(tmp_path) + resp = _handle("section_write", {"name": "Notes"}, state) + assert resp["error"]["code"] == -32602 + + +def test_section_write_invalid_base64(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """T-5C-16: invalid base64 data.""" + state = _make_state(tmp_path) + monkeypatch.setattr(state.cap, "FindSectionByName", lambda n: -1) + resp = _handle("section_write", {"name": "Notes", "data": "!!!notbase64!!!"}, state) + assert resp["error"]["code"] == -32602 + assert "base64" in resp["error"]["message"].lower() + + +def test_section_write_empty_name(tmp_path: Path) -> None: + """T-5C-17: empty section name.""" + state = _make_state(tmp_path) + data_b64 = base64.b64encode(b"x").decode() + resp = _handle("section_write", {"name": "", "data": data_b64}, state) + assert resp["error"]["code"] == -32602 + + +def test_section_write_no_cap(tmp_path: Path) -> None: + """T-5C-18: no capture open.""" + state = make_daemon_state(tmp_path=tmp_path, rd=mock_rd) + data_b64 = base64.b64encode(b"x").decode() + resp = _handle("section_write", {"name": "Notes", "data": data_b64}, state) + assert resp["error"]["code"] == -32002 + + +def test_section_write_api_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-19: WriteSection raises.""" + state = _make_state(tmp_path) + monkeypatch.setattr(state.cap, "FindSectionByName", lambda n: -1) + + def _raise(props: Any, data: bytes) -> None: + raise RuntimeError("disk full") + + monkeypatch.setattr(state.cap, "WriteSection", _raise) + data_b64 = base64.b64encode(b"x").decode() + resp = _handle("section_write", {"name": "Notes", "data": data_b64}, state) + assert resp["error"]["code"] == -32002 + assert "write failed" in resp["error"]["message"].lower() + + +def test_section_write_empty_data( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """T-5C-41: empty data (base64 of empty bytes).""" + state = _make_state(tmp_path) + monkeypatch.setattr(state.cap, "FindSectionByName", lambda n: -1) + resp = _handle("section_write", {"name": "Notes", "data": ""}, state) + assert resp["result"]["bytes"] == 0 diff --git a/tests/unit/test_debug_commands.py b/tests/unit/test_debug_commands.py index eeae5be..238ec7c 100644 --- a/tests/unit/test_debug_commands.py +++ b/tests/unit/test_debug_commands.py @@ -743,7 +743,7 @@ def _patch_helpers(monkeypatch: Any, response: dict[str, Any]) -> None: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(helpers_mod, "load_session", lambda: session) - monkeypatch.setattr(helpers_mod, "send_request", lambda _h, _p, _payload: response) + monkeypatch.setattr(helpers_mod, "send_request", lambda _h, _p, _payload, **_kw: response) def test_debug_pixel_error_plain_rc1(monkeypatch: Any) -> None: diff --git a/tests/unit/test_info_commands.py b/tests/unit/test_info_commands.py index 2817f12..d62a7c4 100644 --- a/tests/unit/test_info_commands.py +++ b/tests/unit/test_info_commands.py @@ -109,7 +109,9 @@ def test_daemon_call_error_response(monkeypatch) -> None: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) monkeypatch.setattr( - mod, "send_request", lambda _h, _p, _payload: {"error": {"message": "no replay loaded"}} + mod, + "send_request", + lambda _h, _p, _payload, **_kw: {"error": {"message": "no replay loaded"}}, ) result = CliRunner().invoke(main, ["info"]) assert result.exit_code == 1 @@ -122,7 +124,7 @@ def test_daemon_call_connection_error(monkeypatch) -> None: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) - def raise_error(*args): + def raise_error(*args, **_kw): raise ConnectionRefusedError("refused") monkeypatch.setattr(mod, "send_request", raise_error) diff --git a/tests/unit/test_pipeline_cli_phase27.py b/tests/unit/test_pipeline_cli_phase27.py index f9af5cf..bdebce9 100644 --- a/tests/unit/test_pipeline_cli_phase27.py +++ b/tests/unit/test_pipeline_cli_phase27.py @@ -16,7 +16,7 @@ def _setup(monkeypatch: pytest.MonkeyPatch, response: dict[str, Any]) -> None: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) - monkeypatch.setattr(mod, "send_request", lambda _h, _p, _payload: {"result": response}) + monkeypatch.setattr(mod, "send_request", lambda _h, _p, _payload, **_kw: {"result": response}) def _capture_calls(monkeypatch: pytest.MonkeyPatch, response: dict[str, Any]) -> list[dict]: @@ -26,7 +26,7 @@ def _capture_calls(monkeypatch: pytest.MonkeyPatch, response: dict[str, Any]) -> session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) - def capture(_h: str, _p: int, payload: dict) -> dict: + def capture(_h: str, _p: int, payload: dict, **_kw: Any) -> dict: calls.append(payload) return {"result": response} diff --git a/tests/unit/test_pipeline_commands.py b/tests/unit/test_pipeline_commands.py index d5e9097..28d152b 100644 --- a/tests/unit/test_pipeline_commands.py +++ b/tests/unit/test_pipeline_commands.py @@ -86,7 +86,7 @@ def test_bindings_with_filters(monkeypatch) -> None: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) - def capture(h, p, payload): + def capture(h, p, payload, **_kw): calls.append(payload) return {"result": {"rows": []}} @@ -202,7 +202,7 @@ def test_shader_stage_only_uses_current_eid(monkeypatch) -> None: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) - def capture(_h, _p, payload): + def capture(_h, _p, payload, **_kw): calls.append(payload) return { "result": { @@ -235,7 +235,7 @@ def test_shader_explicit_eid_stage_form_still_supported(monkeypatch) -> None: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) - def capture(_h, _p, payload): + def capture(_h, _p, payload, **_kw): calls.append(payload) return { "result": { @@ -387,7 +387,7 @@ def test_shaders_with_filters(monkeypatch) -> None: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) - def capture(h, p, payload): + def capture(h, p, payload, **_kw): calls.append(payload) return {"result": {"rows": []}} diff --git a/tests/unit/test_pipeline_shader.py b/tests/unit/test_pipeline_shader.py index 3f6eddf..6a38e52 100644 --- a/tests/unit/test_pipeline_shader.py +++ b/tests/unit/test_pipeline_shader.py @@ -146,7 +146,7 @@ def test_cli_pipeline_json_output(monkeypatch) -> None: # type: ignore[no-untyp monkeypatch.setattr( pipeline_mod, "send_request", - lambda _h, _p, _payload: {"result": {"row": {"eid": 10, "api": "Vulkan"}}}, + lambda _h, _p, _payload, **_kw: {"result": {"row": {"eid": 10, "api": "Vulkan"}}}, ) runner = CliRunner() result = runner.invoke(main, ["pipeline", "--json"]) @@ -162,7 +162,7 @@ def test_cli_shader_invalid_stage(monkeypatch) -> None: # type: ignore[no-untyp monkeypatch.setattr( pipeline_mod, "send_request", - lambda _h, _p, _payload: {"error": {"message": "invalid stage"}}, + lambda _h, _p, _payload, **_kw: {"error": {"message": "invalid stage"}}, ) runner = CliRunner() result = runner.invoke(main, ["shader", "1", "ps"]) @@ -177,7 +177,7 @@ def test_cli_pipeline_replay_unavailable(monkeypatch) -> None: # type: ignore[n monkeypatch.setattr( pipeline_mod, "send_request", - lambda _h, _p, _payload: {"error": {"message": "no replay loaded"}}, + lambda _h, _p, _payload, **_kw: {"error": {"message": "no replay loaded"}}, ) runner = CliRunner() result = runner.invoke(main, ["pipeline"]) diff --git a/tests/unit/test_resources_commands.py b/tests/unit/test_resources_commands.py index f0ccab9..5630ea7 100644 --- a/tests/unit/test_resources_commands.py +++ b/tests/unit/test_resources_commands.py @@ -65,7 +65,7 @@ def test_resource_error(monkeypatch) -> None: monkeypatch.setattr( mod, "send_request", - lambda _h, _p, _payload: {"error": {"message": "resource not found"}}, + lambda _h, _p, _payload, **_kw: {"error": {"message": "resource not found"}}, ) result = CliRunner().invoke(resource_cmd, ["999"]) assert result.exit_code == 1 diff --git a/tests/unit/test_resources_filter.py b/tests/unit/test_resources_filter.py index d89110c..3df8e61 100644 --- a/tests/unit/test_resources_filter.py +++ b/tests/unit/test_resources_filter.py @@ -43,7 +43,7 @@ def _patch_resources(monkeypatch: pytest.MonkeyPatch, response: dict[str, Any]) session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) - monkeypatch.setattr(mod, "send_request", lambda _h, _p, _payload: {"result": response}) + monkeypatch.setattr(mod, "send_request", lambda _h, _p, _payload, **_kw: {"result": response}) # --------------------------------------------------------------------------- @@ -263,7 +263,7 @@ def test_type_option_forwarded(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(mod, "load_session", lambda: session) captured: list[dict[str, Any]] = [] - def fake_send(_h: str, _p: int, payload: dict[str, Any]) -> dict[str, Any]: + def fake_send(_h: str, _p: int, payload: dict[str, Any], **_kw: Any) -> dict[str, Any]: captured.append(payload["params"]) return {"result": {"rows": []}} @@ -278,7 +278,7 @@ def test_name_option_forwarded(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(mod, "load_session", lambda: session) captured: list[dict[str, Any]] = [] - def fake_send(_h: str, _p: int, payload: dict[str, Any]) -> dict[str, Any]: + def fake_send(_h: str, _p: int, payload: dict[str, Any], **_kw: Any) -> dict[str, Any]: captured.append(payload["params"]) return {"result": {"rows": []}} @@ -293,7 +293,7 @@ def test_sort_option_forwarded(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(mod, "load_session", lambda: session) captured: list[dict[str, Any]] = [] - def fake_send(_h: str, _p: int, payload: dict[str, Any]) -> dict[str, Any]: + def fake_send(_h: str, _p: int, payload: dict[str, Any], **_kw: Any) -> dict[str, Any]: captured.append(payload["params"]) return {"result": {"rows": []}} @@ -318,7 +318,7 @@ def test_all_three_options_forwarded(self, monkeypatch: pytest.MonkeyPatch) -> N monkeypatch.setattr(mod, "load_session", lambda: session) captured: list[dict[str, Any]] = [] - def fake_send(_h: str, _p: int, payload: dict[str, Any]) -> dict[str, Any]: + def fake_send(_h: str, _p: int, payload: dict[str, Any], **_kw: Any) -> dict[str, Any]: captured.append(payload["params"]) return {"result": {"rows": []}} diff --git a/tests/unit/test_script_command.py b/tests/unit/test_script_command.py index a5c99c0..8dc7914 100644 --- a/tests/unit/test_script_command.py +++ b/tests/unit/test_script_command.py @@ -18,7 +18,7 @@ def _patch_daemon(monkeypatch: pytest.MonkeyPatch, response: dict[str, Any]) -> session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(helpers_mod, "load_session", lambda: session) - monkeypatch.setattr(helpers_mod, "send_request", lambda _h, _p, _payload: response) + monkeypatch.setattr(helpers_mod, "send_request", lambda _h, _p, _payload, **_kw: response) def _success_response( @@ -115,7 +115,7 @@ def test_single_arg(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> No session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(helpers_mod, "load_session", lambda: session) - def _capture(_h: str, _p: int, payload: dict[str, Any]) -> dict[str, Any]: + def _capture(_h: str, _p: int, payload: dict[str, Any], **_kw: Any) -> dict[str, Any]: captured.append(payload) return _success_response() @@ -133,7 +133,7 @@ def test_multiple_args(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(helpers_mod, "load_session", lambda: session) - def _capture(_h: str, _p: int, payload: dict[str, Any]) -> dict[str, Any]: + def _capture(_h: str, _p: int, payload: dict[str, Any], **_kw: Any) -> dict[str, Any]: captured.append(payload) return _success_response() @@ -158,7 +158,7 @@ def test_no_args_default_empty(self, monkeypatch: pytest.MonkeyPatch, tmp_path: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(helpers_mod, "load_session", lambda: session) - def _capture(_h: str, _p: int, payload: dict[str, Any]) -> dict[str, Any]: + def _capture(_h: str, _p: int, payload: dict[str, Any], **_kw: Any) -> dict[str, Any]: captured.append(payload) return _success_response() @@ -176,7 +176,7 @@ def test_arg_value_with_equals(self, monkeypatch: pytest.MonkeyPatch, tmp_path: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(helpers_mod, "load_session", lambda: session) - def _capture(_h: str, _p: int, payload: dict[str, Any]) -> dict[str, Any]: + def _capture(_h: str, _p: int, payload: dict[str, Any], **_kw: Any) -> dict[str, Any]: captured.append(payload) return _success_response() diff --git a/tests/unit/test_search.py b/tests/unit/test_search.py index a2eb885..08dd255 100644 --- a/tests/unit/test_search.py +++ b/tests/unit/test_search.py @@ -433,7 +433,7 @@ def _patch_search(monkeypatch: pytest.MonkeyPatch, response: dict[str, Any]) -> session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) - monkeypatch.setattr(mod, "send_request", lambda _h, _p, _payload: {"result": response}) + monkeypatch.setattr(mod, "send_request", lambda _h, _p, _payload, **_kw: {"result": response}) class TestSearchCli: @@ -544,7 +544,12 @@ def test_stage_option_passed(self, monkeypatch: pytest.MonkeyPatch) -> None: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) - def _capture_request(_h: str, _p: int, payload: dict[str, Any]) -> dict[str, Any]: + def _capture_request( + _h: str, + _p: int, + payload: dict[str, Any], + **_kw: Any, + ) -> dict[str, Any]: captured.append(payload) return {"result": {"matches": [], "truncated": False}} @@ -561,7 +566,12 @@ def test_case_sensitive_flag(self, monkeypatch: pytest.MonkeyPatch) -> None: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) - def _capture_request(_h: str, _p: int, payload: dict[str, Any]) -> dict[str, Any]: + def _capture_request( + _h: str, + _p: int, + payload: dict[str, Any], + **_kw: Any, + ) -> dict[str, Any]: captured.append(payload) return {"result": {"matches": [], "truncated": False}} diff --git a/tests/unit/test_shader_preload.py b/tests/unit/test_shader_preload.py index 6f1de87..e83f9cb 100644 --- a/tests/unit/test_shader_preload.py +++ b/tests/unit/test_shader_preload.py @@ -270,7 +270,7 @@ def test_preload_calls_rpc(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(helpers_mod, "load_session", lambda: session) - def _capture_send(_h: str, _p: int, payload: dict[str, Any]) -> dict[str, Any]: + def _capture_send(_h: str, _p: int, payload: dict[str, Any], **_kw: Any) -> dict[str, Any]: captured.append(payload) return {"result": {"done": True, "shaders": 5}} @@ -295,7 +295,7 @@ def test_no_preload_does_not_call_rpc( captured: list[dict[str, Any]] = [] import rdc.commands._helpers as helpers_mod - monkeypatch.setattr(helpers_mod, "send_request", lambda *a: captured.append(a)) + monkeypatch.setattr(helpers_mod, "send_request", lambda *a, **_kw: captured.append(a)) capture_file = tmp_path / "test.rdc" capture_file.touch() diff --git a/tests/unit/test_unix_helpers_commands.py b/tests/unit/test_unix_helpers_commands.py index b406411..39776e8 100644 --- a/tests/unit/test_unix_helpers_commands.py +++ b/tests/unit/test_unix_helpers_commands.py @@ -50,7 +50,7 @@ def test_count_error_response(monkeypatch) -> None: session = type("S", (), {"host": "127.0.0.1", "port": 1, "token": "tok"})() monkeypatch.setattr(mod, "load_session", lambda: session) monkeypatch.setattr( - mod, "send_request", lambda _h, _p, _payload: {"error": {"message": "no replay"}} + mod, "send_request", lambda _h, _p, _payload, **_kw: {"error": {"message": "no replay"}} ) result = CliRunner().invoke(main, ["count", "draws"]) assert result.exit_code == 1