from logilab.common.decorators import clear_cache

from cubicweb.devtools import init_test_database

from cubicweb.devtools.repotest import BasePlannerTC, do_monkey_patch, undo_monkey_patch, 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 = {'EUser': 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',))

    cross_relations = set(('multisource_crossed_rel',))

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

        return []

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

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

                     {'X': 'ECache'}, {'X': 'EConstraint'}, {'X': 'EConstraintType'},

                     {'X': 'EEType'}, {'X': 'EFRDef'}, {'X': 'EGroup'},

                     {'X': 'ENFRDef'}, {'X': 'EPermission'}, {'X': 'EProperty'},

                     {'X': 'ERType'}, {'X': 'EUser'}, {'X': 'Email'},

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

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

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

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

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

def clear_ms_caches(repo):

    clear_cache(repo, 'rel_type_sources')

    clear_cache(repo, 'can_cross_relation')

    clear_cache(repo, 'is_multi_sources_relation')

    # XXX source_defs

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

repo, cnx = init_test_database('sqlite')

class BaseMSPlannerTC(BasePlannerTC):

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

    * system source supporting everything

    * ldap source supporting EUser

    * rql source supporting Card

    """

    repo = repo

    def setUp(self):

        #_QuerierTC.setUp(self)

        clear_cache(repo, 'rel_type_sources')

        self.o = repo.querier

        self.session = repo._sessions.values()[0]

        self.pool = self.session.set_pool()

        self.schema = self.o.schema

        # hijack Affaire security

        affreadperms = list(self.schema['Affaire']._groups['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']._groups['read'] = tuple(affreadperms)

        # hijack EUser security

        userreadperms = list(self.schema['EUser']._groups['read'])

        self.prevrqlexpr_user = userreadperms[-1]

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

        self.schema['EUser']._groups['read'] = tuple(userreadperms)

        self.sources = self.o._repo.sources

        self.system = self.sources[-1]

        self.sources.append(FakeUserROSource(self.o._repo, self.o.schema,

                                             {'uri': 'ldapuser'}))

        repo.sources_by_uri['ldapuser'] = self.sources[-1]

        self.ldap = self.sources[-1]

        self.sources.append(FakeCardSource(self.o._repo, self.o.schema,

                                           {'uri': 'cards'}))

        repo.sources_by_uri['cards'] = self.sources[-1]

        self.rql = self.sources[-1]

        do_monkey_patch()

        clear_ms_caches(repo)

    def tearDown(self):

        undo_monkey_patch()

        del self.sources[-1]

        del self.sources[-1]

        del repo.sources_by_uri['ldapuser']

        del repo.sources_by_uri['cards']

        # restore hijacked security

        self.restore_orig_affaire_security()

        self.restore_orig_euser_security()

    def restore_orig_affaire_security(self):

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

        affreadperms[-1] = self.prevrqlexpr_affaire

        self.schema['Affaire']._groups['read'] = tuple(affreadperms)

        clear_cache(self.schema['Affaire'], 'ERSchema_get_rqlexprs')

    def restore_orig_euser_security(self):

        userreadperms = list(self.schema['EUser']._groups['read'])

        userreadperms[-1] = self.prevrqlexpr_user

        self.schema['EUser']._groups['read'] = tuple(userreadperms)

        clear_cache(self.schema['EUser'], 'ERSchema_get_rqlexprs')

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('EGroup X',

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

    def test_simple_system_ldap(self):

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

        """

        self._test('EUser 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.rql: {'X': s[0]}}, False)

    def test_simple_eid_specified(self):

        """retrieve EUser 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 EUser 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 EUser X from system source only (X is invariant and in_group not supported by ldap source)

        """

        self._test('Any X WHERE X is EUser, 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 EUser X from system source only (has_text not supported by ldap source)

        """

        # specify EUser 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('EUser 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 EUser, 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 EUser, 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 EUser 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.is_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.rql : '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.rql: {'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.rql: {'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.rql: {'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.rql: {'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.rql: {'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.rql: {'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.rql: {'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.rql: {'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.rql: {'s': s[0], 'N': s[0]}}, False)

class MSPlannerTC(BaseMSPlannerTC):

    def setUp(self):

        BaseMSPlannerTC.setUp(self)

        self.planner = MSPlanner(self.o.schema, self.o._rqlhelper)

    _test = test_plan

    def test_simple_system_only(self):

        """retrieve entities only supported by the system source

        """

        self._test('EGroup X',

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

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

    def test_simple_system_only_limit(self):

        """retrieve entities only supported by the system source

        """

        self._test('EGroup X LIMIT 10',

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

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

    def test_simple_system_only_limit_offset(self):

        """retrieve entities only supported by the system source

        """

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

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

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

    def test_simple_system_ldap(self):

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

        """

        self._test('EUser X',

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

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

    def test_simple_system_ldap_limit(self):

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

        """

        self._test('EUser X LIMIT 10',

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

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

    def test_simple_system_ldap_limit_offset(self):

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

        """

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

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

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

    def test_simple_system_ldap_ordered_limit_offset(self):

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

        """

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

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

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

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

                       ]),

                   ])

    def test_simple_system_ldap_aggregat(self):

        """retrieve EUser 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 EUser',

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

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

                        [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.rql, self.system], {}, [])])

    def test_simple_eid_specified(self):

        """retrieve EUser 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': 'EUser', 'L': 'String'}])],

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

                   {'x': ueid})

    def test_simple_eid_invariant(self):

        """retrieve EUser 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 EUser X from system source only (X is invariant and in_group not supported by ldap source)

        """

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

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

                                       [{'X': 'EUser', 'G': 'EGroup'}])],

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

    def test_complex_base(self):

        """

        1. retrieve Any X, L WHERE X is EUser, 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 EUser, X in_group G, X login L, G name "users"',

                   [('FetchStep', [('Any X,L WHERE X login L, X is EUser', [{'X': 'EUser', '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 EGroup, X is EUser',

                                       [{'X': 'EUser', 'L': 'String', 'G': 'EGroup'}])],

                     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 EUser, 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 EUser, X in_group G, X login L, G name "users"',

                   [('FetchStep', [('Any X,L WHERE X login L, X is EUser', [{'X': 'EUser', '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 EGroup, X is EUser',

                                       [{'X': 'EUser', 'L': 'String', 'G': 'EGroup'}])],

                     10, 10,

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

                    ])

    def test_complex_ordered(self):

        self._test('Any L ORDERBY L WHERE X login L',

                   [('AggrStep', 'Any L ORDERBY L', None, None, 'table0', None, 

                     [('FetchStep', [('Any L WHERE X login L, X is EUser',

                                      [{'X': 'EUser', 'L': 'String'}])],

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

                      ])

                    ])

    def test_complex_ordered_limit_offset(self):

        self._test('Any L ORDERBY L LIMIT 10 OFFSET 10 WHERE X login L',

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

                     [('FetchStep', [('Any L WHERE X login L, X is EUser',

                                      [{'X': 'EUser', 'L': 'String'}])],

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

                      ])

                    ])

    def test_complex_invariant_ordered(self):

        """

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