# -*- coding: utf-8 -*- # # Copyright (c) The PyAMF Project. # See LICENSE.txt for details. """ Tests for AMF0 Implementation. @since: 0.1.0 """ import unittest import datetime import pyamf from pyamf import amf0, util, xml, python from pyamf.tests.util import ( EncoderMixIn, DecoderMixIn, ClassCacheClearingTestCase, Spam, ClassicSpam) class TypesTestCase(unittest.TestCase): """ Tests the type mappings. """ def test_types(self): self.assertEqual(amf0.TYPE_NUMBER, '\x00') self.assertEqual(amf0.TYPE_BOOL, '\x01') self.assertEqual(amf0.TYPE_STRING, '\x02') self.assertEqual(amf0.TYPE_OBJECT, '\x03') self.assertEqual(amf0.TYPE_MOVIECLIP, '\x04') self.assertEqual(amf0.TYPE_NULL, '\x05') self.assertEqual(amf0.TYPE_UNDEFINED, '\x06') self.assertEqual(amf0.TYPE_REFERENCE, '\x07') self.assertEqual(amf0.TYPE_MIXEDARRAY, '\x08') self.assertEqual(amf0.TYPE_OBJECTTERM, '\x09') self.assertEqual(amf0.TYPE_ARRAY, '\x0a') self.assertEqual(amf0.TYPE_DATE, '\x0b') self.assertEqual(amf0.TYPE_LONGSTRING, '\x0c') self.assertEqual(amf0.TYPE_UNSUPPORTED, '\x0d') self.assertEqual(amf0.TYPE_RECORDSET, '\x0e') self.assertEqual(amf0.TYPE_XML, '\x0f') self.assertEqual(amf0.TYPE_TYPEDOBJECT, '\x10') self.assertEqual(amf0.TYPE_AMF3, '\x11') class EncoderTestCase(ClassCacheClearingTestCase, EncoderMixIn): """ Tests the output from the AMF0 L{Encoder} class. """ amf_type = pyamf.AMF0 def setUp(self): ClassCacheClearingTestCase.setUp(self) EncoderMixIn.setUp(self) def test_number(self): self.assertEncoded(0, '\x00\x00\x00\x00\x00\x00\x00\x00\x00') self.assertEncoded(0.2, '\x00\x3f\xc9\x99\x99\x99\x99\x99\x9a') self.assertEncoded(1, '\x00\x3f\xf0\x00\x00\x00\x00\x00\x00') self.assertEncoded(42, '\x00\x40\x45\x00\x00\x00\x00\x00\x00') self.assertEncoded(-123, '\x00\xc0\x5e\xc0\x00\x00\x00\x00\x00') self.assertEncoded(1.23456789, '\x00\x3f\xf3\xc0\xca\x42\x83\xde\x1b') def test_boolean(self): self.assertEncoded(True, '\x01\x01') self.assertEncoded(False, '\x01\x00') def test_string(self): self.assertEncoded('', '\x02\x00\x00') self.assertEncoded('hello', '\x02\x00\x05hello') # unicode taken from http://www.columbia.edu/kermit/utf8.html self.assertEncoded(u'ᚠᛇᚻ', '\x02\x00\t\xe1\x9a\xa0\xe1\x9b\x87\xe1\x9a\xbb') def test_null(self): self.assertEncoded(None, '\x05') def test_undefined(self): self.assertEncoded(pyamf.Undefined, '\x06') def test_list(self): self.assertEncoded([], '\x0a\x00\x00\x00\x00') self.assertEncoded([1, 2, 3], '\x0a\x00\x00\x00\x03\x00\x3f\xf0\x00\x00\x00\x00\x00\x00\x00\x40' '\x00\x00\x00\x00\x00\x00\x00\x00\x40\x08\x00\x00\x00\x00\x00\x00') self.assertEncoded((1, 2, 3), '\x0a\x00\x00\x00\x03\x00\x3f\xf0\x00' '\x00\x00\x00\x00\x00\x00\x40\x00\x00\x00\x00\x00\x00\x00\x00' '\x40\x08\x00\x00\x00\x00\x00\x00') def test_list_references(self): x = [] self.assertEqual(self.encode(x, x), '\n\x00\x00\x00\x00\x07\x00\x00') def test_longstring(self): s = 'a' * 65537 self.assertEncoded(s, '\x0c\x00\x01\x00\x01' + s) def test_dict(self): self.assertEncoded({'a': 'a'}, '\x03\x00\x01a\x02\x00\x01a\x00\x00\t') self.assertEncoded({12: True, 42: "Testing"}, '\x03', ( '\x00\x0242\x02\x00\x07Testing', '\x00\x0212\x01\x01' ), '\x00\x00\t') def test_mixed_array(self): d = pyamf.MixedArray(a=1, b=2, c=3) bytes = ('\x08\x00\x00\x00\x00', ( '\x00\x01a\x00?\xf0\x00\x00\x00\x00\x00\x00', '\x00\x01c\x00@\x08\x00\x00\x00\x00\x00\x00', '\x00\x01b\x00@\x00\x00\x00\x00\x00\x00\x00' ), '\x00\x00\t') self.assertEncoded(d, bytes) # test the reference self.assertEqual(self.encode(d), '\x07\x00\x00') def test_date(self): self.assertEncoded(datetime.datetime(2005, 3, 18, 1, 58, 31), '\x0bBp+6!\x15\x80\x00\x00\x00') self.assertEncoded(datetime.date(2003, 12, 1), '\x0bBo%\xe2\xb2\x80\x00\x00\x00\x00') self.assertEncoded(datetime.datetime(2009, 3, 8, 23, 30, 47, 770122), '\x0bBq\xfe\x86\xca5\xa1\xf4\x00\x00') self.assertRaises(pyamf.EncodeError, self.encode, datetime.time(22, 3)) def test_xml(self): blob = 'hello world' self.assertEncoded(xml.fromstring(blob), '\x0f\x00\x00\x00\x19' + blob) def test_xml_references(self): blob = 'hello world' x = xml.fromstring(blob) self.assertEncoded([x, x], '\n\x00\x00\x00\x02' + ('\x0f\x00\x00\x00\x19' + blob) * 2) def test_object(self): self.assertEncoded({'a': 'b'}, '\x03\x00\x01a\x02\x00\x01b\x00\x00\x09') def test_force_amf3(self): alias = pyamf.register_class(Spam, 'spam.eggs') alias.amf3 = True x = Spam() x.x = 'y' self.assertEncoded(x, '\x11\n\x0b\x13spam.eggs\x03x\x06\x03y\x01') def test_typed_object(self): pyamf.register_class(Spam, alias='org.pyamf.spam') x = Spam() x.baz = 'hello' self.assertEncoded(x, '\x10\x00\x0eorg.pyamf.spam\x00\x03baz' '\x02\x00\x05hello\x00\x00\t') def test_complex_list(self): self.assertEncoded([[1.0]], '\x0a\x00\x00\x00\x01\x0a\x00\x00\x00\x01' '\x00\x3f\xf0\x00\x00\x00\x00\x00\x00') self.assertEncoded([['test', 'test', 'test', 'test']], '\x0a\x00\x00\x00\x01\x0a\x00\x00\x00\x04' + ('\x02\x00\x04test' * 4)) x = {'a': 'spam', 'b': 'eggs'} self.assertEncoded([[x, x]], '\n\x00\x00\x00\x01\n\x00\x00\x00\x02' '\x03\x00\x01a\x02\x00\x04spam\x00\x01b\x02\x00\x04eggs\x00\x00' '\t\x07\x00\x02') def test_amf3(self): self.encoder.use_amf3 = True o = Spam() self.assertEncoded(o, '\x11\n\x0b\x01\x01') def test_anonymous(self): pyamf.register_class(Spam) x = Spam() x.spam = 'eggs' x.hello = 'world' self.assertEncoded(x, '\x03', ('\x00\x05hello\x02\x00\x05world', '\x00\x04spam\x02\x00\x04eggs'), '\x00\x00\t') def test_dynamic(self): x = Spam() x.foo = 'bar' x.hello = 'world' alias = pyamf.register_class(Spam) alias.exclude_attrs = ['foo'] alias.compile() self.assertTrue(alias.dynamic) self.assertEncoded(x, '\x03\x00\x05hello\x02\x00\x05world\x00\x00\t') def test_dynamic_static(self): x = Spam() x.foo = 'bar' x.hello = 'world' alias = pyamf.register_class(Spam) alias.static_attrs = ['hello'] alias.compile() self.assertTrue(alias.dynamic) self.assertEncoded(x, '\x03', ('\x00\x05hello\x02\x00\x05world', '\x00\x03foo\x02\x00\x03bar'), '\x00\x00\t') def test_dynamic_registered(self): x = Spam() x.foo = 'bar' x.hello = 'world' alias = pyamf.register_class(Spam, 'x') alias.exclude_attrs = ['foo'] alias.compile() self.assertTrue(alias.dynamic) self.assertEncoded(x, '\x10\x00\x01x', '\x00\x05hello\x02\x00\x05world', '\x00\x00\t') def test_custom_type(self): def write_as_list(list_interface_obj, encoder): list_interface_obj.ran = True self.assertEqual(id(encoder), id(self.encoder)) return list(list_interface_obj) class ListWrapper(object): ran = False def __iter__(self): return iter([1, 2, 3]) pyamf.add_type(ListWrapper, write_as_list) x = ListWrapper() self.encoder.writeElement(x) self.assertEqual(x.ran, True) self.assertEqual(self.buf.getvalue(), '\n\x00\x00\x00\x03\x00?\xf0' '\x00\x00\x00\x00\x00\x00\x00@\x00\x00\x00\x00\x00\x00\x00\x00@' '\x08\x00\x00\x00\x00\x00\x00') def test_old_style_classes(self): class Person: pass pyamf.register_class(Person, 'spam.eggs.Person') u = Person() u.family_name = 'Doe' u.given_name = 'Jane' self.encoder.writeElement(u) self.assertEncoded(u, '\x10\x00\x10spam.eggs.Person', ( '\x00\x0bfamily_name\x02\x00\x03Doe', '\x00\ngiven_name\x02\x00\x04Jane' ), '\x00\x00\t') def test_slots(self): class Person(object): __slots__ = ('family_name', 'given_name') u = Person() u.family_name = 'Doe' u.given_name = 'Jane' self.assertEncoded(u, '\x03', ( '\x00\x0bfamily_name\x02\x00\x03Doe', '\x00\ngiven_name\x02\x00\x04Jane' ), '\x00\x00\t') def test_slots_registered(self): class Person(object): __slots__ = ('family_name', 'given_name') u = Person() u.family_name = 'Doe' u.given_name = 'Jane' pyamf.register_class(Person, 'spam.eggs.Person') self.assertEncoded(u, '\x10\x00\x10spam.eggs.Person', ( '\x00\x0bfamily_name\x02\x00\x03Doe', '\x00\ngiven_name\x02\x00\x04Jane' ), '\x00\x00\t') def test_elementtree_tag(self): """ Pretend to look like an ElementTree object to try to fool PyAMF into encoding an xml type. """ class NotAnElement(object): items = lambda self: [] def __iter__(self): return iter([]) foo = NotAnElement() foo.tag = 'foo' foo.text = 'bar' foo.tail = None self.assertEncoded(foo, '\x03', ( '\x00\x04text\x02\x00\x03bar', '\x00\x04tail\x05', '\x00\x03tag\x02\x00\x03foo', ), '\x00\x00\t') def test_funcs(self): def x(): pass for i in (chr, self.assertRaises, lambda x: x, pyamf): self.assertRaises(pyamf.EncodeError, self.encoder.writeElement, i) def test_external_subclassed_list(self): class L(list): class __amf__: external = True def __readamf__(self, o): pass def __writeamf__(self, o): pass pyamf.register_class(L, 'a') a = L() a.append('foo') a.append('bar') self.assertEncoded(a, '\x10\x00\x01a\x00\x00\t') def test_nonexternal_subclassed_list(self): class L(list): pass pyamf.register_class(L, 'a') a = L() a.append('foo') a.append('bar') self.assertEncoded(a, '\n\x00\x00\x00\x02\x02\x00\x03foo\x02\x00\x03bar') def test_amf3_xml(self): self.encoder.use_amf3 = True blob = '
' blob = xml.tostring(xml.fromstring(blob)) bytes = self.encode(xml.fromstring(blob)) buf = util.BufferedByteStream(bytes) self.assertEqual(buf.read_uchar(), 17) self.assertEqual(buf.read_uchar(), 11) self.assertEqual(buf.read_uchar() >> 1, buf.remaining()) self.assertEqual(buf.read(), blob) def test_use_amf3(self): self.encoder.use_amf3 = True x = {'foo': 'bar', 'baz': 'gak'} self.assertEncoded(x, '\x11\n\x0b', ('\x01\x07foo\x06\x07bar', '\x07baz\x06\x07gak\x01')) def test_static_attrs(self): class Foo(object): class __amf__: static = ('foo', 'bar') pyamf.register_class(Foo) x = Foo() x.foo = 'baz' x.bar = 'gak' self.assertEncoded(x, '\x03', ('\x00\x03bar\x02\x00\x03gak', '\x00\x03foo\x02\x00\x03baz'), '\x00\x00\t') def test_class(self): class Classic: pass class New(object): pass self.assertRaises(pyamf.EncodeError, self.encoder.writeElement, Classic) self.assertRaises(pyamf.EncodeError, self.encoder.writeElement, New) def test_timezone(self): d = datetime.datetime(2009, 9, 24, 14, 23, 23) self.encoder.timezone_offset = datetime.timedelta(hours=-5) self.assertEncoded(d, '\x0bBr>\xd8\x1f\xff\x80\x00\x00\x00') def test_generators(self): def foo(): yield [1, 2, 3] yield '\xff' yield pyamf.Undefined self.assertEncoded(foo(), '\n\x00\x00\x00\x03\x00?\xf0\x00\x00\x00\x00' '\x00\x00\x00@\x00\x00\x00\x00\x00\x00\x00\x00@\x08\x00\x00\x00\x00' '\x00\x00\x02\x00\x01\xff\x06') def test_iterate(self): self.assertRaises(StopIteration, self.encoder.next) self.encoder.send('') self.encoder.send('hello') self.encoder.send(u'ƒøø') self.assertEqual(self.encoder.next(), '\x02\x00\x00') self.assertEqual(self.encoder.next(), '\x02\x00\x05hello') self.assertEqual(self.encoder.next(), '\x02\x00\x06\xc6\x92\xc3\xb8\xc3\xb8') self.assertRaises(StopIteration, self.encoder.next) self.assertIdentical(iter(self.encoder), self.encoder) self.assertEqual(self.buf.getvalue(), '\x02\x00\x00\x02\x00\x05hello\x02\x00\x06\xc6\x92\xc3\xb8\xc3\xb8') class DecoderTestCase(ClassCacheClearingTestCase, DecoderMixIn): """ Tests the output from the AMF0 L{Decoder} class. """ amf_type = pyamf.AMF0 def setUp(self): ClassCacheClearingTestCase.setUp(self) DecoderMixIn.setUp(self) def test_undefined(self): self.assertDecoded(pyamf.Undefined, '\x06') def test_number(self): self.assertDecoded(0, '\x00\x00\x00\x00\x00\x00\x00\x00\x00') self.assertDecoded(0.2, '\x00\x3f\xc9\x99\x99\x99\x99\x99\x9a') self.assertDecoded(1, '\x00\x3f\xf0\x00\x00\x00\x00\x00\x00') self.assertDecoded(42, '\x00\x40\x45\x00\x00\x00\x00\x00\x00') self.assertDecoded(-123, '\x00\xc0\x5e\xc0\x00\x00\x00\x00\x00') self.assertDecoded(1.23456789, '\x00\x3f\xf3\xc0\xca\x42\x83\xde\x1b') def test_number_types(self): nr_types = [ ('\x00\x00\x00\x00\x00\x00\x00\x00\x00', int), ('\x00\x3f\xc9\x99\x99\x99\x99\x99\x9a', float), ('\x00\x3f\xf0\x00\x00\x00\x00\x00\x00', int), ('\x00\x40\x45\x00\x00\x00\x00\x00\x00', int), ('\x00\xc0\x5e\xc0\x00\x00\x00\x00\x00', int), ('\x00\x3f\xf3\xc0\xca\x42\x83\xde\x1b', float), ('\x00\xff\xf8\x00\x00\x00\x00\x00\x00', float), # nan ('\x00\xff\xf0\x00\x00\x00\x00\x00\x00', float), # -inf ('\x00\x7f\xf0\x00\x00\x00\x00\x00\x00', float), # inf ] for t in nr_types: bytes, expected_type = t self.buf.truncate() self.buf.write(bytes) self.buf.seek(0) self.assertEqual(type(self.decoder.readElement()), expected_type) def test_infinites(self): self.buf.truncate() self.buf.write('\x00\xff\xf8\x00\x00\x00\x00\x00\x00') self.buf.seek(0) x = self.decoder.readElement() self.assertTrue(python.isNaN(x)) self.buf.truncate() self.buf.write('\x00\xff\xf0\x00\x00\x00\x00\x00\x00') self.buf.seek(0) x = self.decoder.readElement() self.assertTrue(python.isNegInf(x)) self.buf.truncate() self.buf.write('\x00\x7f\xf0\x00\x00\x00\x00\x00\x00') self.buf.seek(0) x = self.decoder.readElement() self.assertTrue(python.isPosInf(x)) def test_boolean(self): self.assertDecoded(True, '\x01\x01') self.assertDecoded(False, '\x01\x00') def test_string(self): self.assertDecoded('', '\x02\x00\x00') self.assertDecoded('hello', '\x02\x00\x05hello') self.assertDecoded(u'ᚠᛇᚻ', '\x02\x00\t\xe1\x9a\xa0\xe1\x9b\x87\xe1\x9a\xbb') def test_longstring(self): a = 'a' * 65537 self.assertDecoded(a, '\x0c\x00\x01\x00\x01' + a) def test_null(self): self.assertDecoded(None, '\x05') def test_list(self): self.assertDecoded([], '\x0a\x00\x00\x00\x00') self.assertDecoded([1, 2, 3], '\x0a\x00\x00\x00\x03\x00\x3f\xf0\x00' '\x00\x00\x00\x00\x00\x00\x40\x00\x00\x00\x00\x00\x00\x00\x00\x40' '\x08\x00\x00\x00\x00\x00\x00') def test_dict(self): bytes = '\x08\x00\x00\x00\x00\x00\x01\x61\x02\x00\x01\x61\x00\x00\x09' self.assertDecoded({'a': 'a'}, bytes) self.buf.write(bytes) self.buf.seek(0) d = self.decoder.readElement() def test_mixed_array(self): bytes = ('\x08\x00\x00\x00\x00\x00\x01a\x00?\xf0\x00\x00\x00\x00\x00' '\x00\x00\x01c\x00@\x08\x00\x00\x00\x00\x00\x00\x00\x01b\x00@\x00' '\x00\x00\x00\x00\x00\x00\x00\x00\t') self.assertDecoded(pyamf.MixedArray(a=1, b=2, c=3), bytes) self.buf.write(bytes) self.buf.seek(0) d = self.decoder.readElement() def test_date(self): self.assertDecoded(datetime.datetime(2005, 3, 18, 1, 58, 31), '\x0bBp+6!\x15\x80\x00\x00\x00') self.assertDecoded(datetime.datetime(2009, 3, 8, 23, 30, 47, 770122), '\x0bBq\xfe\x86\xca5\xa1\xf4\x00\x00') def test_xml(self): e = 'hello world' ret = self.decode('\x0f\x00\x00\x00\x19' + e) self.assertEqual(xml.tostring(ret), e) def test_xml_references(self): self.buf.truncate(0) self.buf.write('\x0f\x00\x00\x00\x19hello world' '\x07\x00\x00') self.buf.seek(0) self.assertEqual( xml.tostring(xml.fromstring('hello world')), xml.tostring(self.decoder.readElement())) self.assertEqual( xml.tostring(xml.fromstring('hello world')), xml.tostring(self.decoder.readElement())) def test_object(self): bytes = '\x03\x00\x01a\x02\x00\x01b\x00\x00\x09' self.assertDecoded({'a': 'b'}, bytes) self.buf.write(bytes) self.buf.seek(0) d = self.decoder.readElement() def test_registered_class(self): pyamf.register_class(Spam, alias='org.pyamf.spam') bytes = ('\x10\x00\x0eorg.pyamf.spam\x00\x03baz' '\x02\x00\x05hello\x00\x00\x09') obj = self.decode(bytes) self.assertEqual(type(obj), Spam) self.assertTrue(hasattr(obj, 'baz')) self.assertEqual(obj.baz, 'hello') def test_complex_list(self): x = datetime.datetime(2007, 11, 3, 8, 7, 37, 437000) self.assertDecoded([['test','test','test','test']], '\x0A\x00\x00\x00\x01\x0A\x00\x00\x00\x04\x02\x00\x04\x74\x65\x73' '\x74\x02\x00\x04\x74\x65\x73\x74\x02\x00\x04\x74\x65\x73\x74\x02' '\x00\x04\x74\x65\x73\x74') self.assertDecoded([x], '\x0a\x00\x00\x00\x01\x0b\x42\x71\x60\x48\xcf' '\xed\xd0\x00\x00\x00') self.assertDecoded( [[{u'a': u'spam', u'b': u'eggs'}, {u'a': u'spam', u'b': u'eggs'}]], '\n\x00\x00\x00\x01\n\x00\x00\x00\x02\x08\x00\x00\x00\x00\x00\x01' 'a\x02\x00\x04spam\x00\x01b\x02\x00\x04eggs\x00\x00\t\x07\x00\x02') self.assertDecoded([[1.0]], '\x0A\x00\x00\x00\x01\x0A\x00\x00\x00\x01' '\x00\x3F\xF0\x00\x00\x00\x00\x00\x00') def test_amf3(self): self.buf.write('\x11\x04\x01') self.buf.seek(0) self.assertEqual(self.decoder.readElement(), 1) def test_dynamic(self): class Foo(pyamf.ASObject): pass x = Foo() x.foo = 'bar' alias = pyamf.register_class(Foo, 'x') alias.exclude_attrs = ['hello'] self.assertDecoded(x, '\x10\x00\x01x\x00\x03foo\x02\x00\x03bar\x00' '\x05hello\x02\x00\x05world\x00\x00\t') def test_classic_class(self): pyamf.register_class(ClassicSpam, 'spam.eggs') self.buf.write('\x10\x00\tspam.eggs\x00\x03foo\x02\x00\x03bar\x00\x00\t') self.buf.seek(0) foo = self.decoder.readElement() self.assertEqual(foo.foo, 'bar') def test_not_strict(self): self.assertFalse(self.decoder.strict) # write a typed object to the stream self.buf.write('\x10\x00\tspam.eggs\x00\x03foo\x02\x00\x03bar\x00\x00\t') self.buf.seek(0) self.assertFalse('spam.eggs' in pyamf.CLASS_CACHE) obj = self.decoder.readElement() self.assertTrue(isinstance(obj, pyamf.TypedObject)) self.assertEqual(obj.alias, 'spam.eggs') self.assertEqual(obj, {'foo': 'bar'}) def test_strict(self): self.decoder.strict = True self.assertTrue(self.decoder.strict) # write a typed object to the stream self.buf.write('\x10\x00\tspam.eggs\x00\x03foo\x02\x00\x03bar\x00\x00\t') self.buf.seek(0) self.assertFalse('spam.eggs' in pyamf.CLASS_CACHE) self.assertRaises(pyamf.UnknownClassAlias, self.decoder.readElement) def test_slots(self): class Person(object): __slots__ = ('family_name', 'given_name') self.buf.write('\x03\x00\x0bfamily_name\x02\x00\x03Doe\x00\n' 'given_name\x02\x00\x04Jane\x00\x00\t') self.buf.seek(0) foo = self.decoder.readElement() self.assertEqual(foo.family_name, 'Doe') self.assertEqual(foo.given_name, 'Jane') def test_slots_registered(self): class Person(object): __slots__ = ('family_name', 'given_name') pyamf.register_class(Person, 'spam.eggs.Person') self.buf.write('\x10\x00\x10spam.eggs.Person\x00\x0bfamily_name\x02' '\x00\x03Doe\x00\ngiven_name\x02\x00\x04Jane\x00\x00\t') self.buf.seek(0) foo = self.decoder.readElement() self.assertTrue(isinstance(foo, Person)) self.assertEqual(foo.family_name, 'Doe') self.assertEqual(foo.given_name, 'Jane') def test_ioerror_buffer_position(self): """ Test to ensure that if an IOError is raised by `readElement` that the original position of the stream is restored. """ bytes = pyamf.encode(u'foo', [1, 2, 3], encoding=pyamf.AMF0).getvalue() self.buf.write(bytes[:-1]) self.buf.seek(0) self.decoder.readElement() self.assertEqual(self.buf.tell(), 6) self.assertRaises(IOError, self.decoder.readElement) self.assertEqual(self.buf.tell(), 6) def test_timezone(self): self.decoder.timezone_offset = datetime.timedelta(hours=-5) self.buf.write('\x0bBr>\xc6\xf5w\x80\x00\x00\x00') self.buf.seek(0) f = self.decoder.readElement() self.assertEqual(f, datetime.datetime(2009, 9, 24, 9, 23, 23)) def test_unsupported(self): self.assertDecoded(None, '\x0D') def test_bad_reference(self): self.assertRaises(pyamf.ReferenceError, self.decode, '\x07\x00\x03') def test_iterate(self): self.assertRaises(StopIteration, self.decoder.next) self.decoder.send('\x02\x00\x00') self.decoder.send('\x02\x00\x05hello') self.decoder.send('\x02\x00\t\xe1\x9a\xa0\xe1\x9b\x87\xe1\x9a\xbb') self.assertEqual(self.decoder.next(), '') self.assertEqual(self.decoder.next(), 'hello') self.assertEqual(self.decoder.next(), u'\u16a0\u16c7\u16bb') self.assertRaises(StopIteration, self.decoder.next) self.assertIdentical(iter(self.decoder), self.decoder) def test_bad_type(self): self.assertRaises(pyamf.DecodeError, self.decode, '\xff') def test_kwargs(self): """ Python <= 3 demand that kwargs keys be bytes instead of unicode/string. """ def f(**kwargs): self.assertEqual(kwargs, {'a': 'a'}) kwargs = self.decode('\x03\x00\x01a\x02\x00\x01a\x00\x00\t') f(**kwargs) class RecordSetTestCase(unittest.TestCase, EncoderMixIn, DecoderMixIn): """ Tests for L{amf0.RecordSet} """ amf_type = pyamf.AMF0 blob = ( '\x10\x00\tRecordSet\x00\nserverInfo\x03', ( '\x00\x06cursor\x00?\xf0\x00\x00\x00\x00\x00\x00', '\x00\x0bcolumnNames\n\x00\x00\x00\x03\x02\x00\x01a\x02\x00\x01b\x02\x00\x01c', '\x00\x0binitialData\n\x00\x00\x00\x03\n\x00\x00\x00\x03\x00?\xf0' '\x00\x00\x00\x00\x00\x00\x00@\x00\x00\x00\x00\x00\x00\x00\x00' '@\x08\x00\x00\x00\x00\x00\x00\n\x00\x00\x00\x03\x00@\x10\x00' '\x00\x00\x00\x00\x00\x00@\x14\x00\x00\x00\x00\x00\x00\x00@\x18' '\x00\x00\x00\x00\x00\x00\n\x00\x00\x00\x03\x00@\x1c\x00\x00' '\x00\x00\x00\x00\x00@ \x00\x00\x00\x00\x00\x00\x00@"\x00\x00' '\x00\x00\x00\x00', '\x00\x07version\x00?\xf0\x00\x00\x00\x00\x00\x00', '\x00\ntotalCount\x00@\x08\x00\x00\x00\x00\x00\x00'), '\x00\x00\t\x00\x00\t') def setUp(self): unittest.TestCase.setUp(self) EncoderMixIn.setUp(self) DecoderMixIn.setUp(self) def test_create(self): x = amf0.RecordSet() self.assertEqual(x.columns, []) self.assertEqual(x.items, []) self.assertEqual(x.service, None) self.assertEqual(x.id, None) x = amf0.RecordSet(columns=['spam', 'eggs'], items=[[1, 2]]) self.assertEqual(x.columns, ['spam', 'eggs']) self.assertEqual(x.items, [[1, 2]]) self.assertEqual(x.service, None) self.assertEqual(x.id, None) x = amf0.RecordSet(service={}, id=54) self.assertEqual(x.columns, []) self.assertEqual(x.items, []) self.assertEqual(x.service, {}) self.assertEqual(x.id, 54) def test_server_info(self): # empty recordset x = amf0.RecordSet() si = x.serverInfo self.assertTrue(isinstance(si, dict)) self.assertEqual(si.cursor, 1) self.assertEqual(si.version, 1) self.assertEqual(si.columnNames, []) self.assertEqual(si.initialData, []) self.assertEqual(si.totalCount, 0) try: si.serviceName except AttributeError: pass try: si.id except AttributeError: pass # basic create x = amf0.RecordSet(columns=['a', 'b', 'c'], items=[ [1, 2, 3], [4, 5, 6], [7, 8, 9]]) si = x.serverInfo self.assertTrue(isinstance(si, dict)) self.assertEqual(si.cursor, 1) self.assertEqual(si.version, 1) self.assertEqual(si.columnNames, ['a', 'b', 'c']) self.assertEqual(si.initialData, [[1, 2, 3], [4, 5, 6], [7, 8, 9]]) self.assertEqual(si.totalCount, 3) try: si.serviceName except AttributeError: pass try: si.id except AttributeError: pass # with service & id service = {'name': 'baz'} x = amf0.RecordSet(columns=['spam'], items=[['eggs']], service=service, id='asdfasdf') si = x.serverInfo self.assertTrue(isinstance(si, dict)) self.assertEqual(si.cursor, 1) self.assertEqual(si.version, 1) self.assertEqual(si.columnNames, ['spam']) self.assertEqual(si.initialData, [['eggs']]) self.assertEqual(si.totalCount, 1) self.assertEqual(si.serviceName, 'baz') self.assertEqual(si.id, 'asdfasdf') def test_encode(self): self.buf = self.encoder.stream x = amf0.RecordSet(columns=['a', 'b', 'c'], items=[ [1, 2, 3], [4, 5, 6], [7, 8, 9]]) self.assertEncoded(x, self.blob) def test_decode(self): self.buf = self.decoder.stream x = self.decode(self.blob) self.assertTrue(isinstance(x, amf0.RecordSet)) self.assertEqual(x.columns, ['a', 'b', 'c']) self.assertEqual(x.items, [[1, 2, 3], [4, 5, 6], [7, 8, 9]]) self.assertEqual(x.service, None) self.assertEqual(x.id, None) def test_repr(self): x = amf0.RecordSet(columns=['spam'], items=[['eggs']], service={'name': 'baz'}, id='asdfasdf') self.assertEqual(repr(x), "" % (id(x),)) class ClassInheritanceTestCase(ClassCacheClearingTestCase, EncoderMixIn): amf_type = pyamf.AMF0 def setUp(self): # wtf ClassCacheClearingTestCase.setUp(self) EncoderMixIn.setUp(self) def test_simple(self): class A(object): class __amf__: static = ('a') class B(A): class __amf__: static = ('b') pyamf.register_class(A, 'A') pyamf.register_class(B, 'B') x = B() x.a = 'spam' x.b = 'eggs' self.assertEncoded(x, '\x10\x00\x01B', ('\x00\x01a\x02\x00\x04spam', '\x00\x01b\x02\x00\x04eggs'), '\x00\x00\t') def test_deep(self): class A(object): class __amf__: static = ('a') class B(A): class __amf__: static = ('b') class C(B): class __amf__: static = ('c') pyamf.register_class(A, 'A') pyamf.register_class(B, 'B') pyamf.register_class(C, 'C') x = C() x.a = 'spam' x.b = 'eggs' x.c = 'foo' self.assertEncoded(x, '\x10\x00\x01C', ('\x00\x01a\x02\x00\x04spam', '\x00\x01c\x02\x00\x03foo', '\x00\x01b\x02\x00\x04eggs'), '\x00\x00\t') class ExceptionEncodingTestCase(ClassCacheClearingTestCase): """ Tests for encoding exceptions. """ def setUp(self): ClassCacheClearingTestCase.setUp(self) self.buffer = util.BufferedByteStream() self.encoder = amf0.Encoder(self.buffer) def test_exception(self): try: raise Exception('foo bar') except Exception, e: self.encoder.writeElement(e) self.assertEqual(self.buffer.getvalue(), '\x03\x00\x07message\x02' '\x00\x07foo bar\x00\x04name\x02\x00\tException\x00\x00\t') def test_user_defined(self): class FooBar(Exception): pass try: raise FooBar('foo bar') except Exception, e: self.encoder.writeElement(e) self.assertEqual(self.buffer.getvalue(), '\x03\x00\x07message\x02' '\x00\x07foo bar\x00\x04name\x02\x00\x06FooBar\x00\x00\t') def test_typed(self): class XYZ(Exception): pass pyamf.register_class(XYZ, 'foo.bar') try: raise XYZ('blarg') except Exception, e: self.encoder.writeElement(e) self.assertEqual(self.buffer.getvalue(), '\x10\x00\x07foo.bar\x00' '\x07message\x02\x00\x05blarg\x00\x04name\x02\x00\x03XYZ\x00\x00\t') class AMF0ContextTestCase(unittest.TestCase): """ """ bytes = ('\x00\x03\x00\x02\x00\x0eServiceLicense\x00\x00\x00\x00O\x11\n\x0b' '\x01-serviceConfigurationId\x06\t1234\x15licenseKey\x06Axxxxxxxxxxxxxx' 'xxxxxxxxxxxxxxxxxx\x01\x00\tSessionId\x00\x00\x00\x00\xb2\x11\n\x0b' '\x01\x0bToken\x06\x82Iyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy' 'yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy' 'yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy\x01\x00\x01\x00\x0cRegi' 'sterUser\x00\x02/3\x00\x00\x01k\n\x00\x00\x00\x07\x11\n#\x01\rformat' '\x0bvalue\x069urn:TribalDDB:identity:email\x06!tester@trial.com\x11\n' '#\x01\x02\ttype\x06\x0fpasswrd\x06Kurn:TribalDDB:authentication:passwo' 'rd\x11\nS\x01\x19EmailAddress\x15PostalCode\x17DateOfBirth\x11LastName' '\x13FirstName\x06\x06\x06\x0b12345\n3\x12\x0bmonth\x07day\tyear\x04' '\x04\x04\x0f\x04\x8fF\x06\rewrwer\x06\x07wer\x11\n3\x1fSectionTracking' '\tCsId\x11TrtmntId\x13LocalCsId\x04\x00\x04\x86\x94z\x04\x00\x11\n' '\x13\x11Tracking\x07CTC\x06\x07555\x11\t\x03\x01\n#\x13UserOptIn\x1dli' 'veModeEnable\x05id\x02\x04\x884\x02\x00\x10wwwwwwwwwwwwwwww') def test_decode(self): from pyamf.remoting import decode e = decode(self.bytes) a, b, c, d, e, f, g = e['/3'].body self.assertEqual(a, {'value': u'tester@trial.com', 'format': u'urn:TribalDDB:identity:email'}) self.assertEqual(b, {'type': u'urn:TribalDDB:authentication:password', 'value': u'passwrd'}) self.assertEqual(c, {'PostalCode': u'12345', 'DateOfBirth': {'month': 4, 'day': 15, 'year': 1990}, 'EmailAddress': u'tester@trial.com', 'FirstName': u'wer', 'LastName': u'ewrwer'}) self.assertEqual(d, {'CsId': 0, 'TrtmntId': 100986, 'LocalCsId': 0}) self.assertEqual(e, {'CTC': u'555'}) self.assertEqual(f, [{'liveModeEnable': False, 'id': 1076}]) self.assertEqual(g, u'wwwwwwwwwwwwwwww')