from unittest.mock import call
from unittest.mock import Mock

from sqlalchemy import exc
from sqlalchemy import testing
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 ne_
from sqlalchemy.testing.entities import ComparableEntity
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(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(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(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(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(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(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(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,
        )

    @testing.variation("include_backrefs", [True, False, "default"])
    @testing.variation("include_removes", [True, False, "default"])
    def test_validator_backrefs(self, include_backrefs, include_removes):
        users, addresses = (self.tables.users, self.tables.addresses)
        canary = Mock()

        need_remove_param = (
            bool(include_removes) and not include_removes.default
        )
        validate_kw = {}
        if not include_removes.default:
            validate_kw["include_removes"] = bool(include_removes)
        if not include_backrefs.default:
            validate_kw["include_backrefs"] = bool(include_backrefs)

        expect_include_backrefs = include_backrefs.default or bool(
            include_backrefs
        )
        expect_include_removes = (
            bool(include_removes) and not include_removes.default
        )

        class User(ComparableEntity):
            if need_remove_param:

                @validates("addresses", **validate_kw)
                def validate_address(self, key, item, remove):
                    canary(key, item, remove)
                    return item

            else:

                @validates("addresses", **validate_kw)
                def validate_address(self, key, item):
                    canary(key, item)
                    return item

        class Address(ComparableEntity):
            if need_remove_param:

                @validates("user", **validate_kw)
                def validate_user(self, key, item, remove):
                    canary(key, item, remove)
                    return item

            else:

                @validates("user", **validate_kw)
                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 expect_include_backrefs:
            if expect_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 expect_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=[])),
                    ],
                )