mirror of
https://github.com/microsoft/debugpy.git
synced 2025-12-23 08:48:12 +00:00
347 lines
9.5 KiB
Python
347 lines
9.5 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 pytest
|
|
import sys
|
|
|
|
import debugpy
|
|
from tests import debug
|
|
from tests.debug import runners
|
|
from tests.patterns import some
|
|
|
|
|
|
@pytest.fixture(params=[runners.launch, runners.attach_by_socket["api"]])
|
|
def run(request):
|
|
return request.param
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"start_method",
|
|
[""]
|
|
if sys.version_info < (3,)
|
|
else ["spawn"]
|
|
if sys.platform == "win32"
|
|
else ["spawn", "fork"],
|
|
)
|
|
def test_multiprocessing(pyfile, target, run, start_method):
|
|
if start_method == "spawn" and sys.platform != "win32":
|
|
pytest.skip("https://github.com/microsoft/ptvsd/issues/1887")
|
|
|
|
@pyfile
|
|
def code_to_debug():
|
|
import debuggee
|
|
import multiprocessing
|
|
import os
|
|
import sys
|
|
|
|
def parent(q, a):
|
|
from debuggee import backchannel
|
|
|
|
debuggee.setup()
|
|
|
|
print("spawning child")
|
|
p = multiprocessing.Process(target=child, args=(q, a))
|
|
p.start()
|
|
print("child spawned")
|
|
|
|
q.put("child_pid?")
|
|
what, child_pid = a.get()
|
|
assert what == "child_pid"
|
|
backchannel.send(child_pid)
|
|
|
|
q.put("grandchild_pid?")
|
|
what, grandchild_pid = a.get()
|
|
assert what == "grandchild_pid"
|
|
backchannel.send(grandchild_pid)
|
|
|
|
assert backchannel.receive() == "continue"
|
|
q.put("exit!")
|
|
p.join()
|
|
|
|
def child(q, a):
|
|
print("entering child")
|
|
assert q.get() == "child_pid?"
|
|
a.put(("child_pid", os.getpid()))
|
|
|
|
print("spawning child of child")
|
|
p = multiprocessing.Process(target=grandchild, args=(q, a))
|
|
p.start()
|
|
p.join()
|
|
|
|
print("leaving child")
|
|
|
|
def grandchild(q, a):
|
|
print("entering grandchild")
|
|
assert q.get() == "grandchild_pid?"
|
|
a.put(("grandchild_pid", os.getpid()))
|
|
|
|
assert q.get() == "exit!"
|
|
print("leaving grandchild")
|
|
|
|
if __name__ == "__main__":
|
|
start_method = sys.argv[1]
|
|
if start_method != "":
|
|
multiprocessing.set_start_method(start_method)
|
|
|
|
q = multiprocessing.Queue()
|
|
a = multiprocessing.Queue()
|
|
try:
|
|
parent(q, a)
|
|
finally:
|
|
q.close()
|
|
a.close()
|
|
|
|
with debug.Session() as parent_session:
|
|
parent_backchannel = parent_session.open_backchannel()
|
|
|
|
with run(parent_session, target(code_to_debug, args=[start_method])):
|
|
pass
|
|
|
|
expected_child_config = dict(parent_session.config)
|
|
expected_child_config.update(
|
|
{
|
|
"name": some.str,
|
|
"request": "attach",
|
|
"subProcessId": some.int,
|
|
"host": some.str,
|
|
"port": some.int,
|
|
}
|
|
)
|
|
|
|
child_config = parent_session.wait_for_next_event("debugpyAttach")
|
|
assert child_config == expected_child_config
|
|
parent_session.proceed()
|
|
|
|
with debug.Session(child_config) as child_session:
|
|
with child_session.start():
|
|
pass
|
|
|
|
expected_grandchild_config = dict(child_session.config)
|
|
expected_grandchild_config.update(
|
|
{
|
|
"name": some.str,
|
|
"request": "attach",
|
|
"subProcessId": some.int,
|
|
"host": some.str,
|
|
"port": some.int,
|
|
}
|
|
)
|
|
|
|
grandchild_config = child_session.wait_for_next_event("debugpyAttach")
|
|
assert grandchild_config == expected_grandchild_config
|
|
|
|
with debug.Session(grandchild_config) as grandchild_session:
|
|
with grandchild_session.start():
|
|
pass
|
|
|
|
parent_backchannel.send("continue")
|
|
|
|
|
|
def test_subprocess(pyfile, target, run):
|
|
@pyfile
|
|
def child():
|
|
import os
|
|
import sys
|
|
|
|
assert "debugpy" in sys.modules
|
|
|
|
import debugpy
|
|
from debuggee import backchannel
|
|
|
|
backchannel.send(os.getpid())
|
|
backchannel.send(debugpy.__file__)
|
|
backchannel.send(sys.argv)
|
|
|
|
@pyfile
|
|
def parent():
|
|
import debuggee
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
debuggee.setup()
|
|
argv = [sys.executable, sys.argv[1], "--arg1", "--arg2", "--arg3"]
|
|
env = os.environ.copy()
|
|
process = subprocess.Popen(
|
|
argv,
|
|
env=env,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
)
|
|
process.wait()
|
|
|
|
with debug.Session() as parent_session:
|
|
backchannel = parent_session.open_backchannel()
|
|
|
|
with run(parent_session, target(parent, args=[child])):
|
|
pass
|
|
|
|
expected_child_config = dict(parent_session.config)
|
|
expected_child_config.update(
|
|
{
|
|
"name": some.str,
|
|
"request": "attach",
|
|
"subProcessId": some.int,
|
|
"host": some.str,
|
|
"port": some.int,
|
|
}
|
|
)
|
|
|
|
child_config = parent_session.wait_for_next_event("debugpyAttach")
|
|
assert child_config == expected_child_config
|
|
parent_session.proceed()
|
|
|
|
with debug.Session(child_config) as child_session:
|
|
with child_session.start():
|
|
pass
|
|
|
|
child_pid = backchannel.receive()
|
|
assert child_pid == child_config["subProcessId"]
|
|
assert str(child_pid) in child_config["name"]
|
|
|
|
debugpy_file = backchannel.receive()
|
|
assert debugpy_file == debugpy.__file__
|
|
|
|
child_argv = backchannel.receive()
|
|
assert child_argv == [child, "--arg1", "--arg2", "--arg3"]
|
|
|
|
|
|
def test_autokill(pyfile, target):
|
|
@pyfile
|
|
def child():
|
|
while True:
|
|
pass
|
|
|
|
@pyfile
|
|
def parent():
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
argv = [sys.executable, sys.argv[1]]
|
|
env = os.environ.copy()
|
|
subprocess.Popen(
|
|
argv,
|
|
env=env,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
).wait()
|
|
|
|
with debug.Session() as parent_session:
|
|
parent_session.expected_exit_code = some.int
|
|
|
|
with parent_session.launch(target(parent, args=[child])):
|
|
pass
|
|
|
|
child_config = parent_session.wait_for_next_event("debugpyAttach")
|
|
parent_session.proceed()
|
|
|
|
with debug.Session(child_config) as child_session:
|
|
with child_session.start():
|
|
pass
|
|
|
|
parent_session.request("terminate")
|
|
child_session.wait_for_exit()
|
|
|
|
|
|
def test_argv_quoting(pyfile, target, run):
|
|
@pyfile
|
|
def args():
|
|
args = [ # noqa
|
|
r"regular",
|
|
r"",
|
|
r"with spaces" r'"quoted"',
|
|
r'" quote at start',
|
|
r'quote at end "',
|
|
r'quote in " the middle',
|
|
r'quotes "in the" middle',
|
|
r"\path with\spaces",
|
|
r"\path\with\terminal\backslash" + "\\",
|
|
r"backslash \" before quote",
|
|
]
|
|
|
|
@pyfile
|
|
def parent():
|
|
import debuggee
|
|
import sys
|
|
import subprocess
|
|
from args import args
|
|
|
|
debuggee.setup()
|
|
child = sys.argv[1]
|
|
subprocess.check_call([sys.executable] + [child] + args)
|
|
|
|
@pyfile
|
|
def child():
|
|
import sys
|
|
from debuggee import backchannel
|
|
from args import args as expected_args
|
|
|
|
backchannel.send(expected_args)
|
|
|
|
actual_args = sys.argv[1:]
|
|
backchannel.send(actual_args)
|
|
|
|
with debug.Session() as parent_session:
|
|
backchannel = parent_session.open_backchannel()
|
|
|
|
with run(parent_session, target(parent, args=[child])):
|
|
pass
|
|
|
|
child_config = parent_session.wait_for_next_event("debugpyAttach")
|
|
parent_session.proceed()
|
|
|
|
with debug.Session(child_config) as child_session:
|
|
with child_session.start():
|
|
pass
|
|
|
|
expected_args = backchannel.receive()
|
|
actual_args = backchannel.receive()
|
|
|
|
assert expected_args == actual_args
|
|
|
|
|
|
def test_echo_and_shell(pyfile, target, run):
|
|
"""
|
|
Checks https://github.com/microsoft/ptvsd/issues/1548
|
|
"""
|
|
|
|
@pyfile
|
|
def code_to_run():
|
|
import debuggee
|
|
import sys
|
|
import subprocess
|
|
import os
|
|
|
|
debuggee.setup()
|
|
|
|
if sys.platform == "win32":
|
|
args = ["dir", "-c", "."]
|
|
else:
|
|
args = ["ls", "-c", "-la"]
|
|
|
|
p = subprocess.Popen(
|
|
args,
|
|
shell=True,
|
|
stderr=subprocess.STDOUT,
|
|
stdout=subprocess.PIPE,
|
|
cwd=os.path.dirname(os.path.abspath(__file__)),
|
|
)
|
|
stdout, _stderr = p.communicate()
|
|
if sys.version_info[0] >= 3:
|
|
stdout = stdout.decode("utf-8")
|
|
|
|
if "code_to_run.py" not in stdout:
|
|
raise AssertionError(
|
|
'Did not find "code_to_run.py" when listing this dir with subprocess. Contents: %s'
|
|
% (stdout,)
|
|
)
|
|
|
|
with debug.Session() as parent_session:
|
|
with run(parent_session, target(code_to_run)):
|
|
pass
|