# -*- coding: utf-8 -*-

"""unit tests for cubicweb.entities.base module

:organization: Logilab

:copyright: 2001-2009 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2.

:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr

:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses

"""

from logilab.common.testlib import unittest_main

from logilab.common.decorators import clear_cache

from logilab.common.interface import implements

from cubicweb.devtools.apptest import EnvBasedTC

from cubicweb import ValidationError

from cubicweb.interfaces import IMileStone, IWorkflowable

from cubicweb.entities import AnyEntity

from cubicweb.entities.authobjs import CWUser

from cubicweb.web.widgets import AutoCompletionWidget

class BaseEntityTC(EnvBasedTC):

    def setup_database(self):

        self.member = self.create_user('member')

class MetadataTC(BaseEntityTC):

    def test_creator(self):

        self.login(u'member')

        entity = self.add_entity('Bookmark', title=u"hello", path=u'project/cubicweb')

        self.commit()

        self.assertEquals(entity.creator.eid, self.member.eid)

        self.assertEquals(entity.dc_creator(), u'member')

    def test_type(self):

        self.assertEquals(self.member.dc_type(), 'cwuser')

    def test_entity_meta_attributes(self):

        # XXX move to yams

        self.assertEquals(self.schema['CWUser'].meta_attributes(), {})

        self.assertEquals(dict((str(k), v) for k, v in self.schema['State'].meta_attributes().iteritems()),

                          {'description_format': ('format', 'description')})

class CWUserTC(BaseEntityTC):

    def test_dc_title_and_name(self):

        e = self.entity('CWUser U WHERE U login "member"')

        self.assertEquals(e.dc_title(), 'member')

        self.assertEquals(e.name(), 'member')

        self.execute(u'SET X firstname "bouah" WHERE X is CWUser, X login "member"')

        self.assertEquals(e.dc_title(), 'member')

        self.assertEquals(e.name(), u'bouah')

        self.execute(u'SET X surname "lôt" WHERE X is CWUser, X login "member"')

        self.assertEquals(e.dc_title(), 'member')

        self.assertEquals(e.name(), u'bouah lôt')

class StateAndTransitionsTC(BaseEntityTC):

    def test_transitions(self):

        user = self.entity('CWUser X')

        e = self.entity('State S WHERE S name "activated"')

        trs = list(e.transitions(user))

        self.assertEquals(len(trs), 1)

        self.assertEquals(trs[0].name, u'deactivate')

        self.assertEquals(trs[0].destination().name, u'deactivated')

        self.assert_(user.can_pass_transition('deactivate'))

        self.assert_(not user.can_pass_transition('activate'))

        # test a std user get no possible transition

        self.login('member')

        # fetch the entity using the new session

        e = self.entity('State S WHERE S name "activated"')

        trs = list(e.transitions(user))

        self.assertEquals(len(trs), 0)

        user = self.entity('CWUser X')

        self.assert_(not user.can_pass_transition('deactivate'))

        self.assert_(not user.can_pass_transition('activate'))

    def test_transitions_with_dest_specfied(self):

        user = self.entity('CWUser X')

        e = self.entity('State S WHERE S name "activated"')

        e2 = self.entity('State S WHERE S name "deactivated"')

        trs = list(e.transitions(user, e2.eid))

        self.assertEquals(len(trs), 1)

        self.assertEquals(trs[0].name, u'deactivate')

        self.assertEquals(trs[0].destination().name, u'deactivated')

        trs = list(e.transitions(user, e.eid))

        self.assertEquals(len(trs), 0)

    def test_transitions_maybe_passed(self):

        self.execute('INSERT RQLExpression X: X exprtype "ERQLExpression", '

                     'X expression "X owned_by U", T condition X '

                     'WHERE T name "deactivate"')

        self._test_deactivated()

    def test_transitions_maybe_passed_using_has_update_perm(self):

        self.execute('INSERT RQLExpression X: X exprtype "ERQLExpression", '

                     'X expression "U has_update_permission X", T condition X '

                     'WHERE T name "deactivate"')

        self._test_deactivated()

    def _test_deactivated(self):

        ueid = self.create_user('toto').eid

        self.create_user('tutu')

        cnx = self.login('tutu')

        cu = cnx.cursor()

        self.assertRaises(ValidationError,

                          cu.execute, 'SET X in_state S WHERE X eid %(x)s, S name "deactivated"',

                          {'x': ueid}, 'x')

        cnx.close()

        cnx = self.login('toto')

        cu = cnx.cursor()

        cu.execute('SET X in_state S WHERE X eid %(x)s, S name "deactivated"',

                   {'x': ueid}, 'x')

        cnx.commit()

        self.assertRaises(ValidationError,

                          cu.execute, 'SET X in_state S WHERE X eid %(x)s, S name "activated"',

                          {'x': ueid}, 'x')

    def test_transitions_selection(self):

        """

        ------------------------  tr1    -----------------

        | state1 (CWGroup, Bookmark) | ------> | state2 (CWGroup) |

        ------------------------         -----------------

                  |  tr2    ------------------

                  `------>  | state3 (Bookmark) |

                            ------------------

        """

        state1 = self.add_entity('State', name=u'state1')

        state2 = self.add_entity('State', name=u'state2')

        state3 = self.add_entity('State', name=u'state3')

        tr1 = self.add_entity('Transition', name=u'tr1')

        tr2 = self.add_entity('Transition', name=u'tr2')

        self.execute('SET X state_of Y WHERE X eid in (%s, %s), Y is CWEType, Y name "CWGroup"' %

                      (state1.eid, state2.eid))

        self.execute('SET X state_of Y WHERE X eid in (%s, %s), Y is CWEType, Y name "Bookmark"' %

                      (state1.eid, state3.eid))

        self.execute('SET X transition_of Y WHERE X eid %s, Y name "CWGroup"' % tr1.eid)

        self.execute('SET X transition_of Y WHERE X eid %s, Y name "Bookmark"' % tr2.eid)

        self.execute('SET X allowed_transition Y WHERE X eid %s, Y eid %s' %

                      (state1.eid, tr1.eid))

        self.execute('SET X allowed_transition Y WHERE X eid %s, Y eid %s' %

                      (state1.eid, tr2.eid))

        self.execute('SET X destination_state Y WHERE X eid %s, Y eid %s' %

                      (tr1.eid, state2.eid))

        self.execute('SET X destination_state Y WHERE X eid %s, Y eid %s' %

                      (tr2.eid, state3.eid))

        self.execute('SET X initial_state Y WHERE Y eid %s, X name "CWGroup"' % state1.eid)

        self.execute('SET X initial_state Y WHERE Y eid %s, X name "Bookmark"' % state1.eid)

        group = self.add_entity('CWGroup', name=u't1')

        transitions = list(state1.transitions(group))

        self.assertEquals(len(transitions), 1)

        self.assertEquals(transitions[0].name, 'tr1')

        bookmark = self.add_entity('Bookmark', title=u'111', path=u'/view')

        transitions = list(state1.transitions(bookmark))

        self.assertEquals(len(transitions), 1)

        self.assertEquals(transitions[0].name, 'tr2')

    def test_transitions_selection2(self):

        """

        ------------------------  tr1 (Bookmark)   -----------------------

        | state1 (CWGroup, Bookmark) | -------------> | state2 (CWGroup,Bookmark) |

        ------------------------                -----------------------

                  |  tr2 (CWGroup)                     |

                  `---------------------------------/

        """

        state1 = self.add_entity('State', name=u'state1')

        state2 = self.add_entity('State', name=u'state2')

        tr1 = self.add_entity('Transition', name=u'tr1')

        tr2 = self.add_entity('Transition', name=u'tr2')

        self.execute('SET X state_of Y WHERE X eid in (%s, %s), Y is CWEType, Y name "CWGroup"' %

                      (state1.eid, state2.eid))

        self.execute('SET X state_of Y WHERE X eid in (%s, %s), Y is CWEType, Y name "Bookmark"' %

                      (state1.eid, state2.eid))

        self.execute('SET X transition_of Y WHERE X eid %s, Y name "CWGroup"' % tr1.eid)

        self.execute('SET X transition_of Y WHERE X eid %s, Y name "Bookmark"' % tr2.eid)

        self.execute('SET X allowed_transition Y WHERE X eid %s, Y eid %s' %

                      (state1.eid, tr1.eid))

        self.execute('SET X allowed_transition Y WHERE X eid %s, Y eid %s' %

                      (state1.eid, tr2.eid))

        self.execute('SET X destination_state Y WHERE X eid %s, Y eid %s' %

                      (tr1.eid, state2.eid))

        self.execute('SET X destination_state Y WHERE X eid %s, Y eid %s' %

                      (tr2.eid, state2.eid))

        self.execute('SET X initial_state Y WHERE Y eid %s, X name "CWGroup"' % state1.eid)

        self.execute('SET X initial_state Y WHERE Y eid %s, X name "Bookmark"' % state1.eid)

        group = self.add_entity('CWGroup', name=u't1')

        transitions = list(state1.transitions(group))

        self.assertEquals(len(transitions), 1)

        self.assertEquals(transitions[0].name, 'tr1')

        bookmark = self.add_entity('Bookmark', title=u'111', path=u'/view')

        transitions = list(state1.transitions(bookmark))

        self.assertEquals(len(transitions), 1)

        self.assertEquals(transitions[0].name, 'tr2')

class EmailAddressTC(BaseEntityTC):

    def test_canonical_form(self):

        eid1 = self.execute('INSERT EmailAddress X: X address "maarten.ter.huurne@philips.com"')[0][0]

        eid2 = self.execute('INSERT EmailAddress X: X address "maarten@philips.com", X canonical TRUE')[0][0]

        self.execute('SET X identical_to Y WHERE X eid %s, Y eid %s' % (eid1, eid2))

        email1 = self.entity('Any X WHERE X eid %(x)s', {'x':eid1}, 'x')

        email2 = self.entity('Any X WHERE X eid %(x)s', {'x':eid2}, 'x')

        self.assertEquals(email1.canonical_form().eid, eid2)

        self.assertEquals(email2.canonical_form(), email2)

        eid3 = self.execute('INSERT EmailAddress X: X address "toto@logilab.fr"')[0][0]

        email3 = self.entity('Any X WHERE X eid %s'%eid3)

        self.assertEquals(email3.canonical_form(), None)

    def test_mangling(self):

        eid = self.execute('INSERT EmailAddress X: X address "maarten.ter.huurne@philips.com"')[0][0]

        email = self.entity('Any X WHERE X eid %(x)s', {'x':eid}, 'x')

        self.assertEquals(email.display_address(), 'maarten.ter.huurne@philips.com')

        self.assertEquals(email.printable_value('address'), 'maarten.ter.huurne@philips.com')

        self.vreg.config.global_set_option('mangle-emails', True)

        self.assertEquals(email.display_address(), 'maarten.ter.huurne at philips dot com')

        self.assertEquals(email.printable_value('address'), 'maarten.ter.huurne at philips dot com')

        eid = self.execute('INSERT EmailAddress X: X address "syt"')[0][0]

        email = self.entity('Any X WHERE X eid %(x)s', {'x':eid}, 'x')

        self.assertEquals(email.display_address(), 'syt')

        self.assertEquals(email.printable_value('address'), 'syt')

class CWUserTC(BaseEntityTC):

    def test_complete(self):

        e = self.entity('CWUser X WHERE X login "admin"')

        e.complete()

    def test_matching_groups(self):

        e = self.entity('CWUser X WHERE X login "admin"')

        self.failUnless(e.matching_groups('managers'))

        self.failIf(e.matching_groups('xyz'))

        self.failUnless(e.matching_groups(('xyz', 'managers')))

        self.failIf(e.matching_groups(('xyz', 'abcd')))

    def test_workflow_base(self):

        e = self.create_user('toto')

        self.assertEquals(e.state, 'activated')

        activatedeid = self.execute('State X WHERE X name "activated"')[0][0]

        deactivatedeid = self.execute('State X WHERE X name "deactivated"')[0][0]

        e.change_state(deactivatedeid, u'deactivate 1')

        self.commit()

        e.change_state(activatedeid, u'activate 1')

        self.commit()

        e.change_state(deactivatedeid, u'deactivate 2')

        self.commit()

        # get a fresh user to avoid potential cache issues

        e = self.entity('CWUser X WHERE X eid %s' % e.eid)

        self.assertEquals([tr.comment for tr in e.reverse_wf_info_for],

                          [None, 'deactivate 1', 'activate 1', 'deactivate 2'])

        self.assertEquals(e.latest_trinfo().comment, 'deactivate 2')

class InterfaceTC(EnvBasedTC):

    def test_nonregr_subclasses_and_mixins_interfaces(self):

        class MyUser(CWUser):

            __implements__ = (IMileStone,)

        self.vreg._loadedmods[__name__] = {}

        self.vreg.register_vobject_class(MyUser)

        self.failUnless(implements(CWUser, IWorkflowable))

        self.failUnless(implements(MyUser, IMileStone))

        self.failUnless(implements(MyUser, IWorkflowable))

class SpecializedEntityClassesTC(EnvBasedTC):

    def select_eclass(self, etype):

        # clear selector cache

        clear_cache(self.vreg, 'etype_class')

        return self.vreg.etype_class(etype)

    def test_etype_class_selection_and_specialization(self):

        # no specific class for Subdivisions, the default one should be selected

        eclass = self.select_eclass('SubDivision')

        self.failUnless(eclass.__autogenerated__)

        #self.assertEquals(eclass.__bases__, (AnyEntity,))

        # build class from most generic to most specific and make

        # sure the most specific is always selected

        self.vreg._loadedmods[__name__] = {}

        for etype in ('Company', 'Division', 'SubDivision'):

            class Foo(AnyEntity):

                id = etype

            self.vreg.register_vobject_class(Foo)

            eclass = self.select_eclass('SubDivision')

            if etype == 'SubDivision':

                self.failUnless(eclass is Foo)

            else:

                self.failUnless(eclass.__autogenerated__)

                self.assertEquals(eclass.__bases__, (Foo,))

        # check Division eclass is still selected for plain Division entities

        eclass = self.select_eclass('Division')

        self.assertEquals(eclass.id, 'Division')

if __name__ == '__main__':

    unittest_main()