# -*- 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

from logilab.common.testlib import TestCase, unittest_main

from rql import BadRQLQuery, RQLSyntaxError

from cubicweb import QueryError, Unauthorized, Binary

from cubicweb.server.sqlutils import SQL_PREFIX

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

from rql.utils import register_function, FunctionDescr

class group_sort_value(FunctionDescr):

    supported_backends = ('sqlite',)

    rtype = 'Int'

try:

    register_function(group_sort_value)

except AssertionError:

    pass

from cubicweb.server.sqlutils import SQL_CONNECT_HOOKS

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 = TestServerConfiguration(apphome=UtilsTC.datadir)

    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

    cnx.close()

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

    def test_preprocess_2(self):

        teid = self.qexecute("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('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)

    def test_preprocess_security(self):

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

                                  'WHERE X is ET, ET name ETN')

        plan.cnx = self.user_groups_session('users')

        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.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.assertListEqual(sorted(solutions),

                              sorted([{'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': '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': '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):

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

        plan.cnx = self.user_groups_session('users')

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

        rqlst = self._prepare('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')

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

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

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

        self.assertRaises(RQLSyntaxError, self.qexecute,

                          'Any X WHERE X is CWRType, X name "�wned_by"')

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

    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("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(), '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):

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

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

        self.assertEqual(rset.description, [('CWGroup',), ('CWGroup',), ('CWGroup',), ('CWGroup',)])

        rset = self.qexecute('Any X ORDERBY N DESC WHERE X is CWGroup, X name N')

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

    def test_select_3(self):

        rset = self.qexecute('Any N GROUPBY N WHERE X is CWGroup, X name N')