# copyright 2003-2016 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.

#

# CubicWeb is distributed in the hope that it will be useful, but WITHOUT

# 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 six import string_types

from logilab.common.testlib import mock_object

from logilab.common.decorators import monkeypatch

from yams import BadSchemaDefinition

from yams.buildobjs import RelationDefinition

from rql import parse, nodes, RQLHelper

from cubicweb import Unauthorized, rqlrewrite, devtools

from cubicweb.rqlrewrite import RQLRewriter

from cubicweb.schema import RRQLExpression, ERQLExpression

from cubicweb.devtools import repotest

from cubicweb.devtools.testlib import CubicWebTC, TestCase

def setUpModule(*args):

    global rqlhelper, schema

    config = devtools.TestServerConfiguration('data-rewrite', __file__)

    config.bootstrap_cubes()

    schema = config.load_schema()

    schema.add_relation_def(RelationDefinition(subject='Card', name='in_state',

                                               object='State', cardinality='1*'))

    rqlhelper = RQLHelper(schema, special_relations={'eid': 'uid',

                                                     'has_text': 'fti'})

    repotest.do_monkey_patch()

def tearDownModule(*args):

    repotest.undo_monkey_patch()

    global rqlhelper, schema

    del rqlhelper, schema

def eid_func_map(eid):

    return {1: 'CWUser',

            2: 'Card',

            3: 'Affaire'}[eid]

def _prepare_rewriter(rewriter_cls, kwargs):

    class FakeVReg:

        schema = schema

        @staticmethod

        def solutions(sqlcursor, rqlst, kwargs):

            rqlhelper.compute_solutions(rqlst, {'eid': eid_func_map}, kwargs=kwargs)

        class rqlhelper:

            @staticmethod

            def annotate(rqlst):

                rqlhelper.annotate(rqlst)

    return rewriter_cls(mock_object(vreg=FakeVReg, user=(mock_object(eid=1))))

def rewrite(rqlst, snippets_map, kwargs, existingvars=None):

    rewriter = _prepare_rewriter(rqlrewrite.RQLRewriter, kwargs)

    # turn {(V1, V2): constraints} into [(varmap, constraints)]

    snippets = []

    snippet_varmap = {}

    for v, exprs in sorted(snippets_map.items()):

        rqlexprs = []

        varmap = dict([v])

        for snippet in exprs:

            # when the same snippet is impacting several variables, group them

            # unless there is some conflicts on the snippet's variable name (we

            # only want that for constraint on relations using both S and O)

            if snippet in snippet_varmap and not (

                    set(varmap.values()) & set(snippet_varmap[snippet].values())):

                snippet_varmap[snippet].update(varmap)

                continue

            snippet_varmap[snippet] = varmap

            if isinstance(snippet, string_types):

                snippet = mock_object(snippet_rqlst=parse(u'Any X WHERE ' + snippet).children[0],

                                      expression=u'Any X WHERE ' + snippet)

            rqlexprs.append(snippet)

        if rqlexprs:

            snippets.append((varmap, rqlexprs))

    rqlhelper.compute_solutions(rqlst.children[0], {'eid': eid_func_map}, kwargs=kwargs)

    rewriter.rewrite(rqlst.children[0], snippets, kwargs, existingvars)

    check_vrefs(rqlst.children[0])

    return rewriter.rewritten

def check_vrefs(node):

    vrefmaps = {}

    selects = []

    for vref in node.iget_nodes(nodes.VariableRef):

        stmt = vref.stmt

        try:

            vrefmaps[stmt].setdefault(vref.name, set()).add(vref)

        except KeyError:

            vrefmaps[stmt] = {vref.name: set((vref,))}

            selects.append(stmt)

    assert node in selects, (node, selects)

    for stmt in selects:

        for var in stmt.defined_vars.values():

            assert var.stinfo['references']

            vrefmap = vrefmaps[stmt]

            assert not (var.stinfo['references'] ^ vrefmap[var.name]), (

                node.as_string(), var, var.stinfo['references'], vrefmap[var.name])

class RQLRewriteTC(TestCase):

    """a faire:

    * optimisation: detecter les relations utilisees dans les rqlexpressions qui

      sont presentes dans la requete de depart pour les reutiliser si possible

    * "has_<ACTION>_permission" ?

    """

    def test_base_var(self):

        constraint = ('X in_state S, U in_group G, P require_state S,'

                      'P name "read", P require_group G')

        rqlst = parse(u'Card C')

        rewrite(rqlst, {('C', 'X'): (constraint,)}, {})

        self.assertEqual(

            rqlst.as_string(),

            u'Any C WHERE C is Card, B eid %(D)s, '

            'EXISTS(C in_state A, B in_group E, F require_state A, '

            'F name "read", F require_group E, A is State, E is CWGroup, F is CWPermission)')

    def test_multiple_var(self):

        card_constraint = ('X in_state S, U in_group G, P require_state S,'

                           'P name "read", P require_group G')

        affaire_constraints = ('X ref LIKE "PUBLIC%"', 'U in_group G, G name "public"')

        kwargs = {'u': 2}

        rqlst = parse(u'Any S WHERE S documented_by C, C eid %(u)s')

        rewrite(rqlst, {('C', 'X'): (card_constraint,), ('S', 'X'): affaire_constraints},

                kwargs)

        self.assertMultiLineEqual(

            rqlst.as_string(),

            u'Any S WHERE S documented_by C, C eid %(u)s, B eid %(D)s, '

            'EXISTS(C in_state A, B in_group E, F require_state A, '

            'F name "read", F require_group E, A is State, E is CWGroup, F is CWPermission), '

            '(EXISTS(S ref LIKE "PUBLIC%")) '

            'OR (EXISTS(B in_group G, G name "public", G is CWGroup)), '

            'S is Affaire')

        self.assertIn('D', kwargs)

    def test_or(self):

        constraint = (

            '(X identity U) OR '

            '(X in_state ST, CL identity U, CL in_state ST, ST name "subscribed")'

        )

        rqlst = parse(u'Any S WHERE S owned_by C, C eid %(u)s, S is in (CWUser, CWGroup)')

        rewrite(rqlst, {('C', 'X'): (constraint,)}, {'u': 1})

        self.assertEqual(

            rqlst.as_string(),

            'Any S WHERE S owned_by C, C eid %(u)s, S is IN(CWUser, CWGroup), A eid %(B)s, '

            'EXISTS((C identity A) OR (C in_state D, E identity A, '

            'E in_state D, D name "subscribed"), D is State, E is CWUser)')

    def test_simplified_rqlst(self):

        constraint = ('X in_state S, U in_group G, P require_state S,'

                      'P name "read", P require_group G')

        rqlst = parse(u'Any 2')  # this is the simplified rql st for Any X WHERE X eid 12

        rewrite(rqlst, {('2', 'X'): (constraint,)}, {})

        self.assertEqual(

            rqlst.as_string(),

            u'Any 2 WHERE B eid %(C)s, '

            'EXISTS(2 in_state A, B in_group D, E require_state A, '

            'E name "read", E require_group D, A is State, D is CWGroup, E is CWPermission)')

    def test_optional_var_1(self):

        constraint = ('X in_state S, U in_group G, P require_state S,'

                      'P name "read", P require_group G')

        rqlst = parse(u'Any A,C WHERE A documented_by C?')

        rewrite(rqlst, {('C', 'X'): (constraint,)}, {})

        self.assertEqual(

            rqlst.as_string(),

            u'Any A,C WHERE A documented_by C?, A is Affaire '

            'WITH C BEING '

            '(Any C WHERE EXISTS(C in_state B, D in_group F, G require_state B, G name "read", '

            'G require_group F), D eid %(A)s, C is Card)')

    def test_optional_var_2(self):

        constraint = ('X in_state S, U in_group G, P require_state S,'

                      'P name "read", P require_group G')

        rqlst = parse(u'Any A,C,T WHERE A documented_by C?, C title T')

        rewrite(rqlst, {('C', 'X'): (constraint,)}, {})

        self.assertEqual(

            rqlst.as_string(),

            u'Any A,C,T WHERE A documented_by C?, A is Affaire '

            'WITH C,T BEING '

            '(Any C,T WHERE C title T, EXISTS(C in_state B, D in_group F, '

            'G require_state B, G name "read", G require_group F), '

            'D eid %(A)s, C is Card)')

    def test_optional_var_3(self):

        constraint1 = ('X in_state S, U in_group G, P require_state S,'

                       'P name "read", P require_group G')

        constraint2 = 'X in_state S, S name "public"'

        rqlst = parse(u'Any A,C,T WHERE A documented_by C?, C title T')

        rewrite(rqlst, {('C', 'X'): (constraint1, constraint2)}, {})

        self.assertEqual(

            rqlst.as_string(),

            u'Any A,C,T WHERE A documented_by C?, A is Affaire '

            'WITH C,T BEING (Any C,T WHERE C title T, '

            '(EXISTS(C in_state B, D in_group F, G require_state B, '

            'G name "read", G require_group F)) '

            'OR (EXISTS(C in_state E, E name "public")), '

            'D eid %(A)s, C is Card)')

    def test_optional_var_4(self):

        constraint1 = 'A created_by U, X documented_by A'

        constraint2 = 'A created_by U, X concerne A'

        constraint3 = 'X created_by U'

        rqlst = parse(u'Any X,LA,Y WHERE LA? documented_by X, LA concerne Y')

        rewrite(rqlst, {('LA', 'X'): (constraint1, constraint2),

                        ('X', 'X'): (constraint3,),

                        ('Y', 'X'): (constraint3,)}, {})

        self.assertEqual(

            rqlst.as_string(),

            u'Any X,LA,Y WHERE LA? documented_by X, LA concerne Y, B eid %(C)s, '

            'EXISTS(X created_by B), EXISTS(Y created_by B), '

            'X is Card, Y is IN(Division, Note, Societe) '

            'WITH LA BEING (Any LA WHERE (EXISTS(A created_by B, LA documented_by A)) '

            'OR (EXISTS(E created_by B, LA concerne E)), '

            'B eid %(D)s, LA is Affaire)')

    def test_ambiguous_optional_same_exprs(self):

        """See #3013535"""

        # see test of the same name in RewriteFullTC: original problem is

        # unreproducible here because it actually lies in

        # RQLRewriter.insert_local_checks

        rqlst = parse(u'Any A,AR,X,CD WHERE A concerne X?, A ref AR, '

                      'A eid %(a)s, X creation_date CD')

        rewrite(rqlst, {('X', 'X'): ('X created_by U',)}, {'a': 3})

        self.assertEqual(

            rqlst.as_string(),

            u'Any A,AR,X,CD WHERE A concerne X?, A ref AR, A eid %(a)s '

            'WITH X,CD BEING (Any X,CD WHERE X creation_date CD, '

            'EXISTS(X created_by B), B eid %(A)s, X is IN(Division, Note, Societe))')

    def test_ambiguous_optional_same_exprs_constant(self):

        rqlst = parse(u'Any A,AR,X WHERE A concerne X?, A ref AR, '

                      'A eid %(a)s, X creation_date TODAY')

        rewrite(rqlst, {('X', 'X'): ('X created_by U',)}, {'a': 3})

        self.assertEqual(

            rqlst.as_string(),

            u'Any A,AR,X WHERE A concerne X?, A ref AR, A eid %(a)s '

            'WITH X BEING (Any X WHERE X creation_date TODAY, '

            'EXISTS(X created_by B), B eid %(A)s, X is IN(Division, Note, Societe))')

    def test_optional_var_inlined(self):

        c1 = ('X require_permission P')

        c2 = ('X inlined_card O, O require_permission P')

        rqlst = parse(u'Any C,A,R WHERE A? inlined_card C, A ref R')

        rewrite(rqlst, {('C', 'X'): (c1,),

                        ('A', 'X'): (c2,),

                        }, {})

        # XXX suboptimal

        self.assertEqual(

            rqlst.as_string(),

            "Any C,A,R WITH A,C,R BEING "

            "(Any A,C,R WHERE A? inlined_card C, A ref R, "

            "(A is NULL) OR (EXISTS(A inlined_card B, B require_permission D, "

            "B is Card, D is CWPermission)), "

            "A is Affaire, C is Card, EXISTS(C require_permission E, E is CWPermission))")

    # def test_optional_var_inlined_has_perm(self):

    #     c1 = ('X require_permission P')

    #     c2 = ('X inlined_card O, U has_read_permission O')

    #     rqlst = parse(u'Any C,A,R WHERE A? inlined_card C, A ref R')

    #     rewrite(rqlst, {('C', 'X'): (c1,),

    #                     ('A', 'X'): (c2,),

    #                     }, {})

    #     self.assertEqual(rqlst.as_string(),

    #                          "")

    def test_optional_var_inlined_imbricated_error(self):

        c1 = ('X require_permission P')

        c2 = ('X inlined_card O, O require_permission P')

        rqlst = parse(u'Any C,A,R,A2,R2 WHERE A? inlined_card C, A ref R,'

                      'A2? inlined_card C, A2 ref R2')

        self.assertRaises(BadSchemaDefinition,

                          rewrite, rqlst, {('C', 'X'): (c1,),

                                           ('A', 'X'): (c2,),

                                           ('A2', 'X'): (c2,),

                                           }, {})

    def test_optional_var_inlined_linked(self):

        c1 = ('X require_permission P')

        rqlst = parse(u'Any A,W WHERE A inlined_card C?, C inlined_note N, '

                      'N inlined_affaire W')

        rewrite(rqlst, {('C', 'X'): (c1,)}, {})

        self.assertEqual(rqlst.as_string(),

                         'Any A,W WHERE A inlined_card C?, A is Affaire '

                         'WITH C,N,W BEING (Any C,N,W WHERE C inlined_note N, '

                         'N inlined_affaire W, EXISTS(C require_permission B), '

                         'C is Card, N is Note, W is Affaire)')

    def test_relation_optimization_1_lhs(self):

        # since Card in_state State as monovalued cardinality, the in_state

        # relation used in the rql expression can be ignored and S replaced by

        # the variable from the incoming query

        snippet = ('X in_state S, S name "hop"')

        rqlst = parse(u'Card C WHERE C in_state STATE')

        rewrite(rqlst, {('C', 'X'): (snippet,)}, {})

        self.assertEqual(rqlst.as_string(),

                         'Any C WHERE C in_state STATE, C is Card, '

                         'EXISTS(STATE name "hop"), STATE is State')

    def test_relation_optimization_1_rhs(self):

        snippet = ('TW subworkflow_exit X, TW name "hop"')

        rqlst = parse(u'WorkflowTransition C WHERE C subworkflow_exit EXIT')

        rewrite(rqlst, {('EXIT', 'X'): (snippet,)}, {})

        self.assertEqual(rqlst.as_string(),

                         'Any C WHERE C subworkflow_exit EXIT, C is WorkflowTransition, '

                         'EXISTS(C name "hop"), EXIT is SubWorkflowExitPoint')

    def test_relation_optimization_2_lhs(self):

        # optional relation can be shared if also optional in the snippet

        snippet = ('X in_state S?, S name "hop"')

        rqlst = parse(u'Card C WHERE C in_state STATE?')

        rewrite(rqlst, {('C', 'X'): (snippet,)}, {})

        self.assertEqual(rqlst.as_string(),

                         'Any C WHERE C in_state STATE?, C is Card, '

                         'EXISTS(STATE name "hop"), STATE is State')

    def test_relation_optimization_2_rhs(self):

        snippet = ('TW? subworkflow_exit X, TW name "hop"')

        rqlst = parse(u'SubWorkflowExitPoint EXIT WHERE C? subworkflow_exit EXIT')

        rewrite(rqlst, {('EXIT', 'X'): (snippet,)}, {})

        self.assertEqual(rqlst.as_string(),

                         'Any EXIT WHERE C? subworkflow_exit EXIT, EXIT is SubWorkflowExitPoint, '

                         'EXISTS(C name "hop"), C is WorkflowTransition')

    def test_relation_optimization_3_lhs(self):

        # optional relation in the snippet but not in the orig tree can be shared

        snippet = ('X in_state S?, S name "hop"')

        rqlst = parse(u'Card C WHERE C in_state STATE')

        rewrite(rqlst, {('C', 'X'): (snippet,)}, {})

        self.assertEqual(rqlst.as_string(),

                         'Any C WHERE C in_state STATE, C is Card, '

                         'EXISTS(STATE name "hop"), STATE is State')

    def test_relation_optimization_3_rhs(self):

        snippet = ('TW? subworkflow_exit X, TW name "hop"')

        rqlst = parse(u'WorkflowTransition C WHERE C subworkflow_exit EXIT')

        rewrite(rqlst, {('EXIT', 'X'): (snippet,)}, {})

        self.assertEqual(rqlst.as_string(),

                         'Any C WHERE C subworkflow_exit EXIT, C is WorkflowTransition, '

                         'EXISTS(C name "hop"), EXIT is SubWorkflowExitPoint')

    def test_relation_non_optimization_1_lhs(self):

        # but optional relation in the orig tree but not in the snippet can't be shared

        snippet = ('X in_state S, S name "hop"')

        rqlst = parse(u'Card C WHERE C in_state STATE?')

        rewrite(rqlst, {('C', 'X'): (snippet,)}, {})

        self.assertEqual(rqlst.as_string(),

                         'Any C WHERE C in_state STATE?, C is Card, '

                         'EXISTS(C in_state A, A name "hop", A is State), STATE is State')

    def test_relation_non_optimization_1_rhs(self):

        snippet = ('TW subworkflow_exit X, TW name "hop"')

        rqlst = parse(u'SubWorkflowExitPoint EXIT WHERE C? subworkflow_exit EXIT')

        rewrite(rqlst, {('EXIT', 'X'): (snippet,)}, {})

        self.assertEqual(rqlst.as_string(),

                         'Any EXIT WHERE C? subworkflow_exit EXIT, EXIT is SubWorkflowExitPoint, '

                         'EXISTS(A subworkflow_exit EXIT, A name "hop", A is WorkflowTransition), '