From 60d4148131c971742373754ada24fb501cc2d0cc Mon Sep 17 00:00:00 2001 From: Eric Snow Date: Wed, 11 Apr 2018 20:48:02 +0000 Subject: [PATCH] Move VSCMessages and PyDevdMessages under the helpers package. --- tests/helpers/pydevd/__init__.py | 1 + tests/helpers/pydevd/_messages.py | 103 +++++++++++++++++++ tests/helpers/vsc/__init__.py | 1 + tests/helpers/vsc/_messages.py | 64 ++++++++++++ tests/highlevel/__init__.py | 163 +----------------------------- 5 files changed, 171 insertions(+), 161 deletions(-) create mode 100644 tests/helpers/pydevd/_messages.py create mode 100644 tests/helpers/vsc/_messages.py diff --git a/tests/helpers/pydevd/__init__.py b/tests/helpers/pydevd/__init__.py index 6c5a9e8e..4cbc8eb9 100644 --- a/tests/helpers/pydevd/__init__.py +++ b/tests/helpers/pydevd/__init__.py @@ -1,2 +1,3 @@ from ._fake import FakePyDevd # noqa +from ._messages import PyDevdMessages # noqa from ._pydevd import RawMessage # noqa diff --git a/tests/helpers/pydevd/_messages.py b/tests/helpers/pydevd/_messages.py new file mode 100644 index 00000000..3c9d59ee --- /dev/null +++ b/tests/helpers/pydevd/_messages.py @@ -0,0 +1,103 @@ +try: + import urllib.parse as urllib +except ImportError: + import urllib + +from _pydevd_bundle import pydevd_xml +from _pydevd_bundle.pydevd_comm import ( + CMD_SEND_CURR_EXCEPTION_TRACE, +) + +from tests.helpers.protocol import MessageCounters +from ._fake import FakePyDevd + + +class PyDevdMessages(object): + + protocol = FakePyDevd.PROTOCOL + + def __init__(self, + request_seq=1000000000, # ptvsd requests to pydevd + response_seq=0, # PyDevd responses/events to ptvsd + event_seq=None, + ): + self.counters = MessageCounters( + request_seq, + response_seq, + event_seq, + ) + + def __getattr__(self, name): + return getattr(self.counters, name) + + def new_request(self, cmdid, *args, **kwargs): + """Return a new PyDevd request message.""" + seq = kwargs.pop('seq', None) + if seq is None: + seq = self.counters.next_request() + return self._new_message(cmdid, seq, args, **kwargs) + + def new_response(self, req, *args): + """Return a new VSC response message.""" + #seq = kwargs.pop('seq', None) + #if seq is None: + # seq = next(self.response_seq) + req = self.protocol.parse(req) + return self._new_message(req.cmdid, req.seq, args) + + def new_event(self, cmdid, *args, **kwargs): + """Return a new VSC event message.""" + seq = kwargs.pop('seq', None) + if seq is None: + seq = self.counters.next_event() + return self._new_message(cmdid, seq, args, **kwargs) + + def _new_message(self, cmdid, seq, args=()): + text = '\t'.join(args) + msg = (cmdid, seq, text) + return self.protocol.parse(msg) + + def format_threads(self, *threads): + text = '' + for thread in threads: # (tid, tname) + text += ''.format(*thread) + text += '' + return text + + def format_frames(self, threadid, reason, *frames): + text = '' + text += ''.format(threadid, reason) + fmt = '' + for frame in frames: # (fid, func, filename, line) + text += fmt.format(*frame) + text += '' + text += '' + return text + + def format_variables(self, *variables): + text = '' + for name, value in variables: + if isinstance(value, str) and value.startswith('err:'): + value = pydevd_xml.ExceptionOnEvaluate(value[4:]) + text += pydevd_xml.var_to_xml(value, name) + text += '' + return urllib.quote(text) + + def format_exception(self, threadid, exc, frame): + frameid, _, _, _ = frame + name = pydevd_xml.make_valid_xml_value(type(exc).__name__) + description = pydevd_xml.make_valid_xml_value(str(exc)) + + info = '' + info += ''.format(threadid) + info += '' + return '{}\t{}\t{}\t{}'.format( + frameid, + name or 'exception: type unknown', + description or 'exception: no description', + self.format_frames( + threadid, + CMD_SEND_CURR_EXCEPTION_TRACE, + frame, + ), + ) diff --git a/tests/helpers/vsc/__init__.py b/tests/helpers/vsc/__init__.py index 066979a1..133bac2b 100644 --- a/tests/helpers/vsc/__init__.py +++ b/tests/helpers/vsc/__init__.py @@ -1,2 +1,3 @@ from ._fake import FakeVSC # noqa +from ._messages import VSCMessages # noqa from ._vsc import parse_message, RawMessage, Request, Response, Event # noqa diff --git a/tests/helpers/vsc/_messages.py b/tests/helpers/vsc/_messages.py new file mode 100644 index 00000000..8005e2e6 --- /dev/null +++ b/tests/helpers/vsc/_messages.py @@ -0,0 +1,64 @@ +from tests.helpers.protocol import MessageCounters +from ._fake import FakeVSC + + +class VSCMessages(object): + + protocol = FakeVSC.PROTOCOL + + def __init__(self, + request_seq=0, # VSC requests to ptvsd + response_seq=0, # ptvsd responses/events to VSC + event_seq=None, + ): + self.counters = MessageCounters( + request_seq, + response_seq, + event_seq, + ) + + def __getattr__(self, name): + return getattr(self.counters, name) + + def new_request(self, command, seq=None, **args): + """Return a new VSC request message.""" + if seq is None: + seq = self.counters.next_request() + return { + 'type': 'request', + 'seq': seq, + 'command': command, + 'arguments': args, + } + + def new_response(self, req, seq=None, **body): + """Return a new VSC response message.""" + return self._new_response(req, None, seq, body) + + def new_failure(self, req, err, seq=None, **body): + """Return a new VSC response message.""" + return self._new_response(req, err, body=body) + + def _new_response(self, req, err=None, seq=None, body=None): + if seq is None: + seq = self.counters.next_response() + return { + 'type': 'response', + 'seq': seq, + 'request_seq': req['seq'], + 'command': req['command'], + 'success': err is None, + 'message': err or '', + 'body': body, + } + + def new_event(self, eventname, seq=None, **body): + """Return a new VSC event message.""" + if seq is None: + seq = self.counters.next_event() + return { + 'type': 'event', + 'seq': seq, + 'event': eventname, + 'body': body, + } diff --git a/tests/highlevel/__init__.py b/tests/highlevel/__init__.py index fce0724c..69441fb8 100644 --- a/tests/highlevel/__init__.py +++ b/tests/highlevel/__init__.py @@ -2,13 +2,8 @@ from collections import namedtuple import contextlib import platform import threading -try: - import urllib.parse as urllib -except ImportError: - import urllib import warnings -from _pydevd_bundle import pydevd_xml from _pydevd_bundle.pydevd_comm import ( CMD_VERSION, CMD_LIST_THREADS, @@ -23,9 +18,8 @@ from _pydevd_bundle.pydevd_comm import ( ) from ptvsd import wrapper -from tests.helpers.protocol import MessageCounters -from tests.helpers.pydevd import FakePyDevd -from tests.helpers.vsc import FakeVSC +from tests.helpers.pydevd import FakePyDevd, PyDevdMessages +from tests.helpers.vsc import FakeVSC, VSCMessages OS_ID = 'WINDOWS' if platform.system() == 'Windows' else 'UNIX' @@ -127,159 +121,6 @@ class Threads(object): self._remove(t) -class PyDevdMessages(object): - - protocol = FakePyDevd.PROTOCOL - - def __init__(self, - request_seq=1000000000, # ptvsd requests to pydevd - response_seq=0, # PyDevd responses/events to ptvsd - event_seq=None, - ): - self.counters = MessageCounters( - request_seq, - response_seq, - event_seq, - ) - - def __getattr__(self, name): - return getattr(self.counters, name) - - def new_request(self, cmdid, *args, **kwargs): - """Return a new PyDevd request message.""" - seq = kwargs.pop('seq', None) - if seq is None: - seq = self.counters.next_request() - return self._new_message(cmdid, seq, args, **kwargs) - - def new_response(self, req, *args): - """Return a new VSC response message.""" - #seq = kwargs.pop('seq', None) - #if seq is None: - # seq = next(self.response_seq) - req = self.protocol.parse(req) - return self._new_message(req.cmdid, req.seq, args) - - def new_event(self, cmdid, *args, **kwargs): - """Return a new VSC event message.""" - seq = kwargs.pop('seq', None) - if seq is None: - seq = self.counters.next_event() - return self._new_message(cmdid, seq, args, **kwargs) - - def _new_message(self, cmdid, seq, args=()): - text = '\t'.join(args) - msg = (cmdid, seq, text) - return self.protocol.parse(msg) - - def format_threads(self, *threads): - text = '' - for thread in threads: # (tid, tname) - text += ''.format(*thread) - text += '' - return text - - def format_frames(self, threadid, reason, *frames): - text = '' - text += ''.format(threadid, reason) - fmt = '' - for frame in frames: # (fid, func, filename, line) - text += fmt.format(*frame) - text += '' - text += '' - return text - - def format_variables(self, *variables): - text = '' - for name, value in variables: - if isinstance(value, str) and value.startswith('err:'): - value = pydevd_xml.ExceptionOnEvaluate(value[4:]) - text += pydevd_xml.var_to_xml(value, name) - text += '' - return urllib.quote(text) - - def format_exception(self, threadid, exc, frame): - frameid, _, _, _ = frame - name = pydevd_xml.make_valid_xml_value(type(exc).__name__) - description = pydevd_xml.make_valid_xml_value(str(exc)) - - info = '' - info += ''.format(threadid) - info += '' - return '{}\t{}\t{}\t{}'.format( - frameid, - name or 'exception: type unknown', - description or 'exception: no description', - self.format_frames( - threadid, - CMD_SEND_CURR_EXCEPTION_TRACE, - frame, - ), - ) - - -class VSCMessages(object): - - protocol = FakeVSC.PROTOCOL - - def __init__(self, - request_seq=0, # VSC requests to ptvsd - response_seq=0, # ptvsd responses/events to VSC - event_seq=None, - ): - self.counters = MessageCounters( - request_seq, - response_seq, - event_seq, - ) - - def __getattr__(self, name): - return getattr(self.counters, name) - - def new_request(self, command, seq=None, **args): - """Return a new VSC request message.""" - if seq is None: - seq = self.counters.next_request() - return { - 'type': 'request', - 'seq': seq, - 'command': command, - 'arguments': args, - } - - def new_response(self, req, seq=None, **body): - """Return a new VSC response message.""" - return self._new_response(req, None, seq, body) - - def new_failure(self, req, err, seq=None, **body): - """Return a new VSC response message.""" - return self._new_response(req, err, body=body) - - def _new_response(self, req, err=None, seq=None, body=None): - if seq is None: - seq = self.counters.next_response() - return { - 'type': 'response', - 'seq': seq, - 'request_seq': req['seq'], - 'command': req['command'], - 'success': err is None, - 'message': err or '', - 'body': body, - } - - def new_event(self, eventname, seq=None, **body): - """Return a new VSC event message.""" - if seq is None: - seq = self.counters.next_event() - return { - 'type': 'event', - 'seq': seq, - 'event': eventname, - 'body': body, - } - - class PyDevdLifecycle(object): def __init__(self, fix):