# -*- coding: iso-8859-1 -*-

# 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/>.

"""unit tests for modules cubicweb.server.querier and cubicweb.server.ssplanner

"""

from datetime import date, datetime, timedelta, tzinfo

import pytz

from six import PY2, integer_types, binary_type, text_type

from logilab.common.testlib import TestCase, unittest_main

from rql import BadRQLQuery

from cubicweb import QueryError, Unauthorized, Binary

from cubicweb.server.utils import crypt_password

from cubicweb.server.querier import manual_build_descr, _make_description

from cubicweb.devtools import get_test_db_handler, TestServerConfiguration

from cubicweb.devtools.testlib import CubicWebTC

from cubicweb.devtools.repotest import tuplify, BaseQuerierTC

class FixedOffset(tzinfo):

    def __init__(self, hours=0):

        self.hours = hours

    def utcoffset(self, dt):

        return timedelta(hours=self.hours)

    def dst(self, dt):

        return timedelta(0)

# register priority/severity sorting registered procedure

class group_sort_value(FunctionDescr):

    supported_backends = ('sqlite',)

    rtype = 'Int'

try:

    register_function(group_sort_value)

except AssertionError:

    pass

def init_sqlite_connexion(cnx):

    def group_sort_value(text):

    cnx.create_function("GROUP_SORT_VALUE", 1, group_sort_value)

SQL_CONNECT_HOOKS['sqlite'].append(init_sqlite_connexion)

def setUpClass(cls, *args):

    global repo, cnx

    config = TestServerConfiguration('data', __file__)

    handler = get_test_db_handler(config)

    handler.build_db_cache()

    repo, cnx = handler.get_repo_and_cnx()

    cls.repo = repo

def tearDownClass(cls, *args):

    global repo, cnx

    repo.shutdown()

    del repo, cnx

class Variable:

    def __init__(self, name):

        self.name = name

        self.children = []

    def get_type(self, solution, args=None):

        return solution[self.name]

    def as_string(self):

        return self.name

class Function:

    def __init__(self, name, varname):

        self.name = name

        self.children = [Variable(varname)]

    def get_type(self, solution, args=None):

        return 'Int'

class MakeDescriptionTC(TestCase):

    def test_known_values(self):

        solution = {'A': 'Int', 'B': 'CWUser'}

        self.assertEqual(_make_description((Function('max', 'A'), Variable('B')), {}, solution),

                          ['Int','CWUser'])

class UtilsTC(BaseQuerierTC):

    setUpClass = classmethod(setUpClass)

    tearDownClass = classmethod(tearDownClass)

    def get_max_eid(self):

        # no need for cleanup here

        return None

    def cleanup(self):

        # no need for cleanup here

        pass

    def test_preprocess_1(self):

        with self.session.new_cnx() as cnx:

            reid = cnx.execute('Any X WHERE X is CWRType, X name "owned_by"')[0][0]

            rqlst = self._prepare(cnx, 'Any COUNT(RDEF) WHERE RDEF relation_type X, X eid %(x)s',

                                  {'x': reid})

            self.assertEqual([{'RDEF': 'CWAttribute'}, {'RDEF': 'CWRelation'}],

                             rqlst.solutions)

    def test_preprocess_2(self):

        with self.session.new_cnx() as cnx:

            teid = cnx.execute("INSERT Tag X: X name 'tag'")[0][0]

            #geid = self.execute("CWGroup G WHERE G name 'users'")[0][0]

            #self.execute("SET X tags Y WHERE X eid %(t)s, Y eid %(g)s",

            #             {'g': geid, 't': teid}, 'g')

            rqlst = self._prepare(cnx, 'Any X WHERE E eid %(x)s, E tags X', {'x': teid})

            # the query may be optimized, should keep only one solution

            # (any one, etype will be discarded)

            self.assertEqual(1, len(rqlst.solutions))

    def assertRQLEqual(self, expected, got):

        from rql import parse

        self.assertMultiLineEqual(text_type(parse(expected)),

                                  text_type(parse(got)))

    def test_preprocess_security(self):

        s = self.user_groups_session('users')

        with s.new_cnx() as cnx:

            plan = self._prepare_plan(cnx, 'Any ETN,COUNT(X) GROUPBY ETN '

                                      'WHERE X is ET, ET name ETN')

            union = plan.rqlst

            plan.preprocess(union)

            self.assertEqual(len(union.children), 1)

            self.assertEqual(len(union.children[0].with_), 1)

            subq = union.children[0].with_[0].query

            self.assertEqual(len(subq.children), 4)

            self.assertEqual([t.as_string() for t in union.children[0].selection],

                              ['ETN','COUNT(X)'])

            self.assertEqual([t.as_string() for t in union.children[0].groupby],

                              ['ETN'])

            partrqls = sorted(((rqlst.as_string(), rqlst.solutions) for rqlst in subq.children))

            rql, solutions = partrqls[0]

            self.assertRQLEqual(rql,

                                'Any ETN,X WHERE X is ET, ET name ETN, (EXISTS(X owned_by %(B)s))'

                                ' OR ((((EXISTS(D concerne C?, C owned_by %(B)s, '

                                '               X identity D, C is Division, D is Affaire))'

                                ' OR (EXISTS(H concerne G?, G owned_by %(B)s, G is SubDivision, '

                                '            X identity H, H is Affaire)))'

                                ' OR (EXISTS(I concerne F?, F owned_by %(B)s, F is Societe, '

                                '            X identity I, I is Affaire)))'

                                ' OR (EXISTS(J concerne E?, E owned_by %(B)s, E is Note, '

                                '            X identity J, J is Affaire)))'

                                ', ET is CWEType, X is Affaire')

            self.assertEqual(solutions, [{'C': 'Division',

                                           'D': 'Affaire',

                                           'E': 'Note',

                                           'F': 'Societe',

                                           'G': 'SubDivision',

                                           'H': 'Affaire',

                                           'I': 'Affaire',

                                           'J': 'Affaire',

                                           'X': 'Affaire',

                                           'ET': 'CWEType', 'ETN': 'String'}])

            rql, solutions = partrqls[1]

            self.assertRQLEqual(rql,  'Any ETN,X WHERE X is ET, ET name ETN, ET is CWEType, '

                                'X is IN(BaseTransition, Bookmark, CWAttribute, CWCache, CWComputedRType, '

                                '        CWConstraint, CWConstraintType, CWEType, CWGroup, CWPermission, CWProperty, '

                                '        CWRType, CWRelation, CWSource, CWUniqueTogetherConstraint, CWUser, Card, '

                                '        Comment, Division, Email, EmailPart, EmailThread, ExternalUri, File, Folder, '

                                '        Frozable, Note, Old, Personne, RQLExpression, Societe, State, SubDivision, '

                                '        SubWorkflowExitPoint, Tag, TrInfo, Transition, Workflow, WorkflowTransition)')

            self.assertCountEqual(solutions,

                                  [{'X': 'BaseTransition', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Bookmark', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Card', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Comment', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Division', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWCache', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWComputedRType', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWConstraint', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWConstraintType', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWEType', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWAttribute', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWGroup', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWRelation', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWPermission', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWProperty', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWRType', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWSource', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWUniqueTogetherConstraint', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'CWUser', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Email', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'EmailPart', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'EmailThread', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'ExternalUri', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'File', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Folder', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Frozable', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Note', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Old', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Personne', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'RQLExpression', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Societe', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'State', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'SubDivision', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'SubWorkflowExitPoint', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Tag', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Transition', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'TrInfo', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'Workflow', 'ETN': 'String', 'ET': 'CWEType'},

                                   {'X': 'WorkflowTransition', 'ETN': 'String', 'ET': 'CWEType'}])

            rql, solutions = partrqls[2]

            self.assertEqual(rql,

                             'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(%(D)s use_email X), '

                             'ET is CWEType, X is EmailAddress')

            self.assertEqual(solutions, [{'X': 'EmailAddress', 'ET': 'CWEType', 'ETN': 'String'}])

            rql, solutions = partrqls[3]

            self.assertEqual(rql,

                              'Any ETN,X WHERE X is ET, ET name ETN, EXISTS(X owned_by %(C)s), '

                              'ET is CWEType, X is Basket')

            self.assertEqual(solutions, [{'X': 'Basket', 'ET': 'CWEType', 'ETN': 'String'}])

    def test_preprocess_security_aggregat(self):

        s = self.user_groups_session('users')

        with s.new_cnx() as cnx:

            plan = self._prepare_plan(cnx, 'Any MAX(X)')

            union = plan.rqlst

            plan.preprocess(union)

            self.assertEqual(len(union.children), 1)

            self.assertEqual(len(union.children[0].with_), 1)

            subq = union.children[0].with_[0].query

            self.assertEqual(len(subq.children), 4)

            self.assertEqual([t.as_string() for t in union.children[0].selection],

                              ['MAX(X)'])

    def test_preprocess_nonregr(self):

        with self.session.new_cnx() as cnx:

            rqlst = self._prepare(cnx, 'Any S ORDERBY SI WHERE NOT S ecrit_par O, S para SI')

            self.assertEqual(len(rqlst.solutions), 1)

    def test_build_description(self):

        # should return an empty result set

        rset = self.qexecute('Any X WHERE X eid %(x)s', {'x': self.session.user.eid})

        self.assertEqual(rset.description[0][0], 'CWUser')

        rset = self.qexecute('Any 1')

        self.assertEqual(rset.description[0][0], 'Int')

        rset = self.qexecute('Any TRUE')

        self.assertEqual(rset.description[0][0], 'Boolean')

        rset = self.qexecute('Any "hop"')

        self.assertEqual(rset.description[0][0], 'String')

        rset = self.qexecute('Any TODAY')

        self.assertEqual(rset.description[0][0], 'Date')

        rset = self.qexecute('Any NOW')

        self.assertEqual(rset.description[0][0], 'Datetime')

        rset = self.qexecute('Any %(x)s', {'x': 1})

        self.assertEqual(rset.description[0][0], 'Int')

        if PY2:

            rset = self.qexecute('Any %(x)s', {'x': long(1)})

            self.assertEqual(rset.description[0][0], 'Int')

        rset = self.qexecute('Any %(x)s', {'x': True})

        self.assertEqual(rset.description[0][0], 'Boolean')

        rset = self.qexecute('Any %(x)s', {'x': 1.0})

        self.assertEqual(rset.description[0][0], 'Float')

        rset = self.qexecute('Any %(x)s', {'x': datetime.now()})

        self.assertEqual(rset.description[0][0], 'Datetime')

        rset = self.qexecute('Any %(x)s', {'x': 'str'})

        self.assertEqual(rset.description[0][0], 'String')

        rset = self.qexecute('Any %(x)s', {'x': u'str'})

        self.assertEqual(rset.description[0][0], 'String')

    def test_build_descr1(self):

        with self.session.new_cnx() as cnx:

            rset = cnx.execute('(Any U,L WHERE U login L) UNION '

                               '(Any G,N WHERE G name N, G is CWGroup)')

            # rset.req = self.session

            orig_length = len(rset)

            rset.rows[0][0] = 9999999

            description = manual_build_descr(cnx, rset.syntax_tree(), None, rset.rows)

            self.assertEqual(len(description), orig_length - 1)

            self.assertEqual(len(rset.rows), orig_length - 1)

            self.assertNotEqual(rset.rows[0][0], 9999999)

    def test_build_descr2(self):

        rset = self.qexecute('Any X,Y WITH X,Y BEING ((Any G,NULL WHERE G is CWGroup) UNION '

                             '(Any U,G WHERE U in_group G))')

        for x, y in rset.description:

            if y is not None:

                self.assertEqual(y, 'CWGroup')

    def test_build_descr3(self):

        rset = self.qexecute('(Any G,NULL WHERE G is CWGroup) UNION '

                             '(Any U,G WHERE U in_group G)')

        for x, y in rset.description:

            if y is not None:

                self.assertEqual(y, 'CWGroup')

class QuerierTC(BaseQuerierTC):

    setUpClass = classmethod(setUpClass)

    tearDownClass = classmethod(tearDownClass)

    def test_unknown_eid(self):

        # should return an empty result set

        self.assertFalse(self.qexecute('Any X WHERE X eid 99999999'))

    def test_typed_eid(self):

        # should return an empty result set

        rset = self.qexecute('Any X WHERE X eid %(x)s', {'x': '1'})

        self.assertIsInstance(rset[0][0], integer_types)

    def test_bytes_storage(self):

        feid = self.qexecute('INSERT File X: X data_name "foo.pdf", '

                             'X data_format "text/plain", X data %(data)s',

                            {'data': Binary(b"xxx")})[0][0]

        fdata = self.qexecute('Any D WHERE X data D, X eid %(x)s', {'x': feid})[0][0]

        self.assertIsInstance(fdata, Binary)

        self.assertEqual(fdata.getvalue(), b'xxx')

    # selection queries tests #################################################

    def test_select_1(self):

        rset = self.qexecute('Any X ORDERBY X WHERE X is CWGroup')

        result, descr = rset.rows, rset.description

        self.assertEqual(tuplify(result), [(2,), (3,), (4,), (5,)])