mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-05-28 03:26:01 -04:00
Khairi Hafsham
772374735d
tested using pycodestyle version 2.2.0 Fixes: #3885 Change-Id: I5df43adc3aefe318f9eeab72a078247a548ec566 Pull-request: https://github.com/zzzeek/sqlalchemy/pull/343
315 lines
11 KiB
Python
315 lines
11 KiB
Python
from test.orm import _fixtures
|
|
from sqlalchemy.testing import fixtures, assert_raises, eq_, ne_, \
|
|
assert_raises_message
|
|
from sqlalchemy.orm import mapper, Session, validates, relationship
|
|
from sqlalchemy.testing.mock import Mock, call
|
|
from sqlalchemy import exc
|
|
|
|
|
|
class ValidatorTest(_fixtures.FixtureTest):
|
|
def test_scalar(self):
|
|
users = self.tables.users
|
|
canary = Mock()
|
|
|
|
class User(fixtures.ComparableEntity):
|
|
@validates('name')
|
|
def validate_name(self, key, name):
|
|
canary(key, name)
|
|
ne_(name, 'fred')
|
|
return name + ' modified'
|
|
|
|
mapper(User, users)
|
|
sess = Session()
|
|
u1 = User(name='ed')
|
|
eq_(u1.name, 'ed modified')
|
|
assert_raises(AssertionError, setattr, u1, "name", "fred")
|
|
eq_(u1.name, 'ed modified')
|
|
eq_(canary.mock_calls, [call('name', 'ed'), call('name', 'fred')])
|
|
|
|
sess.add(u1)
|
|
sess.commit()
|
|
|
|
eq_(
|
|
sess.query(User).filter_by(name='ed modified').one(),
|
|
User(name='ed')
|
|
)
|
|
|
|
def test_collection(self):
|
|
users, addresses, Address = (self.tables.users,
|
|
self.tables.addresses,
|
|
self.classes.Address)
|
|
|
|
canary = Mock()
|
|
|
|
class User(fixtures.ComparableEntity):
|
|
@validates('addresses')
|
|
def validate_address(self, key, ad):
|
|
canary(key, ad)
|
|
assert '@' in ad.email_address
|
|
return ad
|
|
|
|
mapper(User, users, properties={
|
|
'addresses': relationship(Address)}
|
|
)
|
|
mapper(Address, addresses)
|
|
sess = Session()
|
|
u1 = User(name='edward')
|
|
a0 = Address(email_address='noemail')
|
|
assert_raises(AssertionError, u1.addresses.append, a0)
|
|
a1 = Address(id=15, email_address='foo@bar.com')
|
|
u1.addresses.append(a1)
|
|
eq_(canary.mock_calls, [call('addresses', a0), call('addresses', a1)])
|
|
sess.add(u1)
|
|
sess.commit()
|
|
|
|
eq_(
|
|
sess.query(User).filter_by(name='edward').one(),
|
|
User(name='edward', addresses=[
|
|
Address(email_address='foo@bar.com')])
|
|
)
|
|
|
|
def test_validators_dict(self):
|
|
users, addresses, Address = (self.tables.users,
|
|
self.tables.addresses,
|
|
self.classes.Address)
|
|
|
|
class User(fixtures.ComparableEntity):
|
|
|
|
@validates('name')
|
|
def validate_name(self, key, name):
|
|
ne_(name, 'fred')
|
|
return name + ' modified'
|
|
|
|
@validates('addresses')
|
|
def validate_address(self, key, ad):
|
|
assert '@' in ad.email_address
|
|
return ad
|
|
|
|
def simple_function(self, key, value):
|
|
return key, value
|
|
|
|
u_m = mapper(User, users, properties={
|
|
'addresses': relationship(Address)})
|
|
mapper(Address, addresses)
|
|
|
|
eq_(
|
|
dict((k, v[0].__name__) for k, v in list(u_m.validators.items())),
|
|
{'name': 'validate_name',
|
|
'addresses': 'validate_address'}
|
|
)
|
|
|
|
def test_validator_w_removes(self):
|
|
users, addresses, Address = (self.tables.users,
|
|
self.tables.addresses,
|
|
self.classes.Address)
|
|
canary = Mock()
|
|
|
|
class User(fixtures.ComparableEntity):
|
|
|
|
@validates('name', include_removes=True)
|
|
def validate_name(self, key, item, remove):
|
|
canary(key, item, remove)
|
|
return item
|
|
|
|
@validates('addresses', include_removes=True)
|
|
def validate_address(self, key, item, remove):
|
|
canary(key, item, remove)
|
|
return item
|
|
|
|
mapper(User, users, properties={
|
|
'addresses': relationship(Address)
|
|
})
|
|
mapper(Address, addresses)
|
|
|
|
u1 = User()
|
|
u1.name = "ed"
|
|
u1.name = "mary"
|
|
del u1.name
|
|
a1, a2, a3 = Address(), Address(), Address()
|
|
u1.addresses.append(a1)
|
|
u1.addresses.remove(a1)
|
|
u1.addresses = [a1, a2]
|
|
u1.addresses = [a2, a3]
|
|
|
|
eq_(canary.mock_calls, [
|
|
call('name', 'ed', False),
|
|
call('name', 'mary', False),
|
|
call('name', 'mary', True),
|
|
# append a1
|
|
call('addresses', a1, False),
|
|
# remove a1
|
|
call('addresses', a1, True),
|
|
# set to [a1, a2] - this is two appends
|
|
call('addresses', a1, False), call('addresses', a2, False),
|
|
# set to [a2, a3] - this is a remove of a1,
|
|
# append of a3. the appends are first.
|
|
call('addresses', a3, False),
|
|
call('addresses', a1, True),
|
|
])
|
|
|
|
def test_validator_multi_warning(self):
|
|
users = self.tables.users
|
|
|
|
class Foo(object):
|
|
@validates("name")
|
|
def validate_one(self, key, value):
|
|
pass
|
|
|
|
@validates("name")
|
|
def validate_two(self, key, value):
|
|
pass
|
|
|
|
assert_raises_message(
|
|
exc.InvalidRequestError,
|
|
"A validation function for mapped attribute "
|
|
"'name' on mapper Mapper|Foo|users already exists",
|
|
mapper, Foo, users
|
|
)
|
|
|
|
class Bar(object):
|
|
@validates("id")
|
|
def validate_three(self, key, value):
|
|
return value + 10
|
|
|
|
@validates("id", "name")
|
|
def validate_four(self, key, value):
|
|
return value + "foo"
|
|
|
|
assert_raises_message(
|
|
exc.InvalidRequestError,
|
|
"A validation function for mapped attribute "
|
|
"'name' on mapper Mapper|Bar|users already exists",
|
|
mapper, Bar, users
|
|
)
|
|
|
|
def test_validator_wo_backrefs_wo_removes(self):
|
|
self._test_validator_backrefs(False, False)
|
|
|
|
def test_validator_wo_backrefs_w_removes(self):
|
|
self._test_validator_backrefs(False, True)
|
|
|
|
def test_validator_w_backrefs_wo_removes(self):
|
|
self._test_validator_backrefs(True, False)
|
|
|
|
def test_validator_w_backrefs_w_removes(self):
|
|
self._test_validator_backrefs(True, True)
|
|
|
|
def _test_validator_backrefs(self, include_backrefs, include_removes):
|
|
users, addresses = (self.tables.users,
|
|
self.tables.addresses)
|
|
canary = Mock()
|
|
|
|
class User(fixtures.ComparableEntity):
|
|
|
|
if include_removes:
|
|
@validates('addresses', include_removes=True,
|
|
include_backrefs=include_backrefs)
|
|
def validate_address(self, key, item, remove):
|
|
canary(key, item, remove)
|
|
return item
|
|
else:
|
|
@validates('addresses', include_removes=False,
|
|
include_backrefs=include_backrefs)
|
|
def validate_address(self, key, item):
|
|
canary(key, item)
|
|
return item
|
|
|
|
class Address(fixtures.ComparableEntity):
|
|
if include_removes:
|
|
@validates('user', include_backrefs=include_backrefs,
|
|
include_removes=True)
|
|
def validate_user(self, key, item, remove):
|
|
canary(key, item, remove)
|
|
return item
|
|
else:
|
|
@validates('user', include_backrefs=include_backrefs)
|
|
def validate_user(self, key, item):
|
|
canary(key, item)
|
|
return item
|
|
|
|
mapper(User, users, properties={
|
|
'addresses': relationship(Address, backref="user")
|
|
})
|
|
mapper(Address, addresses)
|
|
|
|
u1 = User()
|
|
u2 = User()
|
|
a1, a2 = Address(), Address()
|
|
|
|
# 3 append/set, two removes
|
|
u1.addresses.append(a1)
|
|
u1.addresses.append(a2)
|
|
a2.user = u2
|
|
del a1.user
|
|
u2.addresses.remove(a2)
|
|
|
|
# copy, so that generation of the
|
|
# comparisons don't get caught
|
|
calls = list(canary.mock_calls)
|
|
|
|
if include_backrefs:
|
|
if include_removes:
|
|
eq_(calls,
|
|
[
|
|
# append #1
|
|
call('addresses', Address(), False),
|
|
|
|
# backref for append
|
|
call('user', User(addresses=[]), False),
|
|
|
|
# append #2
|
|
call('addresses', Address(user=None), False),
|
|
|
|
# backref for append
|
|
call('user', User(addresses=[]), False),
|
|
|
|
# assign a2.user = u2
|
|
call('user', User(addresses=[]), False),
|
|
|
|
# backref for u1.addresses.remove(a2)
|
|
call('addresses', Address(user=None), True),
|
|
|
|
# backref for u2.addresses.append(a2)
|
|
call('addresses', Address(user=None), False),
|
|
|
|
# del a1.user
|
|
call('user', User(addresses=[]), True),
|
|
|
|
# backref for u1.addresses.remove(a1)
|
|
call('addresses', Address(), True),
|
|
|
|
# u2.addresses.remove(a2)
|
|
call('addresses', Address(user=None), True),
|
|
|
|
# backref for a2.user = None
|
|
call('user', None, False)
|
|
])
|
|
else:
|
|
eq_(calls,
|
|
[
|
|
call('addresses', Address()),
|
|
call('user', User(addresses=[])),
|
|
call('addresses', Address(user=None)),
|
|
call('user', User(addresses=[])),
|
|
call('user', User(addresses=[])),
|
|
call('addresses', Address(user=None)),
|
|
call('user', None)
|
|
])
|
|
else:
|
|
if include_removes:
|
|
eq_(calls,
|
|
[
|
|
call('addresses', Address(), False),
|
|
call('addresses', Address(user=None), False),
|
|
call('user', User(addresses=[]), False),
|
|
call('user', User(addresses=[]), True),
|
|
call('addresses', Address(user=None), True)
|
|
])
|
|
else:
|
|
eq_(calls,
|
|
[
|
|
call('addresses', Address()),
|
|
call('addresses', Address(user=None)),
|
|
call('user', User(addresses=[]))
|
|
])
|