| # Copyright (c) The PyAMF Project. |
| # See LICENSE.txt for details. |
| |
| """ |
| PyAMF SQLAlchemy adapter tests. |
| |
| @since 0.4 |
| """ |
| |
| import unittest |
| |
| try: |
| import sqlalchemy |
| from sqlalchemy import MetaData, Table, Column, Integer, String, ForeignKey, \ |
| create_engine |
| from sqlalchemy.orm import mapper, relation, sessionmaker, clear_mappers |
| |
| from pyamf.adapters import _sqlalchemy_orm as adapter |
| except ImportError: |
| sqlalchemy = None |
| |
| import pyamf.flex |
| from pyamf.tests.util import Spam |
| |
| |
| class BaseObject(object): |
| def __init__(self, **kwargs): |
| self.__dict__.update(kwargs) |
| |
| |
| class User(BaseObject): |
| def __init__(self, **kwargs): |
| BaseObject.__init__(self, **kwargs) |
| |
| self.lazy_loaded = [LazyLoaded()] |
| |
| |
| class Address(BaseObject): |
| pass |
| |
| |
| class LazyLoaded(BaseObject): |
| pass |
| |
| |
| class AnotherLazyLoaded(BaseObject): |
| pass |
| |
| |
| class BaseTestCase(unittest.TestCase): |
| """ |
| Initialise up all table/mappers. |
| """ |
| |
| def setUp(self): |
| if not sqlalchemy: |
| self.skipTest("'sqlalchemy' is not available") |
| |
| # Create DB and map objects |
| self.metadata = MetaData() |
| self.engine = create_engine('sqlite:///:memory:', echo=False) |
| |
| Session = sessionmaker(bind=self.engine) |
| |
| self.session = Session() |
| self.tables = {} |
| |
| self.tables['users'] = Table('users', self.metadata, |
| Column('id', Integer, primary_key=True), |
| Column('name', String(64))) |
| |
| self.tables['addresses'] = Table('addresses', self.metadata, |
| Column('id', Integer, primary_key=True), |
| Column('user_id', Integer, ForeignKey('users.id')), |
| Column('email_address', String(128))) |
| |
| self.tables['lazy_loaded'] = Table('lazy_loaded', self.metadata, |
| Column('id', Integer, primary_key=True), |
| Column('user_id', Integer, ForeignKey('users.id'))) |
| |
| self.tables['another_lazy_loaded'] = Table('another_lazy_loaded', self.metadata, |
| Column('id', Integer, primary_key=True), |
| Column('user_id', Integer, ForeignKey('users.id'))) |
| |
| self.mappers = {} |
| |
| self.mappers['user'] = mapper(User, self.tables['users'], properties={ |
| 'addresses': relation(Address, backref='user', lazy=False), |
| 'lazy_loaded': relation(LazyLoaded, lazy=True), |
| 'another_lazy_loaded': relation(AnotherLazyLoaded, lazy=True) |
| }) |
| |
| self.mappers['addresses'] = mapper(Address, self.tables['addresses']) |
| self.mappers['lazy_loaded'] = mapper(LazyLoaded, |
| self.tables['lazy_loaded']) |
| self.mappers['another_lazy_loaded'] = mapper(AnotherLazyLoaded, |
| self.tables['another_lazy_loaded']) |
| |
| self.metadata.create_all(self.engine) |
| |
| pyamf.register_class(User, 'server.User') |
| pyamf.register_class(Address, 'server.Address') |
| pyamf.register_class(LazyLoaded, 'server.LazyLoaded') |
| |
| def tearDown(self): |
| clear_mappers() |
| |
| pyamf.unregister_class(User) |
| pyamf.unregister_class(Address) |
| pyamf.unregister_class(LazyLoaded) |
| |
| def _build_obj(self): |
| user = User() |
| user.name = "test_user" |
| user.addresses.append(Address(email_address="test@example.org")) |
| |
| return user |
| |
| def _save(self, obj): |
| # this covers deprecation warnings etc. |
| if hasattr(self.session, 'add'): |
| self.session.add(obj) |
| elif hasattr(self.session, 'save'): |
| self.session.save(obj) |
| else: |
| raise AttributeError('Don\'t know how to save an object') |
| |
| def _clear(self): |
| # this covers deprecation warnings etc. |
| if hasattr(self.session, 'expunge_all'): |
| self.session.expunge_all() |
| elif hasattr(self.session, 'clear'): |
| self.session.clear() |
| else: |
| raise AttributeError('Don\'t know how to clear session') |
| |
| |
| class SATestCase(BaseTestCase): |
| def _test_obj(self, encoded, decoded): |
| self.assertEqual(User, decoded.__class__) |
| self.assertEqual(encoded.name, decoded.name) |
| self.assertEqual(encoded.addresses[0].email_address, decoded.addresses[0].email_address) |
| |
| def test_encode_decode_transient(self): |
| user = self._build_obj() |
| |
| encoder = pyamf.get_encoder(pyamf.AMF3) |
| encoder.writeElement(user) |
| encoded = encoder.stream.getvalue() |
| decoded = pyamf.get_decoder(pyamf.AMF3, encoded).readElement() |
| |
| self._test_obj(user, decoded) |
| |
| def test_encode_decode_persistent(self): |
| user = self._build_obj() |
| self._save(user) |
| self.session.commit() |
| self.session.refresh(user) |
| |
| encoder = pyamf.get_encoder(pyamf.AMF3) |
| encoder.writeElement(user) |
| encoded = encoder.stream.getvalue() |
| decoded = pyamf.get_decoder(pyamf.AMF3, encoded).readElement() |
| |
| self._test_obj(user, decoded) |
| |
| def test_encode_decode_list(self): |
| max = 5 |
| for i in range(0, max): |
| user = self._build_obj() |
| user.name = "%s" % i |
| self._save(user) |
| |
| self.session.commit() |
| users = self.session.query(User).all() |
| |
| encoder = pyamf.get_encoder(pyamf.AMF3) |
| |
| encoder.writeElement(users) |
| encoded = encoder.stream.getvalue() |
| decoded = pyamf.get_decoder(pyamf.AMF3, encoded).readElement() |
| self.assertEqual([].__class__, decoded.__class__) |
| |
| for i in range(0, max): |
| self._test_obj(users[i], decoded[i]) |
| |
| def test_sa_merge(self): |
| user = self._build_obj() |
| |
| for i, string in enumerate(['one', 'two', 'three']): |
| addr = Address(email_address="%s@example.org" % string) |
| user.addresses.append(addr) |
| |
| self._save(user) |
| self.session.commit() |
| self.session.refresh(user) |
| |
| encoder = pyamf.get_encoder(pyamf.AMF3) |
| encoder.writeElement(user) |
| encoded = encoder.stream.getvalue() |
| |
| decoded = pyamf.get_decoder(pyamf.AMF3, encoded).readElement() |
| del decoded.addresses[0] |
| del decoded.addresses[1] |
| |
| merged_user = self.session.merge(decoded) |
| self.assertEqual(len(merged_user.addresses), 2) |
| |
| def test_encode_decode_with_references(self): |
| user = self._build_obj() |
| self._save(user) |
| self.session.commit() |
| self.session.refresh(user) |
| |
| max = 5 |
| users = [] |
| for i in range(0, max): |
| users.append(user) |
| |
| encoder = pyamf.get_encoder(pyamf.AMF3) |
| encoder.writeElement(users) |
| encoded = encoder.stream.getvalue() |
| |
| decoded = pyamf.get_decoder(pyamf.AMF3, encoded).readElement() |
| |
| for i in range(0, max): |
| self.assertEqual(id(decoded[0]), id(decoded[i])) |
| |
| |
| class BaseClassAliasTestCase(BaseTestCase): |
| def setUp(self): |
| BaseTestCase.setUp(self) |
| |
| self.alias = pyamf.get_class_alias(User) |
| |
| |
| class ClassAliasTestCase(BaseClassAliasTestCase): |
| def test_type(self): |
| self.assertEqual(self.alias.__class__, adapter.SaMappedClassAlias) |
| |
| def test_get_mapper(self): |
| self.assertFalse(hasattr(self.alias, 'mapper')) |
| |
| self.alias.compile() |
| mapper = adapter.class_mapper(User) |
| |
| self.assertTrue(hasattr(self.alias, 'mapper')) |
| self.assertEqual(id(mapper), id(self.alias.mapper)) |
| |
| self.assertEqual(self.alias.static_attrs, []) |
| |
| def test_get_attrs(self): |
| u = self._build_obj() |
| attrs = self.alias.getEncodableAttributes(u) |
| |
| self.assertEqual(sorted(attrs.keys()), [ |
| 'addresses', |
| 'another_lazy_loaded', |
| 'id', |
| 'lazy_loaded', |
| 'name', |
| 'sa_key', |
| 'sa_lazy' |
| ]) |
| |
| self.assertEqual(attrs['sa_key'], [None]) |
| self.assertEqual(attrs['sa_lazy'], []) |
| |
| def test_get_attributes(self): |
| u = self._build_obj() |
| |
| self.assertFalse(u in self.session) |
| self.assertEqual([None], self.mappers['user'].primary_key_from_instance(u)) |
| attrs = self.alias.getEncodableAttributes(u) |
| |
| self.assertEqual(attrs, { |
| 'addresses': u.addresses, |
| 'lazy_loaded': u.lazy_loaded, |
| 'another_lazy_loaded': [], |
| 'id': None, |
| 'name': 'test_user', |
| 'sa_lazy': [], |
| 'sa_key': [None] |
| }) |
| |
| def test_property(self): |
| class Person(object): |
| foo = 'bar' |
| baz = 'gak' |
| |
| def _get_rw_property(self): |
| return self.foo |
| |
| def _set_rw_property(self, val): |
| self.foo = val |
| |
| def _get_ro_property(self): |
| return self.baz |
| |
| rw = property(_get_rw_property, _set_rw_property) |
| ro = property(_get_ro_property) |
| |
| self.mappers['person'] = mapper(Person, self.tables['users']) |
| |
| alias = adapter.SaMappedClassAlias(Person, 'person') |
| |
| obj = Person() |
| |
| attrs = alias.getEncodableAttributes(obj) |
| self.assertEqual(attrs, { |
| 'id': None, |
| 'name': None, |
| 'sa_key': [None], |
| 'sa_lazy': [], |
| 'rw': 'bar', |
| 'ro': 'gak'}) |
| |
| self.assertEqual(obj.ro, 'gak') |
| alias.applyAttributes(obj, { |
| 'sa_key': [None], |
| 'sa_lazy': [], |
| 'id': None, |
| 'name': None, |
| 'rw': 'bar', |
| 'ro': 'baz'}) |
| self.assertEqual(obj.ro, 'gak') |
| |
| |
| class ApplyAttributesTestCase(BaseClassAliasTestCase): |
| def test_undefined(self): |
| u = self.alias.createInstance() |
| |
| attrs = { |
| 'sa_lazy': ['another_lazy_loaded'], |
| 'sa_key': [None], |
| 'addresses': [], |
| 'lazy_loaded': [], |
| 'another_lazy_loaded': pyamf.Undefined, # <-- the important bit |
| 'id': None, |
| 'name': 'test_user' |
| } |
| |
| self.alias.applyAttributes(u, attrs) |
| |
| d = u.__dict__.copy() |
| |
| if sqlalchemy.__version__.startswith('0.4'): |
| self.assertTrue('_state' in d) |
| del d['_state'] |
| else: |
| self.assertTrue('_sa_instance_state' in d) |
| del d['_sa_instance_state'] |
| |
| self.assertEqual(d, { |
| 'lazy_loaded': [], |
| 'addresses': [], |
| 'name': 'test_user', |
| 'id': None |
| }) |
| |
| def test_decode_unaliased(self): |
| u = self.alias.createInstance() |
| |
| attrs = { |
| 'sa_lazy': [], |
| 'sa_key': [None], |
| 'addresses': [], |
| 'lazy_loaded': [], |
| # this is important because we haven't registered AnotherLazyLoaded |
| # as an alias and the decoded object for an untyped object is an |
| # instance of pyamf.ASObject |
| 'another_lazy_loaded': [pyamf.ASObject({'id': 1, 'user_id': None})], |
| 'id': None, |
| 'name': 'test_user' |
| } |
| |
| # sqlalchemy can't find any state to work with |
| self.assertRaises(AttributeError, self.alias.applyAttributes, u, attrs) |
| |
| |
| class AdapterTestCase(BaseTestCase): |
| """ |
| Checks to see if the adapter will actually intercept a class correctly. |
| """ |
| |
| def test_mapped(self): |
| self.assertNotEquals(None, adapter.class_mapper(User)) |
| self.assertTrue(adapter.is_class_sa_mapped(User)) |
| |
| def test_instance(self): |
| u = User() |
| |
| self.assertTrue(adapter.is_class_sa_mapped(u)) |
| |
| def test_not_mapped(self): |
| self.assertRaises(adapter.UnmappedInstanceError, adapter.class_mapper, Spam) |
| self.assertFalse(adapter.is_class_sa_mapped(Spam)) |
| |
| |
| class ExcludableAttrsTestCase(BaseTestCase): |
| """ |
| Tests for #790 |
| """ |
| |
| def test_core_attrs(self): |
| """ |
| Ensure that sa_key and sa_lazy can be excluded |
| """ |
| a = adapter.SaMappedClassAlias(Address, exclude_attrs=['sa_lazy', 'sa_key']) |
| u = Address() |
| |
| attrs = a.getEncodableAttributes(u) |
| |
| self.assertFalse('sa_key' in attrs) |
| self.assertFalse('sa_lazy' in attrs) |
