Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_proto_fill_from_dict__func_generator(self):
def number_gen():
yield 1
yield 2
yield 3
utp.proto_fill_from_dict(self.msg, {'list_number32': number_gen()})
self.assertEqual(self.msg.list_number32, [1,2,3])
def test_proto_fill_from_dict__generator(self):
utp.proto_fill_from_dict(self.msg, {'list_number32': (x for x in [1,2,3])})
self.assertEqual(self.msg.list_number32, [1,2,3])
def test_proto_fill_from_dict__dict_filter(self):
utp.proto_fill_from_dict(self.msg, {'messages': filter(lambda x: True, [{'text': 'one'}, {'text': 'two'}])})
self.assertEqual(len(self.msg.messages), 2)
self.assertEqual(self.msg.messages[0].text, 'one')
self.assertEqual(self.msg.messages[1].text, 'two')
def test_proto_from_dict_merge_dict(self):
self.msg.messages.add(text='one')
self.msg.messages.add(text='two')
utp.proto_fill_from_dict(self.msg, {'messages': [{'text': 'three'}]}, clear=False)
self.assertEqual(len(self.msg.messages), 1)
self.assertEqual(self.msg.messages[0].text, 'three')
def test_proto_from_dict_to_dict(self):
DATA = {'buffers': [
{'data': b'some data', 'flags': [{'flag': True}, {'flag': False}, {'flag': False}]},
{'data': b'\x01\x02\x03\x04', 'flags': [{'flag': False}, {'flag': True}, {'flag': True}]}
],
'list_number32': [4,16,64,256,1024,4096,16384,65536,262144,1048576,4194304,16777216,67108864,268435456,1073741824],
'list_number64': [4,64,1024,16384,262144,1125899906842624,18014398509481984,288230376151711744,4611686018427387904],
'messages': [{'text': 'test string'}, {'text': 'another one'}, {'text': 'third'}],
'number32': 16777216,
'number64': 72057594037927936
}
utp.proto_fill_from_dict(self.msg, DATA)
RESULT = utp.proto_to_dict(self.msg)
self.assertEqual(DATA, RESULT)
def test_proto_from_dict__list_insteadof_dict(self):
with self.assertRaises(TypeError):
utp.proto_fill_from_dict(self.msg, {'messages': [1,2,3]})
def test_proto_fill_from_dict__dict_generator(self):
utp.proto_fill_from_dict(self.msg, {'messages': (x for x in [{'text': 'one'}, {'text': 'two'}])})
self.assertEqual(len(self.msg.messages), 2)
self.assertEqual(self.msg.messages[0].text, 'one')
self.assertEqual(self.msg.messages[1].text, 'two')
def _send(self, emsg, data={}, proto=None, jobid=None):
if not isinstance(data, dict):
raise ValueError("data kwarg can only be a dict")
if proto is None:
proto = find_proto(emsg)
if not issubclass(proto, google.protobuf.message.Message):
raise ValueError("Unable to find proto for emsg, or proto kwarg is invalid")
message = proto()
proto_fill_from_dict(message, data)
header = GCMsgHdrProto(emsg)
if jobid is not None:
header.proto.job_id_source = jobid
if self.verbose_debug:
str_message = ''
str_header = str(header)
str_body = str(message)
if str_header:
str_message += "-- header ---------\n%s\n" % str_header
if str_body:
str_message += "-- message --------\n%s\n" % str_body
def _send(self, emsg, data={}, proto=None, jobid=None):
if not isinstance(data, dict):
raise ValueError("data kwarg can only be a dict")
if proto is None:
proto = find_proto(emsg)
if proto is None or not issubclass(proto, google.protobuf.message.Message):
raise ValueError("Unable to find proto for emsg, or proto kwarg is invalid")
message = proto()
proto_fill_from_dict(message, data)
header = GCMsgHdrProto(emsg)
if jobid is not None:
header.proto.job_id_source = jobid
if self.verbose_debug:
str_message = ''
str_header = str(header)
str_body = str(message)
if str_header:
str_message += "-- header ---------\n%s\n" % str_header
if str_body:
str_message += "-- message --------\n%s\n" % str_body