mirror of
https://github.com/microsoft/debugpy.git
synced 2025-12-23 08:48:12 +00:00
Add the base message types.
This commit is contained in:
Eric Snow
parent
b4abe56051
commit
ba0e9963c0
4 changed files with 1302 additions and 22 deletions
|
|
@ -1,22 +0,0 @@
|
|||
from debugger_protocol._base import Readonly, WithRepr
|
||||
|
||||
|
||||
class Base(Readonly, WithRepr):
|
||||
"""Base class for message-related types."""
|
||||
|
||||
_INIT_ARGS = None
|
||||
|
||||
@classmethod
|
||||
def from_data(cls, **kwargs):
|
||||
"""Return an instance based on the given raw data."""
|
||||
return cls(**kwargs)
|
||||
|
||||
def __init__(self):
|
||||
self._validate()
|
||||
|
||||
def _validate(self):
|
||||
pass
|
||||
|
||||
def as_data(self):
|
||||
"""Return serializable data for the instance."""
|
||||
return {}
|
||||
366
debugger_protocol/messages/message.py
Normal file
366
debugger_protocol/messages/message.py
Normal file
|
|
@ -0,0 +1,366 @@
|
|||
from debugger_protocol._base import Readonly, WithRepr
|
||||
from debugger_protocol.arg import param_from_datatype
|
||||
from . import MESSAGE_TYPES, Message
|
||||
|
||||
"""
|
||||
From the schema:
|
||||
|
||||
MESSAGE = [
|
||||
name
|
||||
base
|
||||
description
|
||||
props: [PROPERTY + (properties: [PROPERTY])]
|
||||
]
|
||||
|
||||
PROPERTY = [
|
||||
name
|
||||
type: choices (one or a list)
|
||||
(enum/_enum)
|
||||
description
|
||||
required: True/False (default: False)
|
||||
]
|
||||
|
||||
inheritance: override properties of base
|
||||
"""
|
||||
|
||||
|
||||
class ProtocolMessage(Readonly, WithRepr, Message):
|
||||
"""Base class of requests, responses, and events."""
|
||||
|
||||
_reqid = 0
|
||||
TYPE = None
|
||||
|
||||
@classmethod
|
||||
def from_data(cls, type, seq, **kwargs):
|
||||
"""Return an instance based on the given raw data."""
|
||||
return cls(type=type, seq=seq, **kwargs)
|
||||
|
||||
@classmethod
|
||||
def _next_reqid(cls):
|
||||
reqid = ProtocolMessage._reqid
|
||||
ProtocolMessage._reqid += 1
|
||||
return reqid
|
||||
|
||||
_NOT_SET = object()
|
||||
|
||||
def __init__(self, seq=_NOT_SET, **kwargs):
|
||||
type = kwargs.pop('type', self.TYPE)
|
||||
if seq is self._NOT_SET:
|
||||
seq = self._next_reqid()
|
||||
self._bind_attrs(
|
||||
type=type or None,
|
||||
seq=int(seq) if seq or seq == 0 else None,
|
||||
)
|
||||
self._validate()
|
||||
|
||||
def _validate(self):
|
||||
if self.type is None:
|
||||
raise TypeError('missing type')
|
||||
elif self.TYPE is not None and self.type != self.TYPE:
|
||||
raise ValueError('type must be {!r}'.format(self.TYPE))
|
||||
elif self.type not in MESSAGE_TYPES:
|
||||
raise ValueError('unsupported type {!r}'.format(self.type))
|
||||
|
||||
if self.seq is None:
|
||||
raise TypeError('missing seq')
|
||||
elif self.seq < 0:
|
||||
msg = '"seq" must be a non-negative int, got {!r}'
|
||||
raise ValueError(msg.format(self.seq))
|
||||
|
||||
def _init_args(self):
|
||||
if self.TYPE is None:
|
||||
yield ('type', self.type)
|
||||
yield ('seq', self.seq)
|
||||
|
||||
def as_data(self):
|
||||
"""Return serializable data for the instance."""
|
||||
data = {
|
||||
'type': self.type,
|
||||
'seq': self.seq,
|
||||
}
|
||||
return data
|
||||
|
||||
|
||||
##################################
|
||||
|
||||
class Request(ProtocolMessage):
|
||||
"""A client or server-initiated request."""
|
||||
|
||||
TYPE = 'request'
|
||||
TYPE_KEY = 'command'
|
||||
|
||||
COMMAND = None
|
||||
ARGUMENTS = None
|
||||
ARGUMENTS_REQUIRED = None
|
||||
|
||||
@classmethod
|
||||
def from_data(cls, type, seq, command, arguments=None):
|
||||
"""Return an instance based on the given raw data."""
|
||||
return super(Request, cls).from_data(
|
||||
type, seq,
|
||||
command=command,
|
||||
arguments=arguments,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _arguments_required(cls):
|
||||
if cls.ARGUMENTS_REQUIRED is None:
|
||||
return cls.ARGUMENTS is not None
|
||||
return cls.ARGUMENTS_REQUIRED
|
||||
|
||||
def __init__(self, arguments=None, **kwargs):
|
||||
command = kwargs.pop('command', self.COMMAND)
|
||||
args = None
|
||||
if arguments is not None:
|
||||
try:
|
||||
arguments = dict(arguments)
|
||||
except TypeError:
|
||||
pass
|
||||
if self.ARGUMENTS is not None:
|
||||
param = param_from_datatype(self.ARGUMENTS)
|
||||
args = param.bind(arguments)
|
||||
if args is None:
|
||||
raise TypeError('bad arguments {!r}'.format(arguments))
|
||||
arguments = args.coerce()
|
||||
self._bind_attrs(
|
||||
command=command or None,
|
||||
arguments=arguments or None,
|
||||
_args=args,
|
||||
)
|
||||
super(Request, self).__init__(**kwargs)
|
||||
|
||||
def _validate(self):
|
||||
super(Request, self)._validate()
|
||||
|
||||
if self.command is None:
|
||||
raise TypeError('missing command')
|
||||
elif self.COMMAND is not None and self.command != self.COMMAND:
|
||||
raise ValueError('command must be {!r}'.format(self.COMMAND))
|
||||
|
||||
if self.arguments is None:
|
||||
if self._arguments_required():
|
||||
raise TypeError('missing arguments')
|
||||
else:
|
||||
if self.ARGUMENTS is None:
|
||||
raise TypeError('got unexpected arguments')
|
||||
self._args.validate()
|
||||
|
||||
def _init_args(self):
|
||||
if self.COMMAND is None:
|
||||
yield ('command', self.command)
|
||||
if self.arguments is not None:
|
||||
yield ('arguments', self.arguments)
|
||||
yield ('seq', self.seq)
|
||||
|
||||
def as_data(self):
|
||||
"""Return serializable data for the instance."""
|
||||
data = super(Request, self).as_data()
|
||||
data.update({
|
||||
'command': self.command,
|
||||
})
|
||||
if self.arguments is not None:
|
||||
data.update({
|
||||
'arguments': self.arguments.as_data(),
|
||||
})
|
||||
return data
|
||||
|
||||
|
||||
class Response(ProtocolMessage):
|
||||
"""Response to a request."""
|
||||
|
||||
TYPE = 'response'
|
||||
TYPE_KEY = 'command'
|
||||
|
||||
COMMAND = None
|
||||
BODY = None
|
||||
ERROR_BODY = None
|
||||
BODY_REQUIRED = None
|
||||
ERROR_BODY_REQUIRED = None
|
||||
|
||||
@classmethod
|
||||
def from_data(cls, type, seq, request_seq, command, success,
|
||||
body=None, message=None):
|
||||
"""Return an instance based on the given raw data."""
|
||||
return super(Response, cls).from_data(
|
||||
type, seq,
|
||||
request_seq=request_seq,
|
||||
command=command,
|
||||
success=success,
|
||||
body=body,
|
||||
message=message,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _body_required(cls, success=True):
|
||||
required = cls.BODY_REQUIRED if success else cls.ERROR_BODY_REQUIRED
|
||||
if required is not None:
|
||||
return required
|
||||
bodyclass = cls.BODY if success else cls.ERROR_BODY
|
||||
return bodyclass is not None
|
||||
|
||||
def __init__(self, request_seq, body=None, message=None, success=True,
|
||||
**kwargs):
|
||||
command = kwargs.pop('command', self.COMMAND)
|
||||
reqseq = request_seq
|
||||
bodyarg = None
|
||||
if body is not None:
|
||||
try:
|
||||
body = dict(body)
|
||||
except TypeError:
|
||||
pass
|
||||
bodyclass = self.BODY if success else self.ERROR_BODY
|
||||
if bodyclass is not None:
|
||||
param = param_from_datatype(bodyclass)
|
||||
bodyarg = param.bind(body)
|
||||
if bodyarg is None:
|
||||
raise TypeError('bad body type {!r}'.format(body))
|
||||
body = bodyarg.coerce()
|
||||
self._bind_attrs(
|
||||
command=command or None,
|
||||
request_seq=int(reqseq) if reqseq or reqseq == 0 else None,
|
||||
body=body or None,
|
||||
_bodyarg=bodyarg,
|
||||
message=message or None,
|
||||
success=bool(success),
|
||||
)
|
||||
super(Response, self).__init__(**kwargs)
|
||||
|
||||
def _validate(self):
|
||||
super(Response, self)._validate()
|
||||
|
||||
if self.request_seq is None:
|
||||
raise TypeError('missing request_seq')
|
||||
elif self.request_seq < 0:
|
||||
msg = 'request_seq must be a non-negative int, got {!r}'
|
||||
raise ValueError(msg.format(self.request_seq))
|
||||
|
||||
if not self.command:
|
||||
raise TypeError('missing command')
|
||||
elif self.COMMAND is not None and self.command != self.COMMAND:
|
||||
raise ValueError('command must be {!r}'.format(self.COMMAND))
|
||||
|
||||
if self.body is None:
|
||||
if self._body_required(self.success):
|
||||
raise TypeError('missing body')
|
||||
elif self._bodyarg is None:
|
||||
raise ValueError('got unexpected body')
|
||||
else:
|
||||
self._bodyarg.validate()
|
||||
|
||||
if not self.success and not self.message:
|
||||
raise TypeError('missing message')
|
||||
|
||||
def _init_args(self):
|
||||
if self.COMMAND is None:
|
||||
yield ('command', self.command)
|
||||
yield ('request_seq', self.request_seq)
|
||||
yield ('success', self.success)
|
||||
if not self.success:
|
||||
yield ('message', self.message)
|
||||
if self.body is not None:
|
||||
yield ('body', self.body)
|
||||
yield ('seq', self.seq)
|
||||
|
||||
def as_data(self):
|
||||
"""Return serializable data for the instance."""
|
||||
data = super(Response, self).as_data()
|
||||
data.update({
|
||||
'request_seq': self.request_seq,
|
||||
'command': self.command,
|
||||
'success': self.success,
|
||||
})
|
||||
if self.body is not None:
|
||||
data.update({
|
||||
'body': self.body.as_data(),
|
||||
})
|
||||
if self.message is not None:
|
||||
data.update({
|
||||
'message': self.message,
|
||||
})
|
||||
return data
|
||||
|
||||
|
||||
##################################
|
||||
|
||||
class Event(ProtocolMessage):
|
||||
"""Server-initiated event."""
|
||||
|
||||
TYPE = 'event'
|
||||
TYPE_KEY = 'event'
|
||||
|
||||
EVENT = None
|
||||
BODY = None
|
||||
BODY_REQUIRED = None
|
||||
|
||||
@classmethod
|
||||
def from_data(cls, type, seq, event, body=None):
|
||||
"""Return an instance based on the given raw data."""
|
||||
return super(Event, cls).from_data(type, seq, event=event, body=body)
|
||||
|
||||
@classmethod
|
||||
def _body_required(cls):
|
||||
if cls.BODY_REQUIRED is None:
|
||||
return cls.BODY is not None
|
||||
return cls.BODY_REQUIRED
|
||||
|
||||
def __init__(self, body=None, **kwargs):
|
||||
event = kwargs.pop('event', self.EVENT)
|
||||
bodyarg = None
|
||||
if body is not None:
|
||||
try:
|
||||
body = dict(body)
|
||||
except TypeError:
|
||||
pass
|
||||
if self.BODY is not None:
|
||||
param = param_from_datatype(self.BODY)
|
||||
bodyarg = param.bind(body)
|
||||
if bodyarg is None:
|
||||
raise TypeError('bad body type {!r}'.format(body))
|
||||
body = bodyarg.coerce()
|
||||
|
||||
self._bind_attrs(
|
||||
event=event or None,
|
||||
body=body or None,
|
||||
_bodyarg=bodyarg,
|
||||
)
|
||||
super(Event, self).__init__(**kwargs)
|
||||
|
||||
def _validate(self):
|
||||
super(Event, self)._validate()
|
||||
|
||||
if self.event is None:
|
||||
raise TypeError('missing event')
|
||||
if self.EVENT is not None and self.event != self.EVENT:
|
||||
msg = 'event must be {!r}, got {!r}'
|
||||
raise ValueError(msg.format(self.EVENT, self.event))
|
||||
|
||||
if self.body is None:
|
||||
if self._body_required():
|
||||
raise TypeError('missing body')
|
||||
elif self._bodyarg is None:
|
||||
raise ValueError('got unexpected body')
|
||||
else:
|
||||
self._bodyarg.validate()
|
||||
|
||||
def _init_args(self):
|
||||
if self.EVENT is None:
|
||||
yield ('event', self.event)
|
||||
if self.body is not None:
|
||||
yield ('body', self.body)
|
||||
yield ('seq', self.seq)
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return self.event
|
||||
|
||||
def as_data(self):
|
||||
"""Return serializable data for the instance."""
|
||||
data = super(Event, self).as_data()
|
||||
data.update({
|
||||
'event': self.event,
|
||||
})
|
||||
if self.body is not None:
|
||||
data.update({
|
||||
'body': self.body.as_data(),
|
||||
})
|
||||
return data
|
||||
0
tests/debugger_protocol/messages/__init__.py
Normal file
0
tests/debugger_protocol/messages/__init__.py
Normal file
936
tests/debugger_protocol/messages/test_message.py
Normal file
936
tests/debugger_protocol/messages/test_message.py
Normal file
|
|
@ -0,0 +1,936 @@
|
|||
import unittest
|
||||
|
||||
from debugger_protocol.arg import FieldsNamespace, Field
|
||||
from debugger_protocol.messages import register
|
||||
from debugger_protocol.messages.message import (
|
||||
ProtocolMessage, Request, Response, Event)
|
||||
|
||||
|
||||
@register
|
||||
class DummyRequest(object):
|
||||
TYPE = 'request'
|
||||
TYPE_KEY = 'command'
|
||||
COMMAND = '...'
|
||||
|
||||
|
||||
@register
|
||||
class DummyResponse(object):
|
||||
TYPE = 'response'
|
||||
TYPE_KEY = 'command'
|
||||
COMMAND = '...'
|
||||
|
||||
|
||||
@register
|
||||
class DummyEvent(object):
|
||||
TYPE = 'event'
|
||||
TYPE_KEY = 'event'
|
||||
EVENT = '...'
|
||||
|
||||
|
||||
class FakeMsg(ProtocolMessage):
|
||||
|
||||
SEQ = 0
|
||||
|
||||
@classmethod
|
||||
def _next_reqid(cls):
|
||||
return cls.SEQ
|
||||
|
||||
|
||||
class ProtocolMessageTests(unittest.TestCase):
|
||||
|
||||
def test_from_data(self):
|
||||
data = {
|
||||
'type': 'event',
|
||||
'seq': 10,
|
||||
}
|
||||
msg = ProtocolMessage.from_data(**data)
|
||||
|
||||
self.assertEqual(msg.type, 'event')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
|
||||
def test_defaults(self): # no args
|
||||
class Spam(FakeMsg):
|
||||
SEQ = 10
|
||||
TYPE = 'event'
|
||||
|
||||
msg = Spam()
|
||||
|
||||
self.assertEqual(msg.type, 'event')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
|
||||
def test_all_args(self):
|
||||
msg = ProtocolMessage(10, type='event')
|
||||
|
||||
self.assertEqual(msg.type, 'event')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
|
||||
def test_coercion_seq(self):
|
||||
msg = ProtocolMessage('10', type='event')
|
||||
|
||||
self.assertEqual(msg.seq, 10)
|
||||
|
||||
def test_validation(self):
|
||||
# type
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
ProtocolMessage(type=None)
|
||||
with self.assertRaises(ValueError):
|
||||
ProtocolMessage(type='spam')
|
||||
|
||||
class Other(ProtocolMessage):
|
||||
TYPE = 'spam'
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
Other(type='event')
|
||||
|
||||
# seq
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
ProtocolMessage(None, type='event')
|
||||
with self.assertRaises(ValueError):
|
||||
ProtocolMessage(-1, type='event')
|
||||
|
||||
def test_readonly(self):
|
||||
msg = ProtocolMessage(10, type='event')
|
||||
|
||||
with self.assertRaises(AttributeError):
|
||||
msg.seq = 11
|
||||
with self.assertRaises(AttributeError):
|
||||
msg.type = 'event'
|
||||
with self.assertRaises(AttributeError):
|
||||
msg.spam = object()
|
||||
with self.assertRaises(AttributeError):
|
||||
del msg.seq
|
||||
|
||||
def test_repr(self):
|
||||
msg = ProtocolMessage(10, type='event')
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result, "ProtocolMessage(type='event', seq=10)")
|
||||
|
||||
def test_repr_subclass(self):
|
||||
class Eventish(ProtocolMessage):
|
||||
TYPE = 'event'
|
||||
|
||||
msg = Eventish(10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result, 'Eventish(seq=10)')
|
||||
|
||||
def test_as_data(self):
|
||||
msg = ProtocolMessage(10, type='event')
|
||||
data = msg.as_data()
|
||||
|
||||
self.assertEqual(data, {
|
||||
'type': 'event',
|
||||
'seq': 10,
|
||||
})
|
||||
|
||||
|
||||
class RequestTests(unittest.TestCase):
|
||||
|
||||
def test_from_data_without_arguments(self):
|
||||
data = {
|
||||
'type': 'request',
|
||||
'seq': 10,
|
||||
'command': 'spam',
|
||||
}
|
||||
msg = Request.from_data(**data)
|
||||
|
||||
self.assertEqual(msg.type, 'request')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.command, 'spam')
|
||||
self.assertIsNone(msg.arguments)
|
||||
|
||||
def test_from_data_with_arguments(self):
|
||||
class Spam(Request):
|
||||
class ARGUMENTS(FieldsNamespace):
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
data = {
|
||||
'type': 'request',
|
||||
'seq': 10,
|
||||
'command': 'spam',
|
||||
'arguments': {'a': 'b'},
|
||||
}
|
||||
#msg = Request.from_data(**data)
|
||||
msg = Spam.from_data(**data)
|
||||
|
||||
self.assertEqual(msg.type, 'request')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.command, 'spam')
|
||||
self.assertEqual(msg.arguments, {'a': 'b'})
|
||||
|
||||
def test_defaults(self):
|
||||
class Spam(Request, FakeMsg):
|
||||
SEQ = 10
|
||||
COMMAND = 'spam'
|
||||
|
||||
msg = Spam()
|
||||
|
||||
self.assertEqual(msg.type, 'request')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.command, 'spam')
|
||||
self.assertIsNone(msg.arguments)
|
||||
|
||||
def test_all_args(self):
|
||||
class Spam(Request):
|
||||
class ARGUMENTS(FieldsNamespace):
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
args = {'a': 'b'}
|
||||
msg = Spam(arguments=args, command='spam', seq=10)
|
||||
|
||||
self.assertEqual(msg.type, 'request')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.command, 'spam')
|
||||
self.assertEqual(msg.arguments, args)
|
||||
|
||||
def test_no_arguments_not_required(self):
|
||||
class Spam(Request):
|
||||
COMMAND = 'spam'
|
||||
ARGUMENTS = True
|
||||
ARGUMENTS_REQUIRED = False
|
||||
|
||||
msg = Spam()
|
||||
|
||||
self.assertIsNone(msg.arguments)
|
||||
|
||||
def test_no_args(self):
|
||||
with self.assertRaises(TypeError):
|
||||
Request()
|
||||
|
||||
def test_coercion_arguments(self):
|
||||
class Spam(Request):
|
||||
COMMAND = 'spam'
|
||||
class ARGUMENTS(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
args = [('a', 'b')]
|
||||
msg = Spam(args)
|
||||
|
||||
self.assertEqual(msg.arguments, {'a': 'b'})
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
Spam(command='spam', arguments=11)
|
||||
|
||||
def test_validation(self):
|
||||
with self.assertRaises(TypeError):
|
||||
Request()
|
||||
|
||||
# command
|
||||
|
||||
class Other1(Request):
|
||||
COMMAND = 'eggs'
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
# command doesn't match
|
||||
Other1(arguments=10, command='spam')
|
||||
|
||||
# arguments
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
# unexpected arguments
|
||||
Request(arguments=10, command='spam')
|
||||
|
||||
class Other2(Request):
|
||||
COMMAND = 'spam'
|
||||
ARGUMENTS = int
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
# missing arguments (implicitly required)
|
||||
Other2(command='eggs')
|
||||
|
||||
class Other3(Request):
|
||||
COMMAND = 'eggs'
|
||||
ARGUMENTS = int
|
||||
ARGUMENTS_REQUIRED = True
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
# missing arguments (explicitly required)
|
||||
Other2(command='eggs')
|
||||
|
||||
def test_repr_minimal(self):
|
||||
msg = Request(command='spam', seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result, "Request(command='spam', seq=10)")
|
||||
|
||||
def test_repr_full(self):
|
||||
msg = Request(command='spam', seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result, "Request(command='spam', seq=10)")
|
||||
|
||||
def test_repr_subclass_minimal(self):
|
||||
class SpamRequest(Request):
|
||||
COMMAND = 'spam'
|
||||
|
||||
msg = SpamRequest(seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result, "SpamRequest(seq=10)")
|
||||
|
||||
def test_repr_subclass_full(self):
|
||||
class SpamRequest(Request):
|
||||
COMMAND = 'spam'
|
||||
class ARGUMENTS(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
msg = SpamRequest(arguments={'a': 'b'}, seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result,
|
||||
"SpamRequest(arguments=ARGUMENTS(a='b'), seq=10)")
|
||||
|
||||
def test_as_data_minimal(self):
|
||||
msg = Request(command='spam', seq=10)
|
||||
data = msg.as_data()
|
||||
|
||||
self.assertEqual(data, {
|
||||
'type': 'request',
|
||||
'seq': 10,
|
||||
'command': 'spam',
|
||||
})
|
||||
|
||||
def test_as_data_full(self):
|
||||
class Spam(Request):
|
||||
COMMAND = 'spam'
|
||||
class ARGUMENTS(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
msg = Spam(arguments={'a': 'b'}, seq=10)
|
||||
data = msg.as_data()
|
||||
|
||||
self.assertEqual(data, {
|
||||
'type': 'request',
|
||||
'seq': 10,
|
||||
'command': 'spam',
|
||||
'arguments': {'a': 'b'},
|
||||
})
|
||||
|
||||
|
||||
class ResponseTests(unittest.TestCase):
|
||||
|
||||
def test_from_data_without_body(self):
|
||||
data = {
|
||||
'type': 'response',
|
||||
'seq': 10,
|
||||
'command': 'spam',
|
||||
'request_seq': 9,
|
||||
'success': True,
|
||||
}
|
||||
msg = Response.from_data(**data)
|
||||
|
||||
self.assertEqual(msg.type, 'response')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.command, 'spam')
|
||||
self.assertEqual(msg.request_seq, 9)
|
||||
self.assertTrue(msg.success)
|
||||
self.assertIsNone(msg.body)
|
||||
self.assertIsNone(msg.message)
|
||||
|
||||
def test_from_data_with_body(self):
|
||||
class Spam(Response):
|
||||
class BODY(FieldsNamespace):
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
data = {
|
||||
'type': 'response',
|
||||
'seq': 10,
|
||||
'command': 'spam',
|
||||
'request_seq': 9,
|
||||
'success': True,
|
||||
'body': {'a': 'b'},
|
||||
}
|
||||
msg = Spam.from_data(**data)
|
||||
|
||||
self.assertEqual(msg.type, 'response')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.command, 'spam')
|
||||
self.assertEqual(msg.request_seq, 9)
|
||||
self.assertTrue(msg.success)
|
||||
self.assertEqual(msg.body, {'a': 'b'})
|
||||
self.assertIsNone(msg.message)
|
||||
|
||||
def test_from_data_error_without_body(self):
|
||||
data = {
|
||||
'type': 'response',
|
||||
'seq': 10,
|
||||
'command': 'spam',
|
||||
'request_seq': 9,
|
||||
'success': False,
|
||||
'message': 'oops!',
|
||||
}
|
||||
msg = Response.from_data(**data)
|
||||
|
||||
self.assertEqual(msg.type, 'response')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.command, 'spam')
|
||||
self.assertEqual(msg.request_seq, 9)
|
||||
self.assertFalse(msg.success)
|
||||
self.assertIsNone(msg.body)
|
||||
self.assertEqual(msg.message, 'oops!')
|
||||
|
||||
def test_from_data_error_with_body(self):
|
||||
class Spam(Response):
|
||||
class ERROR_BODY(FieldsNamespace):
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
data = {
|
||||
'type': 'response',
|
||||
'seq': 10,
|
||||
'command': 'spam',
|
||||
'request_seq': 9,
|
||||
'success': False,
|
||||
'message': 'oops!',
|
||||
'body': {'a': 'b'},
|
||||
}
|
||||
msg = Spam.from_data(**data)
|
||||
|
||||
self.assertEqual(msg.type, 'response')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.command, 'spam')
|
||||
self.assertEqual(msg.request_seq, 9)
|
||||
self.assertFalse(msg.success)
|
||||
self.assertEqual(msg.body, {'a': 'b'})
|
||||
self.assertEqual(msg.message, 'oops!')
|
||||
|
||||
def test_defaults(self):
|
||||
class Spam(Response, FakeMsg):
|
||||
SEQ = 10
|
||||
COMMAND = 'spam'
|
||||
|
||||
msg = Spam('9')
|
||||
|
||||
self.assertEqual(msg.type, 'response')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.request_seq, 9)
|
||||
self.assertEqual(msg.command, 'spam')
|
||||
self.assertTrue(msg.success)
|
||||
self.assertIsNone(msg.body)
|
||||
self.assertIsNone(msg.message)
|
||||
|
||||
def test_all_args_not_error(self):
|
||||
class Spam(Response):
|
||||
class BODY(FieldsNamespace):
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
msg = Spam('9', command='spam', success=True, body={'a': 'b'},
|
||||
seq=10, type='response')
|
||||
|
||||
self.assertEqual(msg.type, 'response')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.request_seq, 9)
|
||||
self.assertEqual(msg.command, 'spam')
|
||||
self.assertTrue(msg.success)
|
||||
self.assertEqual(msg.body, {'a': 'b'})
|
||||
self.assertIsNone(msg.message)
|
||||
|
||||
def test_all_args_error(self):
|
||||
class Spam(Response):
|
||||
COMMAND = 'spam'
|
||||
class ERROR_BODY(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
msg = Spam('9', success=False, message='oops!', body={'a': 'b'},
|
||||
seq=10, type='response')
|
||||
|
||||
self.assertEqual(msg.type, 'response')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.command, 'spam')
|
||||
self.assertEqual(msg.request_seq, 9)
|
||||
self.assertFalse(msg.success)
|
||||
self.assertEqual(msg.body, Spam.ERROR_BODY(a='b'))
|
||||
self.assertEqual(msg.message, 'oops!')
|
||||
|
||||
def test_no_body_not_required(self):
|
||||
class Spam(Response):
|
||||
COMMAND = 'spam'
|
||||
BODY = True
|
||||
BODY_REQUIRED = False
|
||||
|
||||
msg = Spam('9')
|
||||
|
||||
self.assertIsNone(msg.body)
|
||||
|
||||
def test_no_error_body_not_required(self):
|
||||
class Spam(Response):
|
||||
COMMAND = 'spam'
|
||||
ERROR_BODY = True
|
||||
ERROR_BODY_REQUIRED = False
|
||||
|
||||
msg = Spam('9', success=False, message='oops!')
|
||||
|
||||
self.assertIsNone(msg.body)
|
||||
|
||||
def test_no_args(self):
|
||||
with self.assertRaises(TypeError):
|
||||
Response()
|
||||
|
||||
def test_coercion_request_seq(self):
|
||||
msg = Response('9', command='spam')
|
||||
|
||||
self.assertEqual(msg.request_seq, 9)
|
||||
|
||||
def test_coercion_success(self):
|
||||
msg1 = Response(9, success=1, command='spam')
|
||||
msg2 = Response(9, success=None, command='spam', message='oops!')
|
||||
|
||||
self.assertIs(msg1.success, True)
|
||||
self.assertIs(msg2.success, False)
|
||||
|
||||
def test_coercion_body(self):
|
||||
class Spam(Response):
|
||||
COMMAND = 'spam'
|
||||
class BODY(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
body = [('a', 'b')]
|
||||
msg = Spam(9, body=body)
|
||||
|
||||
self.assertEqual(msg.body, {'a': 'b'})
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
Spam(9, command='spam', body=11)
|
||||
|
||||
def test_coercion_error_body(self):
|
||||
class Spam(Response):
|
||||
COMMAND = 'spam'
|
||||
class ERROR_BODY(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
body = [('a', 'b')]
|
||||
msg = Spam(9, body=body, success=False, message='oops!')
|
||||
|
||||
self.assertEqual(msg.body, {'a': 'b'})
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
Spam(9, command='spam', success=False, message='oops!', body=11)
|
||||
|
||||
def test_validation(self):
|
||||
# request_seq
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
# missing
|
||||
Response(None, command='spam')
|
||||
with self.assertRaises(TypeError):
|
||||
# missing
|
||||
Response('', command='spam')
|
||||
with self.assertRaises(TypeError):
|
||||
# couldn't convert to int
|
||||
Response(object(), command='spam')
|
||||
with self.assertRaises(ValueError):
|
||||
# not non-negative
|
||||
Response(-1, command='spam')
|
||||
|
||||
# command
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
# missing
|
||||
Response(9, command=None)
|
||||
with self.assertRaises(TypeError):
|
||||
# missing
|
||||
Response(9, command='')
|
||||
|
||||
class Other1(Response):
|
||||
COMMAND = 'eggs'
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
# does not match
|
||||
Other1(9, command='spam')
|
||||
|
||||
# body
|
||||
|
||||
class Other2(Response):
|
||||
class BODY(FieldsNamespace):
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
ERROR_BODY = BODY
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
# unexpected
|
||||
Response(9, command='spam', body=11)
|
||||
with self.assertRaises(TypeError):
|
||||
# missing (implicitly required)
|
||||
Other2(9, command='spam')
|
||||
with self.assertRaises(TypeError):
|
||||
# missing (explicitly required)
|
||||
Other2.BODY_REQUIRED = True
|
||||
Other2(9, command='spam')
|
||||
with self.assertRaises(ValueError):
|
||||
# unexpected (error)
|
||||
Response(9, command='spam', body=11, success=False, message=':(')
|
||||
with self.assertRaises(TypeError):
|
||||
# missing (error) (implicitly required)
|
||||
Other2(9, command='spam', success=False, message=':(')
|
||||
with self.assertRaises(TypeError):
|
||||
# missing (error) (explicitly required)
|
||||
Other2.ERROR_BODY_REQUIRED = True
|
||||
Other2(9, command='spam', success=False, message=':(')
|
||||
|
||||
# message
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
# missing
|
||||
Response(9, command='spam', success=False)
|
||||
|
||||
def test_repr_minimal(self):
|
||||
msg = Response(9, command='spam', seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result,
|
||||
"Response(command='spam', request_seq=9, success=True, seq=10)") # noqa
|
||||
|
||||
def test_repr_full(self):
|
||||
msg = Response(9, command='spam', seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result,
|
||||
"Response(command='spam', request_seq=9, success=True, seq=10)") # noqa
|
||||
|
||||
def test_repr_error_minimal(self):
|
||||
msg = Response(9, command='spam', success=False, message='oops!',
|
||||
seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result,
|
||||
"Response(command='spam', request_seq=9, success=False, message='oops!', seq=10)") # noqa
|
||||
|
||||
def test_repr_error_full(self):
|
||||
msg = Response(9, command='spam', success=False, message='oops!',
|
||||
seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result,
|
||||
"Response(command='spam', request_seq=9, success=False, message='oops!', seq=10)") # noqa
|
||||
|
||||
def test_repr_subclass_minimal(self):
|
||||
class SpamResponse(Response):
|
||||
COMMAND = 'spam'
|
||||
|
||||
msg = SpamResponse(9, seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result,
|
||||
"SpamResponse(request_seq=9, success=True, seq=10)")
|
||||
|
||||
def test_repr_subclass_full(self):
|
||||
class SpamResponse(Response):
|
||||
COMMAND = 'spam'
|
||||
class BODY(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
msg = SpamResponse(9, body={'a': 'b'}, seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result,
|
||||
"SpamResponse(request_seq=9, success=True, body=BODY(a='b'), seq=10)") # noqa
|
||||
|
||||
def test_repr_subclass_error_minimal(self):
|
||||
class SpamResponse(Response):
|
||||
COMMAND = 'spam'
|
||||
|
||||
msg = SpamResponse(9, success=False, message='oops!', seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result,
|
||||
"SpamResponse(request_seq=9, success=False, message='oops!', seq=10)") # noqa
|
||||
|
||||
def test_repr_subclass_error_full(self):
|
||||
class SpamResponse(Response):
|
||||
COMMAND = 'spam'
|
||||
class ERROR_BODY(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
msg = SpamResponse(9, success=False, message='oops!', body={'a': 'b'},
|
||||
seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result,
|
||||
"SpamResponse(request_seq=9, success=False, message='oops!', body=ERROR_BODY(a='b'), seq=10)") # noqa
|
||||
|
||||
def test_as_data_minimal(self):
|
||||
msg = Response(9, command='spam', seq=10)
|
||||
data = msg.as_data()
|
||||
|
||||
self.assertEqual(data, {
|
||||
'type': 'response',
|
||||
'seq': 10,
|
||||
'request_seq': 9,
|
||||
'command': 'spam',
|
||||
'success': True,
|
||||
})
|
||||
|
||||
def test_as_data_full(self):
|
||||
class Spam(Response):
|
||||
COMMAND = 'spam'
|
||||
class BODY(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
msg = Spam(9, body={'a': 'b'}, seq=10)
|
||||
data = msg.as_data()
|
||||
|
||||
self.assertEqual(data, {
|
||||
'type': 'response',
|
||||
'seq': 10,
|
||||
'request_seq': 9,
|
||||
'command': 'spam',
|
||||
'success': True,
|
||||
'body': {'a': 'b'},
|
||||
})
|
||||
|
||||
def test_as_data_error_minimal(self):
|
||||
msg = Response(9, command='spam', success=False, message='oops!',
|
||||
seq=10)
|
||||
data = msg.as_data()
|
||||
|
||||
self.assertEqual(data, {
|
||||
'type': 'response',
|
||||
'seq': 10,
|
||||
'request_seq': 9,
|
||||
'command': 'spam',
|
||||
'success': False,
|
||||
'message': 'oops!',
|
||||
})
|
||||
|
||||
def test_as_data_error_full(self):
|
||||
class Spam(Response):
|
||||
COMMAND = 'spam'
|
||||
class ERROR_BODY(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
msg = Spam(9, success=False, body={'a': 'b'}, message='oops!', seq=10)
|
||||
data = msg.as_data()
|
||||
|
||||
self.assertEqual(data, {
|
||||
'type': 'response',
|
||||
'seq': 10,
|
||||
'request_seq': 9,
|
||||
'command': 'spam',
|
||||
'success': False,
|
||||
'message': 'oops!',
|
||||
'body': {'a': 'b'},
|
||||
})
|
||||
|
||||
|
||||
class EventTests(unittest.TestCase):
|
||||
|
||||
def test_from_data_without_body(self):
|
||||
data = {
|
||||
'type': 'event',
|
||||
'seq': 10,
|
||||
'event': 'spam',
|
||||
}
|
||||
msg = Event.from_data(**data)
|
||||
|
||||
self.assertEqual(msg.type, 'event')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.event, 'spam')
|
||||
self.assertIsNone(msg.body)
|
||||
|
||||
def test_from_data_with_body(self):
|
||||
class Spam(Event):
|
||||
class BODY(FieldsNamespace):
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
data = {
|
||||
'type': 'event',
|
||||
'seq': 10,
|
||||
'event': 'spam',
|
||||
'body': {'a': 'b'},
|
||||
}
|
||||
msg = Spam.from_data(**data)
|
||||
|
||||
self.assertEqual(msg.type, 'event')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.event, 'spam')
|
||||
self.assertEqual(msg.body, {'a': 'b'})
|
||||
|
||||
def test_defaults(self): # no args
|
||||
class Spam(Event, FakeMsg):
|
||||
SEQ = 10
|
||||
EVENT = 'spam'
|
||||
|
||||
msg = Spam()
|
||||
|
||||
self.assertEqual(msg.type, 'event')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.event, 'spam')
|
||||
self.assertIsNone(msg.body)
|
||||
|
||||
def test_all_args(self):
|
||||
class Spam(Event):
|
||||
class BODY(FieldsNamespace):
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
msg = Spam(event='spam', body={'a': 'b'}, seq=10, type='event')
|
||||
|
||||
self.assertEqual(msg.type, 'event')
|
||||
self.assertEqual(msg.seq, 10)
|
||||
self.assertEqual(msg.event, 'spam')
|
||||
self.assertEqual(msg.body, {'a': 'b'})
|
||||
|
||||
def test_no_body_not_required(self):
|
||||
class Spam(Event):
|
||||
EVENT = 'spam'
|
||||
BODY = True
|
||||
BODY_REQUIRED = False
|
||||
|
||||
msg = Spam()
|
||||
|
||||
self.assertIsNone(msg.body)
|
||||
|
||||
def test_no_args(self):
|
||||
with self.assertRaises(TypeError):
|
||||
Event()
|
||||
|
||||
def test_coercion_body(self):
|
||||
class Spam(Event):
|
||||
EVENT = 'spam'
|
||||
class BODY(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
body = [('a', 'b')]
|
||||
msg = Spam(body=body)
|
||||
|
||||
self.assertEqual(msg.body, {'a': 'b'})
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
Spam(event='spam', body=11)
|
||||
|
||||
def test_validation(self):
|
||||
# event
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
# missing
|
||||
Event(event=None)
|
||||
with self.assertRaises(TypeError):
|
||||
# missing
|
||||
Event(event='')
|
||||
|
||||
class Other1(Event):
|
||||
EVENT = 'eggs'
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
# does not match
|
||||
Other1(event='spam')
|
||||
|
||||
# body
|
||||
|
||||
class Other2(Event):
|
||||
class BODY(FieldsNamespace):
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
# unexpected
|
||||
Event(event='spam', body=11)
|
||||
with self.assertRaises(TypeError):
|
||||
# missing (implicitly required)
|
||||
Other2(9, command='spam')
|
||||
with self.assertRaises(TypeError):
|
||||
# missing (explicitly required)
|
||||
Other2.BODY_REQUIRED = True
|
||||
Other2(9, command='spam')
|
||||
|
||||
def test_repr_minimal(self):
|
||||
msg = Event(event='spam', seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result, "Event(event='spam', seq=10)")
|
||||
|
||||
def test_repr_full(self):
|
||||
msg = Event(event='spam', seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result, "Event(event='spam', seq=10)")
|
||||
|
||||
def test_repr_subclass_minimal(self):
|
||||
class SpamEvent(Event):
|
||||
EVENT = 'spam'
|
||||
|
||||
msg = SpamEvent(seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result, 'SpamEvent(seq=10)')
|
||||
|
||||
def test_repr_subclass_full(self):
|
||||
class SpamEvent(Event):
|
||||
EVENT = 'spam'
|
||||
class BODY(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
msg = SpamEvent(body={'a': 'b'}, seq=10)
|
||||
result = repr(msg)
|
||||
|
||||
self.assertEqual(result, "SpamEvent(body=BODY(a='b'), seq=10)")
|
||||
|
||||
def test_as_data_minimal(self):
|
||||
msg = Event(event='spam', seq=10)
|
||||
data = msg.as_data()
|
||||
|
||||
self.assertEqual(data, {
|
||||
'type': 'event',
|
||||
'seq': 10,
|
||||
'event': 'spam',
|
||||
})
|
||||
|
||||
def test_as_data_full(self):
|
||||
class Spam(Event):
|
||||
EVENT = 'spam'
|
||||
class BODY(FieldsNamespace): # noqa
|
||||
FIELDS = [
|
||||
Field('a'),
|
||||
]
|
||||
|
||||
msg = Spam(body={'a': 'b'}, seq=10)
|
||||
data = msg.as_data()
|
||||
|
||||
self.assertEqual(data, {
|
||||
'type': 'event',
|
||||
'seq': 10,
|
||||
'event': 'spam',
|
||||
'body': {'a': 'b'},
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue