from contextlib import contextmanager

import pytest

from _pydev_bundle.pydev_override import overrides

from tests_python.debugger_fixtures import DebuggerRunnerSimple, debugger_runner_simple

from tests_python.debugger_unittest import (

AbstractWriterThread,

SHOW_OTHER_DEBUG_INFO,

start_in_daemon_thread,

wait_for_condition,

IS_JYTHON,

)

from _pydev_bundle.pydev_localhost import get_socket_names, get_socket_name

from _pydev_bundle.pydev_imports import xmlrpclib

from _pydev_bundle.pydev_imports import _queue as queue

from _pydev_bundle.pydev_imports import SimpleXMLRPCServer

import time

import socket

builtin_qualifier = "builtins"

@pytest.fixture

def console_setup(tmpdir):

server_queue = queue.Queue()

def notify_finished(more):

server_queue.put(("notify_finished", more))

return ""

class ConsoleRunner(DebuggerRunnerSimple):

@overrides(DebuggerRunnerSimple.add_command_line_args)

def add_command_line_args(self, args, dap=False):

port, client_port = get_socket_names(2, close=True)

args.extend((writer.get_pydevconsole_file(), str(port[1]), str(client_port[1])))

self.port = port

self.client_port = client_port

server = SimpleXMLRPCServer(client_port)

server.register_function(notify_finished, "NotifyFinished")

start_in_daemon_thread(server.serve_forever, [])

self.proxy = xmlrpclib.ServerProxy("http://%s:%s/" % port)

return args

class WriterThread(AbstractWriterThread):

if IS_JYTHON:

EXPECTED_RETURNCODE = "any"

@overrides(AbstractWriterThread.additional_output_checks)

def additional_output_checks(self, stdout, stderr):

print("output found: %s - %s" % (stdout, stderr))

@overrides(AbstractWriterThread.write_dump_threads)

def write_dump_threads(self):

pass # no-op (may be called on timeout).

def execute_line(self, command, more=False):

runner.proxy.execLine(command)

assert server_queue.get(timeout=5.0) == ("notify_finished", more)

def hello(self):

def _hello():

try:

msg = runner.proxy.hello("ignored")

if msg is not None:

if isinstance(msg, (list, tuple)):

msg = next(iter(msg))

if msg.lower().startswith("hello"):

return True

except:

# That's ok, communication still not ready.

pass

return False

wait_for_condition(_hello)

def close(self):

try:

runner.proxy.close()

except:

# Ignore any errors on close.

pass

def connect_to_debugger(self, debugger_port):

runner.proxy.connectToDebugger(debugger_port)

runner = ConsoleRunner(tmpdir)

writer = WriterThread()

class CaseSetup(object):

@contextmanager

def check_console(self, **kwargs):

for key, value in kwargs.items():

assert hasattr(WriterThread, key)

setattr(WriterThread, key, value)

self.writer = writer

args = runner.get_command_line()

args = runner.add_command_line_args(args)

if SHOW_OTHER_DEBUG_INFO:

print("executing: %s" % (" ".join(args),))

try:

with runner.run_process(args, writer) as dct_with_stdout_stder:

writer.get_stdout = lambda: "".join(dct_with_stdout_stder["stdout"])

writer.get_stderr = lambda: "".join(dct_with_stdout_stder["stderr"])

# Make sure communication is setup.

writer.hello()

yield writer

finally:

writer.log = []

stdout = dct_with_stdout_stder["stdout"]

stderr = dct_with_stdout_stder["stderr"]

writer.additional_output_checks("".join(stdout), "".join(stderr))

return CaseSetup()

def test_console_simple(console_setup):

with console_setup.check_console() as writer:

writer.execute_line("a = 10")

writer.execute_line('print("TEST SUCEEDED")')

writer.close()

writer.finished_ok = True

def test_console_debugger_connected(console_setup):

class _DebuggerWriterThread(AbstractWriterThread):

FORCE_KILL_PROCESS_WHEN_FINISHED_OK = True

def __init__(self):

AbstractWriterThread.__init__(self)

socket_name = get_socket_name(close=True)

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

server_socket.bind(socket_name)

server_socket.listen(1)

self.port = socket_name[1]

self.__server_socket = server_socket

def run(self):

print("waiting for second process")

self.sock, addr = self.__server_socket.accept()

print("accepted second process")

from tests_python.debugger_unittest import ReaderThread

self.reader_thread = ReaderThread(self.sock)

self.reader_thread.start()

self._sequence = -1

# initial command is always the version

self.write_version()

self.log.append("start_socket")

self.write_make_initial_run()

time.sleep(1)

seq = self.write_list_threads()

msg = self.wait_for_list_threads(seq)

assert msg.thread["name"] == "MainThread"

assert msg.thread["id"] == "console_main"

self.write_get_frame("console_main", "1")

self.wait_for_vars(

[

[

'<var name="a" type="int" qualifier="%s" value="int: 10"' % (builtin_qualifier,),

'<var name="a" type="int" value="int', # jython

],

]

)

self.finished_ok = True

with console_setup.check_console() as writer:

writer.execute_line("a = 10")

debugger_writer_thread = _DebuggerWriterThread()

debugger_writer_thread.start()

writer.connect_to_debugger(debugger_writer_thread.port)

wait_for_condition(lambda: debugger_writer_thread.finished_ok)

writer.execute_line('print("TEST SUCEEDED")')

writer.close()

writer.finished_ok = True