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

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

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

"""

from contextlib import contextmanager

from datetime import date, datetime, timedelta, tzinfo

import unittest

import pytz

from rql import BadRQLQuery

from rql.utils import register_function, FunctionDescr

from cubicweb import QueryError, Unauthorized, Binary, devtools

from cubicweb.server.sqlutils import SQL_CONNECT_HOOKS, SQL_PREFIX

from cubicweb.server.utils import crypt_password

from cubicweb.server.querier import manual_build_descr, _make_description

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):

        return {"managers": "3", "users": "2", "guests": "1", "owners": "0"}[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 = devtools.TestServerConfiguration('data', __file__)

    handler = devtools.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(unittest.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 test_preprocess_1(self):

        with self.admin_access.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.admin_access.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(str(parse(expected)),

                                  str(parse(got)))

    def test_preprocess_security(self):

        with self.user_groups_session('users') 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, 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': '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):

        with self.user_groups_session('users') 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.admin_access.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.admin_access._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')

        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.admin_access.cnx() as cnx:

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

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

            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 setUp(self):

        super(QuerierTC, self).setUp()

        with self.admin_access.cnx() as cnx:

            self.maxeid = cnx.execute('Any MAX(X)')[0][0]

    def tearDown(self):

        super(QuerierTC, self).tearDown()

        with self.admin_access.cnx() as cnx:

            cnx.execute('DELETE Any X WHERE X eid > %s' % self.maxeid)

            cnx.commit()

    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], int)

    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,)])

        self.assertEqual(descr, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)])

    def test_select_2(self):