from unittest.mock import call from unittest.mock import Mock from sqlalchemy import exc from sqlalchemy.orm import collections from sqlalchemy.orm import relationship from sqlalchemy.orm import validates from sqlalchemy.testing import assert_raises from sqlalchemy.testing import assert_raises_message from sqlalchemy.testing import eq_ from sqlalchemy.testing import fixtures from sqlalchemy.testing import ne_ from sqlalchemy.testing.fixtures import fixture_session from test.orm import _fixtures 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" self.mapper_registry.map_imperatively(User, users) sess = fixture_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 self.mapper_registry.map_imperatively( User, users, properties={"addresses": relationship(Address)} ) self.mapper_registry.map_imperatively(Address, addresses) sess = fixture_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 = self.mapper_registry.map_imperatively( User, users, properties={"addresses": relationship(Address)} ) self.mapper_registry.map_imperatively(Address, addresses) eq_( {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 self.mapper_registry.map_imperatively( User, users, properties={"addresses": relationship(Address)} ) self.mapper_registry.map_imperatively(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. # in 1.2 due to #3896, we also get 'a2' in the # validates as it is part of the set call("addresses", a2, False), call("addresses", a3, False), call("addresses", a1, True), ], ) def test_validator_bulk_collection_set(self): users, addresses, Address = ( self.tables.users, self.tables.addresses, self.classes.Address, ) class User(fixtures.ComparableEntity): @validates("addresses", include_removes=True) def validate_address(self, key, item, remove): if not remove: assert isinstance(item, str) else: assert isinstance(item, Address) item = Address(email_address=item) return item self.mapper_registry.map_imperatively( User, users, properties={"addresses": relationship(Address)} ) self.mapper_registry.map_imperatively(Address, addresses) u1 = User() u1.addresses.append("e1") u1.addresses.append("e2") eq_( u1.addresses, [Address(email_address="e1"), Address(email_address="e2")], ) u1.addresses = ["e3", "e4"] eq_( u1.addresses, [Address(email_address="e3"), Address(email_address="e4")], ) def test_validator_bulk_dict_set(self): users, addresses, Address = ( self.tables.users, self.tables.addresses, self.classes.Address, ) class User(fixtures.ComparableEntity): @validates("addresses", include_removes=True) def validate_address(self, key, item, remove): if not remove: assert isinstance(item, str) else: assert isinstance(item, Address) item = Address(email_address=item) return item self.mapper_registry.map_imperatively( User, users, properties={ "addresses": relationship( Address, collection_class=collections.attribute_keyed_dict( "email_address" ), ) }, ) self.mapper_registry.map_imperatively(Address, addresses) u1 = User() u1.addresses["e1"] = "e1" u1.addresses["e2"] = "e2" eq_( u1.addresses, { "e1": Address(email_address="e1"), "e2": Address(email_address="e2"), }, ) u1.addresses = {"e3": "e3", "e4": "e4"} eq_( u1.addresses, { "e3": Address(email_address="e3"), "e4": Address(email_address="e4"), }, ) def test_validator_as_callable_object(self): """test #6538""" users = self.tables.users canary = Mock() class SomeValidator: def __call__(self, obj, key, name): canary(key, name) ne_(name, "fred") return name + " modified" class User(fixtures.ComparableEntity): sv = validates("name")(SomeValidator()) self.mapper_registry.map_imperatively(User, users) u1 = User(name="ed") eq_(u1.name, "ed modified") def test_validator_multi_warning(self): users = self.tables.users class Foo: @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", self.mapper_registry.map_imperatively, Foo, users, ) class Bar: @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", self.mapper_registry.map_imperatively, 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 self.mapper_registry.map_imperatively( User, users, properties={"addresses": relationship(Address, backref="user")}, ) self.mapper_registry.map_imperatively(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=[])), ], )