# copyright 2003-2010 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

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

#

# This file is part of CubicWeb.

#

# CubicWeb is free software: you can redistribute it and/or modify it under the

# terms of the GNU Lesser General Public License as published by the Free

# Software Foundation, either version 2.1 of the License, or (at your option)

# any later version.

#

# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

# FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more

# details.

#

# You should have received a copy of the GNU Lesser General Public License along

# with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.

"""

"""

from cubicweb.devtools import init_test_database

from cubicweb.devtools.repotest import BasePlannerTC, test_plan

class _SetGenerator(object):

    """singleton to easily create set using "s[0]" or "s[0,1,2]" for instance

    """

    def __getitem__(self, key):

        try:

            it = iter(key)

        except TypeError:

            it = (key,)

        return set(it)

s = _SetGenerator()

from cubicweb.schema import ERQLExpression

from cubicweb.server.sources import AbstractSource

from cubicweb.server.msplanner import MSPlanner, PartPlanInformation

class FakeUserROSource(AbstractSource):

    uri = 'zzz'

    support_entities = {'CWUser': False}

    support_relations = {}

    def syntax_tree_search(self, *args, **kwargs):

        return []

class FakeCardSource(AbstractSource):

    uri = 'ccc'

    support_entities = {'Card': True, 'Note': True, 'State': True}

    support_relations = {'in_state': True, 'multisource_rel': True, 'multisource_inlined_rel': True,

                         'multisource_crossed_rel': True}

    dont_cross_relations = set(('fiche', 'state_of'))

    cross_relations = set(('multisource_crossed_rel',))

    def syntax_tree_search(self, *args, **kwargs):

        return []

X_ALL_SOLS = sorted([{'X': 'Affaire'}, {'X': 'BaseTransition'}, {'X': 'Basket'},

                     {'X': 'Bookmark'}, {'X': 'CWAttribute'}, {'X': 'CWCache'},

                     {'X': 'CWConstraint'}, {'X': 'CWConstraintType'}, {'X': 'CWEType'},

                     {'X': 'CWGroup'}, {'X': 'CWPermission'}, {'X': 'CWProperty'},

                     {'X': 'CWRType'}, {'X': 'CWRelation'}, {'X': 'CWUser'},

                     {'X': 'Card'}, {'X': 'Comment'}, {'X': 'Division'},

                     {'X': 'Email'}, {'X': 'EmailAddress'}, {'X': 'EmailPart'},

                     {'X': 'EmailThread'}, {'X': 'ExternalUri'}, {'X': 'File'},

                     {'X': 'Folder'}, {'X': 'Image'}, {'X': 'Note'},

                     {'X': 'Personne'}, {'X': 'RQLExpression'}, {'X': 'Societe'},

                     {'X': 'State'}, {'X': 'SubDivision'}, {'X': 'SubWorkflowExitPoint'},

                     {'X': 'Tag'}, {'X': 'TrInfo'}, {'X': 'Transition'},

                     {'X': 'Workflow'}, {'X': 'WorkflowTransition'}])

# keep cnx so it's not garbage collected and the associated session is closed

repo, cnx = init_test_database()

def teardown_module(*args):

    global repo, cnx

    del repo, cnx

class BaseMSPlannerTC(BasePlannerTC):

    """test planner related feature on a 3-sources repository:

    * system source supporting everything

    * ldap source supporting CWUser

    * rql source supporting Card

    """

    repo = repo

    def setUp(self):

        #_QuerierTC.setUp(self)

        self.setup()

        # hijack Affaire security

        affreadperms = list(self.schema['Affaire'].permissions['read'])

        self.prevrqlexpr_affaire = affreadperms[-1]

        # add access to type attribute so S can't be invariant

        affreadperms[-1] = ERQLExpression('X concerne S?, S owned_by U, S type "X"')

        self.schema['Affaire'].set_action_permissions('read', affreadperms)

        # hijack CWUser security

        userreadperms = list(self.schema['CWUser'].permissions['read'])

        self.prevrqlexpr_user = userreadperms[-1]

        userreadperms[-1] = ERQLExpression('X owned_by U')

        self.schema['CWUser'].set_action_permissions('read', userreadperms)

        self.add_source(FakeUserROSource, 'ldap')

        self.add_source(FakeCardSource, 'cards')

    def tearDown(self):

        # restore hijacked security

        self.restore_orig_affaire_security()

        self.restore_orig_cwuser_security()

        super(BaseMSPlannerTC, self).tearDown()

    def restore_orig_affaire_security(self):

        affreadperms = list(self.schema['Affaire'].permissions['read'])

        affreadperms[-1] = self.prevrqlexpr_affaire

        self.schema['Affaire'].set_action_permissions('read', affreadperms)

    def restore_orig_cwuser_security(self):

        if hasattr(self, '_orig_cwuser_security_restored'):

            return

        self._orig_cwuser_security_restored = True

        userreadperms = list(self.schema['CWUser'].permissions['read'])

        userreadperms[-1] = self.prevrqlexpr_user

        self.schema['CWUser'].set_action_permissions('read', userreadperms)

class PartPlanInformationTC(BaseMSPlannerTC):

    def _test(self, rql, *args):

        if len(args) == 3:

            kwargs, sourcesterms, needsplit = args

        else:

            sourcesterms, needsplit = args

            kwargs = None

        plan = self._prepare_plan(rql, kwargs)

        union = plan.rqlst

        plan.preprocess(union)

        ppi = PartPlanInformation(plan, union.children[0])

        for sourcevars in ppi._sourcesterms.itervalues():

            for var in sourcevars.keys():

                solindices = sourcevars.pop(var)

                sourcevars[var._ms_table_key()] = solindices

        self.assertEquals(ppi._sourcesterms, sourcesterms)

        self.assertEquals(ppi.needsplit, needsplit)

    def test_simple_system_only(self):

        """retrieve entities only supported by the system source"""

        self._test('CWGroup X',

                   {self.system: {'X': s[0]}}, False)

    def test_simple_system_ldap(self):

        """retrieve CWUser X from both sources and return concatenation of results

        """

        self._test('CWUser X',

                   {self.system: {'X': s[0]}, self.ldap: {'X': s[0]}}, False)

    def test_simple_system_rql(self):

        """retrieve Card X from both sources and return concatenation of results

        """

        self._test('Any X, XT WHERE X is Card, X title XT',

                   {self.system: {'X': s[0]}, self.cards: {'X': s[0]}}, False)

    def test_simple_eid_specified(self):

        """retrieve CWUser X from system source (eid is specified, can locate the entity)

        """

        ueid = self.session.user.eid

        self._test('Any X,L WHERE X eid %(x)s, X login L', {'x': ueid},

                   {self.system: {'X': s[0]}}, False)

    def test_simple_eid_invariant(self):

        """retrieve CWUser X from system source (eid is specified, can locate the entity)

        """

        ueid = self.session.user.eid

        self._test('Any X WHERE X eid %(x)s', {'x': ueid},

                   {self.system: {'x': s[0]}}, False)

    def test_simple_invariant(self):

        """retrieve CWUser X from system source only (X is invariant and in_group not supported by ldap source)

        """

        self._test('Any X WHERE X is CWUser, X in_group G, G name "users"',

                   {self.system: {'X': s[0], 'G': s[0], 'in_group': s[0]}}, False)

    def test_security_has_text(self):

        """retrieve CWUser X from system source only (has_text not supported by ldap source)

        """

        # specify CWUser instead of any since the way this test is written we aren't well dealing

        # with ambigous query (eg only considering the first solution)

        self._test('CWUser X WHERE X has_text "bla"',

                   {self.system: {'X': s[0]}}, False)

    def test_complex_base(self):

        """

        1. retrieve Any X, L WHERE X is CWUser, X login L from system and ldap sources, store

           concatenation of results into a temporary table

        2. return the result of Any X, L WHERE X is TMP, X login L, X in_group G,

           G name 'users' on the system source

        """

        self._test('Any X,L WHERE X is CWUser, X in_group G, X login L, G name "users"',

                   {self.system: {'X': s[0], 'G': s[0], 'in_group': s[0]},

                    self.ldap : {'X': s[0]}}, True)

    def test_complex_invariant_ordered(self):

        """

        1. retrieve Any X,AA WHERE X modification_date AA from system and ldap sources, store

           concatenation of results into a temporary table

        2. return the result of Any X,AA ORDERBY AA WHERE %s owned_by X, X modification_date AA

           on the system source

        """

        ueid = self.session.user.eid

        self._test('Any X,AA ORDERBY AA WHERE E eid %(x)s, E owned_by X, X modification_date AA', {'x': ueid},

                   {self.system: {'x': s[0], 'X': s[0], 'owned_by': s[0]},

                    self.ldap : {'X': s[0]}}, True)

    def test_complex_invariant(self):

        """

        1. retrieve Any X,L,AA WHERE X login L, X modification_date AA from system and ldap sources, store

           concatenation of results into a temporary table

        2. return the result of Any X,L,AA WHERE %s owned_by X, X login L, X modification_date AA

           on the system source

        """

        ueid = self.session.user.eid

        self._test('Any X,L,AA WHERE E eid %(x)s, E owned_by X, X login L, X modification_date AA', {'x': ueid},

                   {self.system: {'x': s[0], 'X': s[0], 'owned_by': s[0]},

                    self.ldap : {'X': s[0]}}, True)

    def test_complex_ambigous(self):

        """retrieve CWUser X from system and ldap sources, Person X from system source only

        """

        self._test('Any X,F WHERE X firstname F',

                   {self.system: {'X': s[0, 1]},

                    self.ldap: {'X': s[0]}}, True)

    def test_complex_multiple(self):

        """

        1. retrieve Any X,A,Y,B WHERE X login A, Y login B from system and ldap sources, store

           cartesian product of results into a temporary table

        2. return the result of Any X,Y WHERE X login 'syt', Y login 'adim'

           on the system source

        """

        ueid = self.session.user.eid

        self._test('Any X,Y WHERE X login "syt", Y login "adim"', {'x': ueid},

                   {self.system: {'Y': s[0], 'X': s[0]},

                    self.ldap: {'Y': s[0], 'X': s[0]}}, True)

    def test_complex_aggregat(self):

        solindexes = set(range(len([e for e in self.schema.entities() if not e.final])))

        self._test('Any MAX(X)',

                   {self.system: {'X': solindexes}}, False)

    def test_complex_optional(self):

        ueid = self.session.user.eid

        self._test('Any U WHERE WF wf_info_for X, X eid %(x)s, WF owned_by U?, WF from_state FS', {'x': ueid},

                   {self.system: {'WF': s[0], 'FS': s[0], 'U': s[0],

                                  'from_state': s[0], 'owned_by': s[0], 'wf_info_for': s[0],

                                  'x': s[0]}},

                   False)

    def test_exists4(self):

        """

        State S could come from both rql source and system source,

        but since X cannot come from the rql source, the solution

        {self.cards : 'S'} must be removed

        """

        self._test('Any G,L WHERE X in_group G, X login L, G name "managers", '

                   'EXISTS(X copain T, T login L, T login in ("comme", "cochon")) OR '

                   'EXISTS(X in_state S, S name "pascontent", NOT X copain T2, T2 login "billy")',

                   {self.system: {'X': s[0], 'S': s[0], 'T2': s[0], 'T': s[0], 'G': s[0], 'copain': s[0], 'in_group': s[0]},

                    self.ldap: {'X': s[0], 'T2': s[0], 'T': s[0]}},

                   True)

    def test_relation_need_split(self):

        self._test('Any X, S WHERE X in_state S',

                   {self.system: {'X': s[0, 1, 2], 'S': s[0, 1, 2]},

                    self.cards: {'X': s[2], 'S': s[2]}},

                   True)

    def test_not_relation_need_split(self):

        self._test('Any SN WHERE NOT X in_state S, S name SN',

                   {self.cards: {'X': s[2], 'S': s[0, 1, 2]},

                    self.system: {'X': s[0, 1, 2], 'S': s[0, 1, 2]}},

                   True)

    def test_not_relation_no_split_external(self):

        repo._type_source_cache[999999] = ('Note', 'cards', 999999)

        # similar to the above test but with an eid coming from the external source.

        # the same plan may be used, since we won't find any record in the system source

        # linking 9999999 to a state

        self._test('Any SN WHERE NOT X in_state S, X eid %(x)s, S name SN',

                   {'x': 999999},

                   {self.cards: {'x': s[0], 'S': s[0]},

                    self.system: {'x': s[0], 'S': s[0]}},

                   False)

    def test_relation_restriction_ambigous_need_split(self):

        self._test('Any X,T WHERE X in_state S, S name "pending", T tags X',

                   {self.system: {'X': s[0, 1, 2], 'S': s[0, 1, 2], 'T': s[0, 1, 2], 'tags': s[0, 1, 2]},

                    self.cards: {'X': s[2], 'S': s[2]}},

                   True)

    def test_simplified_var(self):

        repo._type_source_cache[999999] = ('Note', 'cards', 999999)

        self._test('Any U WHERE U in_group G, (G name IN ("managers", "logilab") OR (X require_permission P?, P name "bla", P require_group G)), X eid %(x)s, U eid %(u)s',

                   {'x': 999999, 'u': self.session.user.eid},

                   {self.system: {'P': s[0], 'G': s[0], 'X': s[0],

                                  'require_permission': s[0], 'in_group': s[0], 'P': s[0], 'require_group': s[0],

                                  'u': s[0]}},

                   False)

    def test_delete_relation1(self):

        ueid = self.session.user.eid

        self._test('Any X, Y WHERE X created_by Y, X eid %(x)s, NOT Y eid %(y)s',

                   {'x': ueid, 'y': ueid},

                   {self.system: {'Y': s[0], 'created_by': s[0], 'x': s[0]}},

                   False)

    def test_crossed_relation_eid_1_needattr(self):

        repo._type_source_cache[999999] = ('Note', 'system', 999999)

        ueid = self.session.user.eid

        self._test('Any Y,T WHERE X eid %(x)s, X multisource_crossed_rel Y, Y type T',

                   {'x': 999999,},

                   {self.cards: {'Y': s[0]}, self.system: {'Y': s[0], 'x': s[0]}},

                   True)

    def test_crossed_relation_eid_1_invariant(self):

        repo._type_source_cache[999999] = ('Note', 'system', 999999)

        self._test('Any Y WHERE X eid %(x)s, X multisource_crossed_rel Y',

                   {'x': 999999},

                   {self.system: {'Y': s[0], 'x': s[0]}},

                   False)

    def test_crossed_relation_eid_2_invariant(self):

        repo._type_source_cache[999999] = ('Note', 'cards', 999999)

        self._test('Any Y WHERE X eid %(x)s, X multisource_crossed_rel Y',

                   {'x': 999999,},

                   {self.cards: {'Y': s[0], 'multisource_crossed_rel': s[0], 'x': s[0]},

                    self.system: {'Y': s[0], 'multisource_crossed_rel': s[0], 'x': s[0]}},

                   False)

    def test_version_crossed_depends_on_1(self):

        repo._type_source_cache[999999] = ('Note', 'cards', 999999)

        self._test('Any X,AD,AE WHERE E eid %(x)s, E multisource_crossed_rel X, X in_state AD, AD name AE',

                   {'x': 999999},

                   {self.cards: {'X': s[0], 'AD': s[0], 'multisource_crossed_rel': s[0], 'x': s[0]},

                    self.system: {'X': s[0], 'AD': s[0], 'multisource_crossed_rel': s[0], 'x': s[0]}},

                   True)

    def test_version_crossed_depends_on_2(self):

        repo._type_source_cache[999999] = ('Note', 'system', 999999)

        self._test('Any X,AD,AE WHERE E eid %(x)s, E multisource_crossed_rel X, X in_state AD, AD name AE',

                   {'x': 999999},

                   {self.cards: {'X': s[0], 'AD': s[0]},

                    self.system: {'X': s[0], 'AD': s[0], 'x': s[0]}},

                    True)

    def test_simplified_var_3(self):

        repo._type_source_cache[999999] = ('Note', 'cards', 999999)

        repo._type_source_cache[999998] = ('State', 'cards', 999998)

        self._test('Any S,T WHERE S eid %(s)s, N eid %(n)s, N type T, N is Note, S is State',

                   {'n': 999999, 's': 999998},

                   {self.cards: {'s': s[0], 'N': s[0]}}, False)

class MSPlannerTC(BaseMSPlannerTC):

    def setUp(self):

        BaseMSPlannerTC.setUp(self)

        self.planner = MSPlanner(self.o.schema, self.repo.vreg.rqlhelper)

    _test = test_plan

    def test_simple_system_only(self):

        """retrieve entities only supported by the system source

        """

        self._test('CWGroup X',

                   [('OneFetchStep', [('Any X WHERE X is CWGroup', [{'X': 'CWGroup'}])],

                     None, None, [self.system], {}, [])])

    def test_simple_system_only_limit(self):

        """retrieve entities only supported by the system source

        """

        self._test('CWGroup X LIMIT 10',

                   [('OneFetchStep', [('Any X LIMIT 10 WHERE X is CWGroup', [{'X': 'CWGroup'}])],

                     10, None, [self.system], {}, [])])

    def test_simple_system_only_limit_offset(self):

        """retrieve entities only supported by the system source

        """

        self._test('CWGroup X LIMIT 10 OFFSET 10',

                   [('OneFetchStep', [('Any X LIMIT 10 OFFSET 10 WHERE X is CWGroup', [{'X': 'CWGroup'}])],

                     10, 10, [self.system], {}, [])])

    def test_simple_system_ldap(self):

        """retrieve CWUser X from both sources and return concatenation of results

        """

        self._test('CWUser X',

                   [('OneFetchStep', [('Any X WHERE X is CWUser', [{'X': 'CWUser'}])],

                     None, None, [self.ldap, self.system], {}, [])])

    def test_simple_system_ldap_limit(self):

        """retrieve CWUser X from both sources and return concatenation of results

        """

        self._test('CWUser X LIMIT 10',

                   [('OneFetchStep', [('Any X LIMIT 10 WHERE X is CWUser', [{'X': 'CWUser'}])],

                     10, None, [self.ldap, self.system], {}, [])])

    def test_simple_system_ldap_limit_offset(self):

        """retrieve CWUser X from both sources and return concatenation of results

        """

        self._test('CWUser X LIMIT 10 OFFSET 10',

                   [('OneFetchStep', [('Any X LIMIT 10 OFFSET 10 WHERE X is CWUser', [{'X': 'CWUser'}])],

                     10, 10, [self.ldap, self.system], {}, [])])

    def test_simple_system_ldap_ordered_limit_offset(self):

        """retrieve CWUser X from both sources and return concatenation of results

        """

        self._test('CWUser X ORDERBY X LIMIT 10 OFFSET 10',

                   [('AggrStep', 'Any X ORDERBY X', 10, 10, 'table0', None, [

                       ('FetchStep', [('Any X WHERE X is CWUser', [{'X': 'CWUser'}])],

                        [self.ldap, self.system], {}, {'X': 'table0.C0'}, []),

                       ]),

                   ])

    def test_simple_system_ldap_aggregat(self):

        """retrieve CWUser X from both sources and return concatenation of results

        """

        # COUNT(X) is kept in sub-step and transformed into SUM(X) in the AggrStep

        self._test('Any COUNT(X) WHERE X is CWUser',

                   [('AggrStep', 'Any COUNT(X)', None, None, 'table0', None, [

                       ('FetchStep', [('Any COUNT(X) WHERE X is CWUser', [{'X': 'CWUser'}])],

                        [self.ldap, self.system], {}, {'COUNT(X)': 'table0.C0'}, []),

                       ]),

                   ])

    def test_simple_system_rql(self):

        """retrieve Card X from both sources and return concatenation of results

        """

        self._test('Any X, XT WHERE X is Card, X title XT',

                   [('OneFetchStep', [('Any X,XT WHERE X is Card, X title XT', [{'X': 'Card', 'XT': 'String'}])],

                     None, None, [self.cards, self.system], {}, [])])

    def test_simple_eid_specified(self):

        """retrieve CWUser X from system source (eid is specified, can locate the entity)

        """

        ueid = self.session.user.eid

        self._test('Any X,L WHERE X eid %(x)s, X login L',

                   [('OneFetchStep', [('Any X,L WHERE X eid %s, X login L'%ueid, [{'X': 'CWUser', 'L': 'String'}])],

                     None, None, [self.system], {}, [])],

                   {'x': ueid})

    def test_simple_eid_invariant(self):

        """retrieve CWUser X from system source (eid is specified, can locate the entity)

        """

        ueid = self.session.user.eid

        self._test('Any X WHERE X eid %(x)s',

                   [('OneFetchStep', [('Any %s'%ueid, [{}])],

                     None, None, [self.system], {}, [])],

                   {'x': ueid})

    def test_simple_invariant(self):

        """retrieve CWUser X from system source only (X is invariant and in_group not supported by ldap source)

        """

        self._test('Any X WHERE X is CWUser, X in_group G, G name "users"',

                   [('OneFetchStep', [('Any X WHERE X is CWUser, X in_group G, G name "users"',

                                       [{'X': 'CWUser', 'G': 'CWGroup'}])],

                     None, None, [self.system], {}, [])])

    def test_complex_base(self):

        """

        1. retrieve Any X, L WHERE X is CWUser, X login L from system and ldap sources, store

           concatenation of results into a temporary table

        2. return the result of Any X, L WHERE X is TMP, X login LX in_group G,

           G name 'users' on the system source

        """

        self._test('Any X,L WHERE X is CWUser, X in_group G, X login L, G name "users"',

                   [('FetchStep', [('Any X,L WHERE X login L, X is CWUser', [{'X': 'CWUser', 'L': 'String'}])],

                     [self.ldap, self.system], None,

                     {'X': 'table0.C0', 'X.login': 'table0.C1', 'L': 'table0.C1'}, []),

                    ('OneFetchStep', [('Any X,L WHERE X in_group G, X login L, G name "users", G is CWGroup, X is CWUser',

                                       [{'X': 'CWUser', 'L': 'String', 'G': 'CWGroup'}])],

                     None, None, [self.system],

                     {'X': 'table0.C0', 'X.login': 'table0.C1', 'L': 'table0.C1'}, [])

                    ])

    def test_complex_base_limit_offset(self):

        """

        1. retrieve Any X, L WHERE X is CWUser, X login L from system and ldap sources, store

           concatenation of results into a temporary table

        2. return the result of Any X, L WHERE X is TMP, X login LX in_group G,

           G name 'users' on the system source

        """

        self._test('Any X,L LIMIT 10 OFFSET 10 WHERE X is CWUser, X in_group G, X login L, G name "users"',

                   [('FetchStep', [('Any X,L WHERE X login L, X is CWUser', [{'X': 'CWUser', 'L': 'String'}])],

                     [self.ldap, self.system], None,

                     {'X': 'table0.C0', 'X.login': 'table0.C1', 'L': 'table0.C1'}, []),

                    ('OneFetchStep', [('Any X,L LIMIT 10 OFFSET 10 WHERE X in_group G, X login L, G name "users", G is CWGroup, X is CWUser',