Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions ipykernel/debugger.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,8 @@
else:
raise e

if t.TYPE_CHECKING:
from IPython.core.interactiveshell import InteractiveShell

# Required for backwards compatibility
ROUTING_ID = getattr(zmq, "ROUTING_ID", None) or zmq.IDENTITY
Expand Down Expand Up @@ -88,7 +90,7 @@ def __init__(self):

def track(self):
"""Start tracking."""
var = get_ipython().user_ns
var = t.cast("InteractiveShell", get_ipython()).user_ns
self.frame = _FakeFrame(_FakeCode("<module>", get_file_name("sys._getframe()")), var, var)
self.tracker.track("thread1", pydevd_frame_utils.create_frames_list_from_frame(self.frame))

Expand Down Expand Up @@ -443,7 +445,8 @@ def start(self):
self.debugpy_initialized = msg["content"]["status"] == "ok"

# Don't remove leading empty lines when debugging so the breakpoints are correctly positioned
cleanup_transforms = get_ipython().input_transformer_manager.cleanup_transforms
shell = t.cast("InteractiveShell", get_ipython())
cleanup_transforms = shell.input_transformer_manager.cleanup_transforms
if leading_empty_lines in cleanup_transforms:
index = cleanup_transforms.index(leading_empty_lines)
self._removed_cleanup[index] = cleanup_transforms.pop(index)
Expand All @@ -456,7 +459,8 @@ def stop(self):
self.debugpy_client.disconnect_tcp_socket()

# Restore remove cleanup transformers
cleanup_transforms = get_ipython().input_transformer_manager.cleanup_transforms
shell = t.cast("InteractiveShell", get_ipython())
cleanup_transforms = shell.input_transformer_manager.cleanup_transforms
for index in sorted(self._removed_cleanup):
func = self._removed_cleanup.pop(index)
cleanup_transforms.insert(index, func)
Expand Down Expand Up @@ -641,7 +645,8 @@ async def richInspectVariables(self, message):
if not self.stopped_threads:
# The code did not hit a breakpoint, we use the interpreter
# to get the rich representation of the variable
result = get_ipython().user_expressions({var_name: var_name})[var_name]
shell = t.cast("InteractiveShell", get_ipython())
result = shell.user_expressions({var_name: var_name})[var_name]
if result.get("status", "error") == "ok":
repr_data = result.get("data", {})
repr_metadata = result.get("metadata", {})
Expand Down
34 changes: 23 additions & 11 deletions ipykernel/kernelbase.py
Original file line number Diff line number Diff line change
Expand Up @@ -1382,24 +1382,36 @@ def _no_raw_input(self):
msg = "raw_input was called, but this frontend does not support stdin."
raise StdinNotImplementedError(msg)

def getpass(self, prompt="", stream=None):
def getpass(
self,
prompt: str = "",
stream: t.TextIO | None = None,
*,
echo_char: str | None = None,
) -> str:
"""Forward getpass to frontends

The signature mirrors :func:`getpass.getpass`, which this replaces on
the kernel side; the parameters that only make sense for a local
terminal are accepted but ignored.

Raises
------
StdinNotImplementedError if active frontend doesn't support stdin.
"""
if not self._allow_stdin:
msg = "getpass was called, but this frontend does not support input requests."
raise StdinNotImplementedError(msg)
if stream is not None:
import warnings

warnings.warn(
"The `stream` parameter of `getpass.getpass` will have no effect when using ipykernel",
UserWarning,
stacklevel=2,
)
for name, value in (("stream", stream), ("echo_char", echo_char)):
if value is not None:
import warnings

warnings.warn(
f"The `{name}` parameter of `getpass.getpass` will have no effect"
" when using ipykernel",
UserWarning,
stacklevel=2,
)
return self._input_request(
prompt,
self._get_shell_context_var(self._shell_parent_ident),
Expand All @@ -1424,7 +1436,7 @@ def raw_input(self, prompt=""):
password=False,
)

def _input_request(self, prompt, ident, parent, password=False):
def _input_request(self, prompt, ident, parent, password=False) -> str:
# Flush output before making the request.
if sys.stdout is not None:
sys.stdout.flush()
Expand Down Expand Up @@ -1467,7 +1479,7 @@ def _input_request(self, prompt, ident, parent, password=False):
self.log.warning("Invalid Message:", exc_info=True)

try:
value = reply["content"]["value"] # type:ignore[index]
value: str = reply["content"]["value"] # type:ignore[index]
except Exception:
self.log.error("Bad input_reply: %s", parent)
value = ""
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,7 @@ filterwarnings= [

# Ignore our own warnings
"ignore:The `stream` parameter of `getpass.getpass` will have no effect:UserWarning",
"ignore:The `echo_char` parameter of `getpass.getpass` will have no effect:UserWarning",

# IPython warnings
"ignore: `Completer.complete` is pending deprecation since IPython 6.0 and will be replaced by `Completer.completions`:PendingDeprecationWarning",
Expand Down
9 changes: 9 additions & 0 deletions tests/inprocess/test_kernel.py
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,15 @@ def test_getpass_stream(kc):
kernel.getpass(stream="non empty")


def test_getpass_echo_char(kc):
"""Tests that kernel getpass accepts the echo_char parameter"""
kernel = InProcessKernel()
kernel._allow_stdin = True
kernel._input_request = lambda *args, **kwargs: None # type:ignore

kernel.getpass(echo_char="*")


async def test_do_execute(kc):
kernel = InProcessKernel()
await kernel.do_execute("a=1", True)
Expand Down