mirror of
https://github.com/microsoft/debugpy.git
synced 2025-12-23 08:48:12 +00:00
575 lines
17 KiB
Python
575 lines
17 KiB
Python
# Copyright (c) Microsoft Corporation. All rights reserved.
|
|
# Licensed under the MIT License. See LICENSE in the project root
|
|
# for license information.
|
|
|
|
from __future__ import absolute_import, division, print_function, unicode_literals
|
|
|
|
|
|
import os
|
|
import ptvsd
|
|
import psutil
|
|
import py.path
|
|
import pytest
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
from ptvsd.common import compat, fmt, json, log
|
|
from ptvsd.common.compat import unicode
|
|
from tests import net, timeline, watchdog
|
|
from tests.patterns import some
|
|
|
|
|
|
PTVSD_DIR = py.path.local(ptvsd.__file__) / ".."
|
|
PTVSD_PORT = net.get_test_server_port(5678, 5800)
|
|
|
|
# Code that is injected into the debuggee process when it does `import debug_me`,
|
|
# and start_method is attach_socket_*
|
|
PTVSD_DEBUG_ME = """
|
|
import ptvsd
|
|
ptvsd.enable_attach(("127.0.0.1", {ptvsd_port}), log_dir={log_dir!r})
|
|
ptvsd.wait_for_attach()
|
|
"""
|
|
|
|
|
|
class DebugStartBase(object):
|
|
ignore_unobserved = []
|
|
|
|
def __init__(self, session, method="base"):
|
|
self.session = session
|
|
self.method = method
|
|
self.debuggee_process = None
|
|
self.expected_exit_code = None
|
|
|
|
def start_debugging(self, **kwargs):
|
|
pass
|
|
|
|
def wait_for_debuggee(self):
|
|
# TODO: Exit should not be restricted to launch tests only
|
|
if "launch" in self.method:
|
|
exited = self.session.timeline.wait_until_realized(
|
|
timeline.Event("exited")
|
|
).body
|
|
assert exited == some.dict.containing(
|
|
{
|
|
"exitCode": some.int
|
|
if self.expected_exit_code is None
|
|
else self.expected_exit_code
|
|
}
|
|
)
|
|
|
|
self.session.timeline.wait_until_realized(timeline.Event("terminated"))
|
|
|
|
if self.debuggee_process is None:
|
|
return
|
|
|
|
try:
|
|
self.debuggee_process.wait()
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
watchdog.unregister_spawn(
|
|
self.debuggee_process.pid, self.session.debuggee_id
|
|
)
|
|
|
|
def run_in_terminal(self, request, **kwargs):
|
|
raise request.isnt_valid("not supported")
|
|
|
|
def _build_common_args(
|
|
self,
|
|
args,
|
|
showReturnValue=None,
|
|
justMyCode=True,
|
|
subProcess=None,
|
|
django=None,
|
|
jinja=None,
|
|
flask=None,
|
|
pyramid=None,
|
|
logToFile=None,
|
|
redirectOutput=True,
|
|
noDebug=None,
|
|
maxExceptionStackFrames=None,
|
|
steppingResumesAllThreads=None,
|
|
rules=None,
|
|
successExitCodes=None,
|
|
breakOnSystemExitZero=None,
|
|
pathMappings=None,
|
|
):
|
|
if logToFile:
|
|
args["logToFile"] = logToFile
|
|
if "env" in args:
|
|
args["env"]["PTVSD_LOG_DIR"] = self.session.log_dir
|
|
|
|
if showReturnValue:
|
|
args["showReturnValue"] = showReturnValue
|
|
args["debugOptions"] += ["ShowReturnValue"]
|
|
|
|
if redirectOutput:
|
|
args["redirectOutput"] = redirectOutput
|
|
args["debugOptions"] += ["RedirectOutput"]
|
|
|
|
if justMyCode is False:
|
|
# default behavior is Just-my-code = true
|
|
args["justMyCode"] = justMyCode
|
|
args["debugOptions"] += ["DebugStdLib"]
|
|
|
|
if django:
|
|
args["django"] = django
|
|
args["debugOptions"] += ["Django"]
|
|
|
|
if jinja:
|
|
args["jinja"] = jinja
|
|
args["debugOptions"] += ["Jinja"]
|
|
|
|
if flask:
|
|
args["flask"] = flask
|
|
args["debugOptions"] += ["Flask"]
|
|
|
|
if pyramid:
|
|
args["pyramid"] = pyramid
|
|
args["debugOptions"] += ["Pyramid"]
|
|
|
|
# VS Code uses noDebug in both attach and launch cases. Even though
|
|
# noDebug on attach does not make any sense.
|
|
if noDebug:
|
|
args["noDebug"] = True
|
|
|
|
if subProcess:
|
|
args["subProcess"] = subProcess
|
|
args["debugOptions"] += ["Multiprocess"]
|
|
|
|
if maxExceptionStackFrames:
|
|
args["maxExceptionStackFrames"] = maxExceptionStackFrames
|
|
|
|
if steppingResumesAllThreads is not None:
|
|
args["steppingResumesAllThreads"] = steppingResumesAllThreads
|
|
|
|
if rules is not None:
|
|
args["rules"] = rules
|
|
|
|
if successExitCodes:
|
|
args["successExitCodes"] = successExitCodes
|
|
|
|
if breakOnSystemExitZero:
|
|
args["debugOptions"] += ["BreakOnSystemExitZero"]
|
|
|
|
if pathMappings is not None:
|
|
args["pathMappings"] = pathMappings
|
|
|
|
def __str__(self):
|
|
return self.method
|
|
|
|
|
|
class Launch(DebugStartBase):
|
|
def __init__(self, session):
|
|
super(Launch, self).__init__(session, "launch")
|
|
self._launch_args = None
|
|
|
|
def _build_launch_args(
|
|
self,
|
|
launch_args,
|
|
run_as,
|
|
target,
|
|
pythonPath=sys.executable,
|
|
args=(),
|
|
cwd=None,
|
|
env=None,
|
|
stopOnEntry=None,
|
|
gevent=None,
|
|
sudo=None,
|
|
waitOnNormalExit=None,
|
|
waitOnAbnormalExit=None,
|
|
console="externalTerminal",
|
|
internalConsoleOptions="neverOpen",
|
|
**kwargs
|
|
):
|
|
assert console in ("internalConsole", "integratedTerminal", "externalTerminal")
|
|
env = {} if env is None else dict(env)
|
|
debug_options = []
|
|
launch_args.update(
|
|
{
|
|
"name": "Terminal",
|
|
"type": "python",
|
|
"request": "launch",
|
|
"console": console,
|
|
"env": env,
|
|
"pythonPath": pythonPath,
|
|
"args": args,
|
|
"internalConsoleOptions": internalConsoleOptions,
|
|
"debugOptions": debug_options,
|
|
}
|
|
)
|
|
|
|
if stopOnEntry:
|
|
launch_args["stopOnEntry"] = stopOnEntry
|
|
debug_options += ["StopOnEntry"]
|
|
|
|
if gevent:
|
|
launch_args["gevent"] = gevent
|
|
env["GEVENT_SUPPORT"] = "True"
|
|
|
|
if sudo:
|
|
launch_args["sudo"] = sudo
|
|
|
|
if waitOnNormalExit:
|
|
debug_options += ["WaitOnNormalExit"]
|
|
|
|
if waitOnAbnormalExit:
|
|
debug_options += ["WaitOnAbnormalExit"]
|
|
|
|
target_str = target
|
|
if isinstance(target, py.path.local):
|
|
target_str = target.strpath
|
|
|
|
if cwd:
|
|
launch_args["cwd"] = cwd
|
|
elif os.path.isfile(target_str) or os.path.isdir(target_str):
|
|
launch_args["cwd"] = os.path.dirname(target_str)
|
|
else:
|
|
launch_args["cwd"] = os.getcwd()
|
|
|
|
if "PYTHONPATH" not in env:
|
|
env["PYTHONPATH"] = ""
|
|
|
|
if run_as == "program":
|
|
launch_args["program"] = target_str
|
|
elif run_as == "module":
|
|
if os.path.isfile(target_str) or os.path.isdir(target_str):
|
|
env["PYTHONPATH"] += os.pathsep + os.path.dirname(target_str)
|
|
try:
|
|
launch_args["module"] = target_str[
|
|
(len(os.path.dirname(target_str)) + 1) : -3
|
|
]
|
|
except Exception:
|
|
launch_args["module"] = "code_to_debug"
|
|
else:
|
|
launch_args["module"] = target_str
|
|
elif run_as == "code":
|
|
with open(target_str, "rb") as f:
|
|
launch_args["code"] = f.read().decode("utf-8")
|
|
else:
|
|
pytest.fail()
|
|
|
|
self._build_common_args(launch_args, **kwargs)
|
|
return launch_args
|
|
|
|
def configure(self, run_as, target, **kwargs):
|
|
self._launch_args = self._build_launch_args({}, run_as, target, **kwargs)
|
|
self.no_debug = self._launch_args.get("noDebug", False)
|
|
|
|
if not self.no_debug:
|
|
self._launch_request = self.session.send_request(
|
|
"launch", self._launch_args
|
|
)
|
|
self.session.wait_for_next_event("initialized")
|
|
|
|
def start_debugging(self):
|
|
if self.no_debug:
|
|
self._launch_request = self.session.send_request(
|
|
"launch", self._launch_args
|
|
)
|
|
else:
|
|
self.session.request("configurationDone")
|
|
|
|
self._launch_request.wait_for_response(freeze=False)
|
|
return self._launch_request
|
|
|
|
def run_in_terminal(self, request):
|
|
args = request("args", json.array(unicode))
|
|
cwd = request("cwd", ".")
|
|
|
|
env = os.environ.copy()
|
|
env.update(request("env", json.object(unicode)))
|
|
|
|
if sys.version_info < (3,):
|
|
args = [compat.filename_str(s) for s in args]
|
|
env = {
|
|
compat.filename_str(k): compat.filename_str(v) for k, v in env.items()
|
|
}
|
|
|
|
log.info(
|
|
'{0} spawning {1} via "runInTerminal" request',
|
|
self.session,
|
|
self.session.debuggee_id,
|
|
)
|
|
self.debuggee_process = psutil.Popen(
|
|
args,
|
|
cwd=cwd,
|
|
env=env,
|
|
bufsize=0,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
watchdog.register_spawn(self.debuggee_process.pid, self.session.debuggee_id)
|
|
self.session.captured_output.capture(self.debuggee_process)
|
|
return {}
|
|
|
|
|
|
class AttachBase(DebugStartBase):
|
|
ignore_unobserved = DebugStartBase.ignore_unobserved + []
|
|
|
|
def __init__(self, session, method):
|
|
super(AttachBase, self).__init__(session, method)
|
|
self._attach_args = {}
|
|
|
|
def _build_attach_args(
|
|
self,
|
|
attach_args,
|
|
run_as,
|
|
target,
|
|
host="127.0.0.1",
|
|
port=PTVSD_PORT,
|
|
**kwargs
|
|
):
|
|
assert host is not None
|
|
assert port is not None
|
|
debug_options = []
|
|
attach_args.update(
|
|
{
|
|
"name": "Attach",
|
|
"type": "python",
|
|
"request": "attach",
|
|
"debugOptions": debug_options,
|
|
}
|
|
)
|
|
|
|
attach_args["host"] = host
|
|
attach_args["port"] = port
|
|
|
|
self._build_common_args(attach_args, **kwargs)
|
|
return attach_args
|
|
|
|
def configure(self, run_as, target, **kwargs):
|
|
target_str = target
|
|
if isinstance(target, py.path.local):
|
|
target_str = target.strpath
|
|
|
|
env = os.environ.copy()
|
|
env.update(kwargs["env"])
|
|
|
|
cli_args = kwargs.get("cli_args")
|
|
if run_as == "program":
|
|
cli_args += [target_str]
|
|
elif run_as == "module":
|
|
if os.path.isfile(target_str) or os.path.isdir(target_str):
|
|
env["PYTHONPATH"] += os.pathsep + os.path.dirname(target_str)
|
|
try:
|
|
module = target_str[(len(os.path.dirname(target_str)) + 1) : -3]
|
|
except Exception:
|
|
module = "code_to_debug"
|
|
else:
|
|
module = target_str
|
|
cli_args += ["-m", module]
|
|
elif run_as == "code":
|
|
with open(target_str, "rb") as f:
|
|
cli_args += ["-c", f.read()]
|
|
else:
|
|
pytest.fail()
|
|
|
|
cli_args += kwargs.get("args")
|
|
cli_args = [compat.filename_str(s) for s in cli_args]
|
|
env = {compat.filename_str(k): compat.filename_str(v) for k, v in env.items()}
|
|
|
|
cwd = kwargs.get("cwd")
|
|
if cwd:
|
|
pass
|
|
elif os.path.isfile(target_str) or os.path.isdir(target_str):
|
|
cwd = os.path.dirname(target_str)
|
|
else:
|
|
cwd = os.getcwd()
|
|
|
|
if "pathMappings" not in self._attach_args:
|
|
self._attach_args["pathMappings"] = [{"localRoot": cwd, "remoteRoot": "."}]
|
|
|
|
env_str = "\n".join((fmt(" {0}={1}", k, env[k]) for k in sorted(env.keys())))
|
|
log.info(
|
|
"Spawning {0}: {1!j}\n\n" "with cwd:\n {2!j}\n\n" "with env:\n{3}",
|
|
self.session.debuggee_id,
|
|
cli_args,
|
|
cwd,
|
|
env_str,
|
|
)
|
|
|
|
self.debuggee_process = psutil.Popen(
|
|
cli_args,
|
|
cwd=cwd,
|
|
env=env,
|
|
bufsize=0,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
watchdog.register_spawn(self.debuggee_process.pid, self.session.debuggee_id)
|
|
self.session.captured_output.capture(self.debuggee_process)
|
|
pid = self.debuggee_process.pid
|
|
|
|
if self.method == "attach_pid":
|
|
self._attach_args["processId"] = pid
|
|
else:
|
|
log.info(
|
|
"Waiting for {0} to open listener socket...", self.session.debuggee_id
|
|
)
|
|
for i in range(0, 100):
|
|
connections = psutil.net_connections()
|
|
if any(p == pid for (_, _, _, _, _, _, p) in connections):
|
|
break
|
|
time.sleep(0.1)
|
|
else:
|
|
log.warning("Couldn't detect open listener socket; proceeding anyway.")
|
|
|
|
self._attach_request = self.session.send_request("attach", self._attach_args)
|
|
self.session.wait_for_next_event("initialized")
|
|
|
|
def start_debugging(self):
|
|
self.session.request("configurationDone")
|
|
|
|
self.no_debug = self._attach_args.get("noDebug", False)
|
|
if self.no_debug:
|
|
log.info('{0} ignoring "noDebug" in "attach"', self.session)
|
|
|
|
self._attach_request.wait_for_response()
|
|
return self._attach_request
|
|
|
|
|
|
class AttachSocketImport(AttachBase):
|
|
def __init__(self, session):
|
|
super(AttachSocketImport, self).__init__(session, "attach_socket_import")
|
|
|
|
def _check_ready_for_import(self, path_or_code):
|
|
if isinstance(path_or_code, py.path.local):
|
|
path_or_code = path_or_code.strpath
|
|
|
|
if os.path.isfile(path_or_code):
|
|
with open(path_or_code, "rb") as f:
|
|
code = f.read()
|
|
elif "\n" in path_or_code:
|
|
code = path_or_code
|
|
else:
|
|
# path_or_code is a module name
|
|
return
|
|
assert b"debug_me" in code, fmt(
|
|
"{0} is started via {1}, but it doesn't import debug_me.",
|
|
path_or_code,
|
|
self.method,
|
|
)
|
|
|
|
def configure(
|
|
self,
|
|
run_as,
|
|
target,
|
|
pythonPath=sys.executable,
|
|
args=(),
|
|
cwd=None,
|
|
env=None,
|
|
**kwargs
|
|
):
|
|
env = {} if env is None else dict(env)
|
|
self._attach_args = self._build_attach_args({}, run_as, target, **kwargs)
|
|
|
|
ptvsd_port = self._attach_args["port"]
|
|
log_dir = (
|
|
self.session.log_dir
|
|
if not self._attach_args.get("logToFile", False)
|
|
else None
|
|
)
|
|
env["PTVSD_DEBUG_ME"] = fmt(
|
|
PTVSD_DEBUG_ME, ptvsd_port=ptvsd_port, log_dir=log_dir
|
|
)
|
|
|
|
self._check_ready_for_import(target)
|
|
|
|
cli_args = [pythonPath]
|
|
super(AttachSocketImport, self).configure(
|
|
run_as, target, cwd=cwd, env=env, args=args, cli_args=cli_args, **kwargs
|
|
)
|
|
|
|
|
|
class AttachSocketCmdLine(AttachBase):
|
|
def __init__(self, session):
|
|
super(AttachSocketCmdLine, self).__init__(session, "attach_socket_cmdline")
|
|
|
|
def configure(
|
|
self,
|
|
run_as,
|
|
target,
|
|
pythonPath=sys.executable,
|
|
args=(),
|
|
cwd=None,
|
|
env=None,
|
|
**kwargs
|
|
):
|
|
env = {} if env is None else dict(env)
|
|
self._attach_args = self._build_attach_args({}, run_as, target, **kwargs)
|
|
|
|
cli_args = [pythonPath]
|
|
cli_args += [PTVSD_DIR.strpath]
|
|
cli_args += ["--wait"]
|
|
cli_args += [
|
|
"--host",
|
|
self._attach_args["host"],
|
|
"--port",
|
|
str(self._attach_args["port"]),
|
|
]
|
|
|
|
log_dir = (
|
|
self.session.log_dir if self._attach_args.get("logToFile", False) else None
|
|
)
|
|
if log_dir:
|
|
cli_args += ["--log-dir", log_dir]
|
|
|
|
if self._attach_args.get("subProcess", False):
|
|
cli_args += ["--multiprocess"]
|
|
|
|
super(AttachSocketCmdLine, self).configure(
|
|
run_as, target, cwd=cwd, env=env, args=args, cli_args=cli_args, **kwargs
|
|
)
|
|
|
|
|
|
class AttachProcessId(AttachBase):
|
|
def __init__(self, session):
|
|
super(AttachProcessId, self).__init__(session, "attach_pid")
|
|
|
|
def configure(
|
|
self,
|
|
run_as,
|
|
target,
|
|
pythonPath=sys.executable,
|
|
args=(),
|
|
cwd=None,
|
|
env=None,
|
|
**kwargs
|
|
):
|
|
env = {} if env is None else dict(env)
|
|
self._attach_args = self._build_attach_args({}, run_as, target, **kwargs)
|
|
|
|
log_dir = (
|
|
self.session.log_dir if self._attach_args.get("logToFile", False) else None
|
|
)
|
|
if log_dir:
|
|
self._attach_args["ptvsdArgs"] = ["--log-dir", log_dir]
|
|
|
|
cli_args = [pythonPath]
|
|
super(AttachProcessId, self).configure(
|
|
run_as, target, cwd=cwd, env=env, args=args, cli_args=cli_args, **kwargs
|
|
)
|
|
|
|
|
|
class CustomServer(DebugStartBase):
|
|
def __init__(self, session):
|
|
super().__init__(session, "custom_server")
|
|
|
|
|
|
class CustomClient(DebugStartBase):
|
|
def __init__(self, session):
|
|
super().__init__(session, "custom_client")
|
|
|
|
|
|
__all__ = [
|
|
Launch, # ptvsd --client ... foo.py
|
|
AttachSocketCmdLine, # ptvsd ... foo.py
|
|
AttachSocketImport, # python foo.py (foo.py must import debug_me)
|
|
AttachProcessId, # python foo.py && ptvsd ... --pid
|
|
CustomClient, # python foo.py (foo.py has to manually call ptvsd.attach)
|
|
CustomServer, # python foo.py (foo.py has to manually call ptvsd.enable_attach)
|
|
]
|