本文整理汇总了Python中pyamf.encode函数的典型用法代码示例。如果您正苦于以下问题：Python encode函数的具体用法？Python encode怎么用？Python encode使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了encode函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: _encode
def _encode(self, mod):
element = mod.Element('foo', bar='baz')
return (
pyamf.encode(element, encoding=pyamf.AMF0).getvalue(),
pyamf.encode(element, encoding=pyamf.AMF3).getvalue()
)
开发者ID:cardmagic，项目名称:PyAMF，代码行数:7，代码来源:test_xml.py
示例2: test_amf3_encode
def test_amf3_encode(self):
x = decimal.Decimal('1.23456463452345')
self.assertEqual(
pyamf.encode(x, encoding=pyamf.AMF3, strict=False).getvalue(),
'\x05?\xf3\xc0\xc6\xd8\xa18\xfa'
)
with self.assertRaises(pyamf.EncodeError):
pyamf.encode(x, encoding=pyamf.AMF3, strict=True)
开发者ID:nervatura，项目名称:nerva2py，代码行数:10，代码来源:test_decimal.py
示例3: test_CommandMessage
def test_CommandMessage(self):
m = messages.CommandMessage()
bytes = pyamf.encode(m.getSmallMessage(), encoding=pyamf.AMF3).getvalue()
self.assertEquals(bytes, '\n\x07\x07DSC\x00\x01\x01\x01\x04\xce\x10')
m = messages.CommandMessage(operation='foo.bar')
bytes = pyamf.encode(m.getSmallMessage(), encoding=pyamf.AMF3).getvalue()
self.assertEquals(bytes, '\n\x07\x07DSC\x00\x01\x01\x01\x06\x0ffoo.bar')
开发者ID:hydralabs，项目名称:plasma，代码行数:11，代码来源:test_small_messages.py
示例4: test_default_encoding
def test_default_encoding(self):
pyamf.DEFAULT_ENCODING = pyamf.AMF3
x = pyamf.encode('foo').getvalue()
self.assertEqual(x, '\x06\x07foo')
pyamf.DEFAULT_ENCODING = pyamf.AMF0
x = pyamf.encode('foo').getvalue()
self.assertEqual(x, '\x02\x00\x03foo')
开发者ID:0xmilk，项目名称:appscale，代码行数:12，代码来源:test_basic.py
示例5: save
def save(self, out, format):
"""
"""
#
# Serialize
#
if format == "WKT":
if "wkt" in self.content["crs"]:
out.write(self.content["crs"]["wkt"])
else:
out.write(_sref_4326().ExportToWkt())
return
if format in ("GeoJSON", "GeoBSON", "GeoAMF"):
content = self.content
if "wkt" in content["crs"]:
content["crs"] = {"type": "link", "properties": {"href": "0.wkt", "type": "ogcwkt"}}
else:
del content["crs"]
elif format in ("ArcJSON", "ArcBSON", "ArcAMF"):
content = reserialize_to_arc(self.content, format == "ArcAMF")
else:
raise KnownUnknown(
'Vector response only saves .geojson, .arcjson, .geobson, .arcbson, .geoamf, .arcamf and .wkt tiles, not "%s"'
% format
)
#
# Encode
#
if format in ("GeoJSON", "ArcJSON"):
indent = self.verbose and 2 or None
encoded = JSONEncoder(indent=indent).iterencode(content)
float_pat = compile(r"^-?\d+\.\d+$")
for atom in encoded:
if float_pat.match(atom):
out.write("%.6f" % float(atom))
else:
out.write(atom)
elif format in ("GeoBSON", "ArcBSON"):
import bson
encoded = bson.dumps(content)
out.write(encoded)
elif format in ("GeoAMF", "ArcAMF"):
import pyamf
for class_name in pyamf_classes.items():
pyamf.register_class(*class_name)
encoded = pyamf.encode(content, 0).read()
out.write(encoded)
开发者ID:netconstructor，项目名称:TileStache，代码行数:60，代码来源:__init__.py
示例6: test_command
def test_command(self):
bytes = ('\n\x07\x07DSC\x88\x02\n\x0b\x01\tDSId\x06IEE0D161D-C11D-'
'25CB-8DBE-3B77B54B55D9\x01\x0c!\xc0\xdf\xb7|\xd6\xee$1s\x152f'
'\xe11\xa8f\x01\x06\x01\x01\x04\x02')
self.buffer.write(bytes)
self.buffer.seek(0)
msg = self.decoder.readElement()
self.assertTrue(isinstance(msg, small.CommandMessageExt))
self.assertEquals(msg.body, None)
self.assertEquals(msg.destination, None)
self.assertEquals(msg.timeToLive, None)
self.assertEquals(msg.timestamp, None)
self.assertEquals(msg.headers, {
'DSId': u'EE0D161D-C11D-25CB-8DBE-3B77B54B55D9'
})
self.assertEquals(msg.clientId, None)
self.assertEquals(msg.messageId,
uuid.UUID('c0dfb77c-d6ee-2431-7315-3266e131a866'))
self.assertEquals(msg.correlationId, u'')
self.assertEquals(self.buffer.remaining(), 0)
# now encode the msg to check that encoding is byte for byte the same
buffer = pyamf.encode(msg, encoding=pyamf.AMF3).getvalue()
self.assertEquals(buffer, bytes)
开发者ID:hydralabs，项目名称:plasma，代码行数:29，代码来源:test_small_messages.py
示例7: test_encode
def test_encode(self):
from django.utils.translation import ugettext_lazy
self.assertEqual(
pyamf.encode(ugettext_lazy('Hello')).getvalue(),
'\x06\x0bHello'
)
开发者ID:LiiiQin，项目名称:test，代码行数:7，代码来源:test_django.py
示例8: test_acknowledge
def test_acknowledge(self):
bytes = ('\n\x07\x07DSK\xa8\x03\n\x0b\x01%DSMessagingVersion\x05?\xf0'
'\x00\x00\x00\x00\x00\x00\tDSId\x06IEE0D161D-C11D-25CB-8DBE-3B77B'
'54B55D9\x01\x05Br3&m\x85\x10\x00\x0c!\xee\r\x16\x1d\xc1(&[\xc9'
'\x80RK\x9bE\xc6\xc4\x0c!\xee\r\x16\x1d\xc1=\x8e\xa3\xe0\x10\xef'
'\xad;\xe5\xc5j\x02\x0c!S\x84\x83\xdb\xa9\xc8\xcaM`\x952f\xdbQ'
'\xc9<\x00')
self.buffer.write(bytes)
self.buffer.seek(0)
msg = self.decoder.readElement()
self.assertTrue(isinstance(msg, messaging.AcknowledgeMessageExt))
self.assertEqual(msg.body, None)
self.assertEqual(msg.destination, None)
self.assertEqual(msg.timeToLive, None)
self.assertEqual(msg.timestamp, datetime.datetime(2009, 8, 19, 11, 24, 43, 985000))
self.assertEqual(msg.headers, {
'DSMessagingVersion': 1.0,
'DSId': u'EE0D161D-C11D-25CB-8DBE-3B77B54B55D9'
})
self.assertEqual(msg.clientId, uuid.UUID('ee0d161d-c128-265b-c980-524b9b45c6c4'))
self.assertEqual(msg.messageId, uuid.UUID('ee0d161d-c13d-8ea3-e010-efad3be5c56a'))
self.assertEqual(msg.correlationId, uuid.UUID('538483db-a9c8-ca4d-6095-3266db51c93c'))
self.assertEqual(self.buffer.remaining(), 0)
# now encode the msg to check that encoding is byte for byte the same
buffer = pyamf.encode(msg, encoding=pyamf.AMF3).getvalue()
self.assertEqual(buffer, bytes)
开发者ID:meemeer，项目名称:kayframework-pyamf，代码行数:31，代码来源:test_flex_messaging.py
示例9: test_post_process
def test_post_process(self):
"""
Ensure that postprocessing happens when data has been decoded.
"""
self.executed = False
post_procs = pyamf.POST_DECODE_PROCESSORS[:]
def restore_post_procs():
pyamf.POST_DECODE_PROCESSORS = post_procs
self.addCleanup(restore_post_procs)
pyamf.POST_DECODE_PROCESSORS = []
def postprocess(payload, context):
self.assertEqual(payload, u'foo')
self.assertEqual(context, {})
self.executed = True
return payload
pyamf.add_post_decode_processor(postprocess)
# setup complete
bytes = pyamf.encode(u'foo', encoding=pyamf.AMF0).getvalue()
self.decoder.send(bytes)
ret = self.decoder.next()
self.assertTrue(self.executed)
self.assertEqual(ret, u'foo')
开发者ID:LiiiQin，项目名称:test，代码行数:32，代码来源:test_amf0.py
示例10: test_dynamic_property_referenced_object
def test_dynamic_property_referenced_object(self):
class Author(db.Model):
name = db.StringProperty()
class Novel(db.Model):
title = db.StringProperty()
author = db.ReferenceProperty(Author)
try:
a = Author(name='Jane Austen')
a.put()
b = Novel(title='Sense and Sensibility', author=a)
b.put()
x = Novel.all().filter('title = ', 'Sense and Sensibility').get()
foo = [1, 2, 3]
x.author.bar = foo
k = str(x.key())
l = str(x.author.key())
stream = pyamf.encode(x)
self.assertEquals(stream.getvalue(), '\x03\x00\x04_key\x02%s%s'
'\x00\x06author\x03\x00\x04_key\x02%s%s\x00\x04name\x02\x00'
'\x0bJane Austen\x00\x03bar\n\x00\x00\x00\x03\x00?\xf0\x00'
'\x00\x00\x00\x00\x00\[email protected]\x00\x00\x00\x00\x00\x00\x00\[email protected]'
'\x08\x00\x00\x00\x00\x00\x00\x00\x00\t\x00\x05title\x02\x00'
'\x15Sense and Sensibility\x00\x00\t' % (
struct.pack('>H', len(k)), k,
struct.pack('>H', len(l)), l))
finally:
a.delete()
b.delete()
开发者ID:danielor，项目名称:TowerSaint，代码行数:35，代码来源:test_google.py
示例11: test_referenced_encode
def test_referenced_encode(self):
f = self.ParentReference()
f.name = "foo"
b = self.ChildReference()
b.name = "bar"
f.save()
b.foo = f
b.save()
f.bar = b
f.save()
self.assertEquals(f.id, 2)
foo = self.ParentReference.objects.select_related().get(id=2)
# ensure the referenced attribute resolves
foo.bar.foo
self.assertEquals(
pyamf.encode(foo).getvalue(),
"\x03\x00\x02id\x00"
"@\x00\x00\x00\x00\x00\x00\x00\x00\x04name\x02\x00\x03foo\x00"
"\x03bar\x03\x00\x02id\[email protected]\x00\x00\x00\x00\x00\x00\x00\x00\x04na"
"me\x02\x00\x03bar\x00\x03foo\x07\x00\x00\x00\x00\t\x00\x00\t",
)
开发者ID:wogooo，项目名称:TowerSaint，代码行数:26，代码来源:test_django.py
示例12: test_CommandMessage
def test_CommandMessage(self):
m = messaging.CommandMessage(operation='foo.bar')
self.assertEqual(pyamf.encode(m).getvalue(),
'\n\x81\x1bMflex.messaging.messages.CommandMessage\x1bcorrelationId'
'\tbody\x11clientId\x17destination\x0fheaders\x13messageId\x13'
'timestamp\x15timeToLive\x13operation\x01\x01\x01\x01\n\x0b\x01\x01'
'\x01\x01\x01\x06\x0ffoo.bar\x01')
开发者ID:meemeer，项目名称:kayframework-pyamf，代码行数:8，代码来源:test_flex_messaging.py
示例13: test_amf0_encode
def test_amf0_encode(self):
x = decimal.Decimal("1.23456463452345")
self.assertEqual(
pyamf.encode(x, encoding=pyamf.AMF0, strict=False).getvalue(), "\x00?\xf3\xc0\xc6\xd8\xa18\xfa"
)
self.assertRaises(pyamf.EncodeError, pyamf.encode, x, encoding=pyamf.AMF0, strict=True)
开发者ID:Armedite，项目名称:xbmc-catchuptv-au，代码行数:8，代码来源:test_decimal.py
示例14: test_Query_type
def test_Query_type(self):
"""
L{db.Query} instances get converted to lists ..
"""
q = PetModel.all()
self.assertTrue(isinstance(q, db.Query))
self.assertEquals(pyamf.encode(q).getvalue(), "\n\x00\x00\x00\x00")
开发者ID:jrolfs，项目名称:google-calendar-amf，代码行数:8，代码来源:test_google.py
示例15: test_amf3_encode
def test_amf3_encode(self):
x = sets.ImmutableSet(["1", "2", "3"])
self.assertTrue(
check_buffer(
pyamf.encode(x, encoding=pyamf.AMF3).getvalue(), ("\t\x07\x01", ("\x06\x031", "\x06\x033", "\x06\x032"))
)
)
开发者ID:nmatthews，项目名称:BlueCop-XBMC-Plugins，代码行数:8，代码来源:test_sets.py
示例16: test_RemotingMessage
def test_RemotingMessage(self):
m = messaging.RemotingMessage(source='foo.bar')
self.assertEqual(pyamf.encode(m).getvalue(),
'\n\x81\x1bOflex.messaging.messages.RemotingMessage\x15timeToLive'
'\x13timestamp\x13messageId\x0fheaders\x17destination\x11clientId'
'\tbody\x13operation\rsource\x01\x01\x01\n\x0b\x01\x01\x01\x01\x01'
'\x01\x06\x0ffoo.bar\x01')
开发者ID:meemeer，项目名称:kayframework-pyamf，代码行数:8，代码来源:test_flex_messaging.py
示例17: save
def save(self, out, format):
"""
"""
#
# Serialize
#
if format == 'WKT':
if 'wkt' in self.content['crs']:
out.write(self.content['crs']['wkt'])
else:
out.write(_sref_4326().ExportToWkt())
return
if format in ('GeoJSON', 'GeoBSON', 'GeoAMF'):
content = self.content
if 'wkt' in content['crs']:
content['crs'] = {'type': 'link', 'properties': {'href': '0.wkt', 'type': 'ogcwkt'}}
else:
del content['crs']
elif format in ('ArcJSON', 'ArcBSON', 'ArcAMF'):
content = reserialize_to_arc(self.content, format == 'ArcAMF')
else:
raise KnownUnknown('Vector response only saves .geojson, .arcjson, .geobson, .arcbson, .geoamf, .arcamf and .wkt tiles, not "%s"' % format)
#
# Encode
#
if format in ('GeoJSON', 'ArcJSON'):
indent = self.verbose and 2 or None
encoded = JSONEncoder(indent=indent).iterencode(content)
float_pat = compile(r'^-?\d+\.\d+$')
for atom in encoded:
if float_pat.match(atom):
piece = ('%%.%if' % self.precision) % float(atom)
else:
piece = atom
out.write(piece.encode('utf8'))
elif format in ('GeoBSON', 'ArcBSON'):
import bson
encoded = bson.dumps(content)
out.write(encoded)
elif format in ('GeoAMF', 'ArcAMF'):
import pyamf
for class_name in pyamf_classes.items():
pyamf.register_class(*class_name)
encoded = pyamf.encode(content, 0).read()
out.write(encoded)
开发者ID:iandees，项目名称:TileStache，代码行数:58，代码来源:__init__.py
示例18: test_ErrorMessage
def test_ErrorMessage(self):
m = messaging.ErrorMessage(faultString='ValueError')
self.assertEqual(pyamf.encode(m).getvalue(),
'\n\x81[Iflex.messaging.messages.ErrorMessage\x1bcorrelationId\x15'
'timeToLive\x13timestamp\x13messageId\x0fheaders\x17destination'
'\x11clientId\tbody\x19extendedData\x13faultCode\x17faultDetail'
'\x17faultString\x13rootCause\x01\x01\x01\x01\n\x0b\x01\x01\x01'
'\x01\x01\n\x05\x01\x01\x01\x06\x15ValueError\n\x05\x01\x01')
开发者ID:meemeer，项目名称:kayframework-pyamf，代码行数:9，代码来源:test_flex_messaging.py
示例19: test_NOT_PROVIDED
def test_NOT_PROVIDED(self):
from django.db.models import fields
self.assertEqual(pyamf.encode(fields.NOT_PROVIDED, encoding=pyamf.AMF0).getvalue(),
'\x06')
encoder = pyamf.get_encoder(pyamf.AMF3)
encoder.writeElement(fields.NOT_PROVIDED)
self.assertEqual(encoder.stream.getvalue(), '\x00')
开发者ID:84322146，项目名称:pyamf，代码行数:9，代码来源:test_django.py
示例20: test_amf0_encode
def test_amf0_encode(self):
x = sets.ImmutableSet(["1", "2", "3"])
self.assertTrue(
check_buffer(
pyamf.encode(x, encoding=pyamf.AMF0).getvalue(),
("\n\x00\x00\x00\x03", ("\x02\x00\x011", "\x02\x00\x013", "\x02\x00\x012")),
)
)
开发者ID:nmatthews，项目名称:BlueCop-XBMC-Plugins，代码行数:9，代码来源:test_sets.py
注：本文中的pyamf.encode函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论