indigo
ec5f52c7c6
Bundles debugpy 1.7.0 directly in python/ so "Maya: Attach (debugpy)" works with no per-machine pip install step. Installed via Maya 2022's own pip so it resolved a version actually compatible with Python 3.7 (Maya 2022's interpreter), then verified import + listen() succeeds under all three target Maya Python versions (3.7/3.9/3.10). Also fixes a latent __file__-under-exec() bug in start_debug_server.py (same pitfall as Maya's own plugin loader, never hit until this exercised it) and corrects the README's Maya Python version claim -- 2022 ships Python 3.7, not 3.9 as previously stated, which was never independently verified until now. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
278 lines
11 KiB
Python
278 lines
11 KiB
Python
import threading
|
|
import unittest
|
|
import sys
|
|
import pydevconsole
|
|
from _pydev_bundle.pydev_imports import xmlrpclib, SimpleXMLRPCServer
|
|
from _pydevd_bundle import pydevd_io
|
|
from contextlib import contextmanager
|
|
import pytest
|
|
|
|
try:
|
|
from ast import PyCF_ALLOW_TOP_LEVEL_AWAIT # @UnusedImport
|
|
CAN_EVALUATE_TOP_LEVEL_ASYNC = True
|
|
except:
|
|
CAN_EVALUATE_TOP_LEVEL_ASYNC = False
|
|
|
|
|
|
#=======================================================================================================================
|
|
# Test
|
|
#=======================================================================================================================
|
|
class Test(unittest.TestCase):
|
|
|
|
@contextmanager
|
|
def interpreter(self):
|
|
self.original_stdout = sys.stdout
|
|
self.original_stderr = sys.stderr
|
|
sys.stdout = pydevd_io.IOBuf()
|
|
sys.stderr = pydevd_io.IOBuf()
|
|
try:
|
|
sys.stdout.encoding = sys.stdin.encoding
|
|
sys.stderr.encoding = sys.stdin.encoding
|
|
except AttributeError:
|
|
# In Python 3 encoding is not writable (whereas in Python 2 it doesn't exist).
|
|
pass
|
|
|
|
try:
|
|
client_port, _server_port = self.get_free_addresses()
|
|
client_thread = self.start_client_thread(client_port) # @UnusedVariable
|
|
import time
|
|
time.sleep(.3) # let's give it some time to start the threads
|
|
|
|
from _pydev_bundle import pydev_localhost
|
|
interpreter = pydevconsole.InterpreterInterface(pydev_localhost.get_localhost(), client_port, threading.current_thread())
|
|
yield interpreter
|
|
except:
|
|
# if there's some error, print the output to the actual output.
|
|
self.original_stdout.write(sys.stdout.getvalue())
|
|
self.original_stderr.write(sys.stderr.getvalue())
|
|
raise
|
|
finally:
|
|
sys.stderr = self.original_stderr
|
|
sys.stdout = self.original_stdout
|
|
|
|
def test_console_hello(self):
|
|
with self.interpreter() as interpreter:
|
|
(result,) = interpreter.hello("Hello pydevconsole")
|
|
self.assertEqual(result, "Hello eclipse")
|
|
|
|
@pytest.mark.skipif(not CAN_EVALUATE_TOP_LEVEL_ASYNC, reason='Requires top-level async.')
|
|
def test_console_async(self):
|
|
with self.interpreter() as interpreter:
|
|
from _pydev_bundle.pydev_console_utils import CodeFragment
|
|
more = interpreter.add_exec(CodeFragment('''
|
|
async def async_func(a):
|
|
return a
|
|
'''))
|
|
assert not more
|
|
assert not sys.stderr.getvalue()
|
|
assert not sys.stdout.getvalue()
|
|
|
|
more = interpreter.add_exec(CodeFragment('''x = await async_func(1111)'''))
|
|
assert not more
|
|
assert not sys.stderr.getvalue()
|
|
assert not sys.stdout.getvalue()
|
|
|
|
more = interpreter.add_exec(CodeFragment('''print(x)'''))
|
|
assert not more
|
|
assert not sys.stderr.getvalue()
|
|
assert '1111' in sys.stdout.getvalue()
|
|
|
|
def test_console_requests(self):
|
|
with self.interpreter() as interpreter:
|
|
from _pydev_bundle.pydev_console_utils import CodeFragment
|
|
interpreter.add_exec(CodeFragment('class Foo:\n CONSTANT=1\n'))
|
|
interpreter.add_exec(CodeFragment('foo=Foo()'))
|
|
interpreter.add_exec(CodeFragment('foo.__doc__=None'))
|
|
interpreter.add_exec(CodeFragment('val = input()'))
|
|
interpreter.add_exec(CodeFragment('50'))
|
|
interpreter.add_exec(CodeFragment('print (val)'))
|
|
found = sys.stdout.getvalue().split()
|
|
try:
|
|
self.assertEqual(['50', 'input_request'], found)
|
|
except:
|
|
try:
|
|
self.assertEqual(['input_request'], found) # IPython
|
|
except:
|
|
self.assertEqual([u'50', u'input_request'], found[1:]) # IPython 5.1
|
|
self.assertTrue(found[0].startswith(u'Out'))
|
|
|
|
comps = interpreter.getCompletions('foo.', 'foo.')
|
|
self.assertTrue(
|
|
('CONSTANT', '', '', '3') in comps or ('CONSTANT', '', '', '4') in comps, \
|
|
'Found: %s' % comps
|
|
)
|
|
|
|
comps = interpreter.getCompletions('"".', '"".')
|
|
self.assertTrue(
|
|
('__add__', 'x.__add__(y) <==> x+y', '', '3') in comps or
|
|
('__add__', '', '', '4') in comps or
|
|
('__add__', 'x.__add__(y) <==> x+y\r\nx.__add__(y) <==> x+y', '()', '2') in comps or
|
|
('__add__', 'x.\n__add__(y) <==> x+yx.\n__add__(y) <==> x+y', '()', '2'),
|
|
'Did not find __add__ in : %s' % (comps,)
|
|
)
|
|
|
|
completions = interpreter.getCompletions('', '')
|
|
for c in completions:
|
|
if c[0] == 'AssertionError':
|
|
break
|
|
else:
|
|
self.fail('Could not find AssertionError')
|
|
|
|
completions = interpreter.getCompletions('Assert', 'Assert')
|
|
for c in completions:
|
|
if c[0] == 'RuntimeError':
|
|
self.fail('Did not expect to find RuntimeError there')
|
|
|
|
assert ('__doc__', None, '', '3') not in interpreter.getCompletions('foo.CO', 'foo.')
|
|
|
|
comps = interpreter.getCompletions('va', 'va')
|
|
assert ('val', '', '', '3') in comps or ('val', '', '', '4') in comps
|
|
|
|
interpreter.add_exec(CodeFragment('s = "mystring"'))
|
|
|
|
desc = interpreter.getDescription('val')
|
|
self.assertTrue(desc.find('str(object) -> string') >= 0 or
|
|
desc == "'input_request'" or
|
|
desc.find('str(string[, encoding[, errors]]) -> str') >= 0 or
|
|
desc.find('str(Char* value)') >= 0 or
|
|
desc.find('str(object=\'\') -> string') >= 0 or
|
|
desc.find('str(value: Char*)') >= 0 or
|
|
desc.find('str(object=\'\') -> str') >= 0 or
|
|
desc.find('The most base type') >= 0 # Jython 2.7 is providing this :P
|
|
,
|
|
'Could not find what was needed in %s' % desc)
|
|
|
|
desc = interpreter.getDescription('val.join')
|
|
self.assertTrue(desc.find('S.join(sequence) -> string') >= 0 or
|
|
desc.find('S.join(sequence) -> str') >= 0 or
|
|
desc.find('S.join(iterable) -> string') >= 0 or
|
|
desc == "<builtin method 'join'>" or
|
|
desc == "<built-in method join of str object>" or
|
|
desc.find('str join(str self, list sequence)') >= 0 or
|
|
desc.find('S.join(iterable) -> str') >= 0 or
|
|
desc.find('join(self: str, sequence: list) -> str') >= 0 or
|
|
desc.find('Concatenate any number of strings.') >= 0 or
|
|
desc.find('bound method str.join') >= 0, # PyPy
|
|
"Could not recognize: %s" % (desc,))
|
|
|
|
def start_client_thread(self, client_port):
|
|
|
|
class ClientThread(threading.Thread):
|
|
|
|
def __init__(self, client_port):
|
|
threading.Thread.__init__(self)
|
|
self.client_port = client_port
|
|
|
|
def run(self):
|
|
|
|
class HandleRequestInput:
|
|
|
|
def RequestInput(self):
|
|
client_thread.requested_input = True
|
|
return 'input_request'
|
|
|
|
def NotifyFinished(self, *args, **kwargs):
|
|
client_thread.notified_finished += 1
|
|
return 1
|
|
|
|
handle_request_input = HandleRequestInput()
|
|
|
|
from _pydev_bundle import pydev_localhost
|
|
client_server = SimpleXMLRPCServer((pydev_localhost.get_localhost(), self.client_port), logRequests=False)
|
|
client_server.register_function(handle_request_input.RequestInput)
|
|
client_server.register_function(handle_request_input.NotifyFinished)
|
|
client_server.serve_forever()
|
|
|
|
client_thread = ClientThread(client_port)
|
|
client_thread.requested_input = False
|
|
client_thread.notified_finished = 0
|
|
client_thread.daemon = True
|
|
client_thread.start()
|
|
return client_thread
|
|
|
|
def start_debugger_server_thread(self, debugger_port, socket_code):
|
|
|
|
class DebuggerServerThread(threading.Thread):
|
|
|
|
def __init__(self, debugger_port, socket_code):
|
|
threading.Thread.__init__(self)
|
|
self.debugger_port = debugger_port
|
|
self.socket_code = socket_code
|
|
|
|
def run(self):
|
|
import socket
|
|
s = socket.socket()
|
|
s.bind(('', debugger_port))
|
|
s.listen(1)
|
|
socket, unused_addr = s.accept()
|
|
socket_code(socket)
|
|
|
|
debugger_thread = DebuggerServerThread(debugger_port, socket_code)
|
|
debugger_thread.daemon = True
|
|
debugger_thread.start()
|
|
return debugger_thread
|
|
|
|
def get_free_addresses(self):
|
|
from _pydev_bundle.pydev_localhost import get_socket_names
|
|
socket_names = get_socket_names(2, True)
|
|
port0 = socket_names[0][1]
|
|
port1 = socket_names[1][1]
|
|
|
|
assert port0 != port1
|
|
assert port0 > 0
|
|
assert port1 > 0
|
|
|
|
return port0, port1
|
|
|
|
def test_server(self):
|
|
self.original_stdout = sys.stdout
|
|
sys.stdout = pydevd_io.IOBuf()
|
|
try:
|
|
client_port, server_port = self.get_free_addresses()
|
|
|
|
class ServerThread(threading.Thread):
|
|
|
|
def __init__(self, client_port, server_port):
|
|
threading.Thread.__init__(self)
|
|
self.client_port = client_port
|
|
self.server_port = server_port
|
|
|
|
def run(self):
|
|
from _pydev_bundle import pydev_localhost
|
|
pydevconsole.start_server(pydev_localhost.get_localhost(), self.server_port, self.client_port)
|
|
|
|
server_thread = ServerThread(client_port, server_port)
|
|
server_thread.daemon = True
|
|
server_thread.start()
|
|
|
|
client_thread = self.start_client_thread(client_port) # @UnusedVariable
|
|
|
|
import time
|
|
time.sleep(.3) # let's give it some time to start the threads
|
|
sys.stdout = pydevd_io.IOBuf()
|
|
|
|
from _pydev_bundle import pydev_localhost
|
|
server = xmlrpclib.Server('http://%s:%s' % (pydev_localhost.get_localhost(), server_port))
|
|
server.execLine('class Foo:')
|
|
server.execLine(' pass')
|
|
server.execLine('')
|
|
server.execLine('foo = Foo()')
|
|
server.execLine('a = input()')
|
|
server.execLine('print (a)')
|
|
initial = time.time()
|
|
while not client_thread.requested_input:
|
|
if time.time() - initial > 2:
|
|
raise AssertionError('Did not get the return asked before the timeout.')
|
|
time.sleep(.1)
|
|
|
|
found = sys.stdout.getvalue()
|
|
while ['input_request'] != found.split():
|
|
found += sys.stdout.getvalue()
|
|
if time.time() - initial > 2:
|
|
break
|
|
time.sleep(.1)
|
|
self.assertEqual(['input_request'], found.split())
|
|
finally:
|
|
sys.stdout = self.original_stdout
|
|
|