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

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

"""

from datetime import date, datetime

from logilab.common.testlib import TestCase, unittest_main

from rql import BadRQLQuery, RQLSyntaxError

from cubicweb import QueryError, Unauthorized

from cubicweb.server.sqlutils import SQL_PREFIX

from cubicweb.server.utils import crypt_password

from cubicweb.server.sources.native import make_schema

from cubicweb.devtools import init_test_database

from cubicweb.devtools.repotest import tuplify, BaseQuerierTC

from unittest_session import Variable

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

from logilab.common.adbh import _GenericAdvFuncHelper

TYPEMAP = _GenericAdvFuncHelper.TYPE_MAPPING

class MakeSchemaTC(TestCase):

    def test_known_values(self):

        self.assertEquals(make_schema((Variable('A'), Variable('B')), solution, 

                                      'table0', TYPEMAP),

                          ('C0 text,C1 integer', {'A': 'table0.C0', 'B': 'table0.C1'}))

repo, cnx = init_test_database('sqlite')

class UtilsTC(BaseQuerierTC):

    repo = repo

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

        rqlst = self._prepare('Any COUNT(RDEF) WHERE RDEF relation_type X, X eid %(x)s', {'x': reid})

    def test_preprocess_2(self):

        teid = self.execute("INSERT Tag X: X name 'tag'")[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)

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

    def test_preprocess_security(self):

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

                                  'WHERE X is ET, ET name ETN')

        plan.session = self._user_session(('users',))[1]

        union = plan.rqlst

        plan.preprocess(union)

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

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

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

        self.assertEquals(len(subq.children), 3)

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

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

        self.assertEquals([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.assertEquals(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)))'

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

                                       'D': 'Affaire',

                                       'E': 'Note',

                                       'F': 'Societe',

                                       'G': 'SubDivision',

                                       'H': 'Affaire',

                                       'I': 'Affaire',

                                       'J': 'Affaire',

                                       'X': 'Affaire',

        rql, solutions = partrqls[1]

        self.assertListEquals(sorted(solutions),

        rql, solutions = partrqls[2]

        self.assertEquals(rql,

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

                                       'X': 'Basket',

                                       'ETN': 'String',

                                       }])

    def test_preprocess_security_aggregat(self):

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

        plan.session = self._user_session(('users',))[1]

        union = plan.rqlst

        plan.preprocess(union)

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

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

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

        self.assertEquals(len(subq.children), 3)

        self.assertEquals([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.assertEquals(len(rqlst.solutions), 1)

    def test_build_description(self):

        # should return an empty result set

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

        rset = self.execute('Any 1')

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

        rset = self.execute('Any TRUE')

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

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

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

        rset = self.execute('Any TODAY')

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

        rset = self.execute('Any NOW')

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

class QuerierTC(BaseQuerierTC):

    repo = repo

    def test_encoding_pb(self):

        self.assertRaises(RQLSyntaxError, self.execute,

    def test_unknown_eid(self):

        # should return an empty result set

        self.failIf(self.execute('Any X WHERE X eid 99999999'))

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

    def test_select_1(self):

        result, descr = rset.rows, rset.description

        self.assertEquals(tuplify(result), [(1,), (2,), (3,), (4,)])

    def test_select_2(self):

        self.assertEquals(tuplify(rset.rows), [(3,), (1,), (4,), (2,)])

        self.assertEquals(tuplify(rset.rows), [(2,), (4,), (1,), (3,)])

    def test_select_3(self):

        result, descr = rset.rows, rset.description

        result.sort()

        self.assertEquals(tuplify(result), [('guests',), ('managers',), ('owners',), ('users',)])

        self.assertEquals(descr, [('String',), ('String',), ('String',), ('String',)])

    def test_select_is(self):

        rset = self.execute('Any X, TN ORDERBY TN LIMIT 10 WHERE X is T, T name TN')

        result, descr = rset.rows, rset.description

        self.assertEquals(result[0][1], descr[0][0])

    def test_select_is_aggr(self):

        rset = self.execute('Any TN, COUNT(X) GROUPBY TN ORDERBY 2 DESC WHERE X is T, T name TN')

        result, descr = rset.rows, rset.description

        self.assertEquals(descr[0][0], 'String')

        self.assertEquals(descr[0][1], 'Int')

    def test_select_groupby_orderby(self):

        self.assertEquals(tuplify(rset.rows), [('guests',), ('managers',), ('owners',), ('users',)])

        self.assertEquals(rset.description, [('String',), ('String',), ('String',), ('String',)])

    def test_select_complex_groupby(self):

        rset = self.execute('Any N GROUPBY N WHERE X name N')

        rset = self.execute('Any N,MAX(D) GROUPBY N LIMIT 5 WHERE X name N, X creation_date D')

    def test_select_inlined_groupby(self):

        seid = self.execute('State X WHERE X name "deactivated"')[0][0]

        rset = self.execute('Any U,L,S GROUPBY U,L,S WHERE X in_state S, U login L, S eid %s' % seid)

    def test_select_complex_orderby(self):

        rset1 = self.execute('Any N ORDERBY N WHERE X name N')

        self.assertEquals(sorted(rset1.rows), rset1.rows)

        rset = self.execute('Any N ORDERBY N LIMIT 5 OFFSET 1 WHERE X name N')

        self.assertEquals(rset.rows[0][0], rset1.rows[1][0]) 

        self.assertEquals(len(rset), 5)

    def test_select_5(self):

        self.assertEquals(tuplify(rset.rows), [(3, 'guests',), (1, 'managers',), (4, 'owners',), (2, 'users',)])

    def test_select_6(self):

        self.execute("INSERT Personne X: X nom 'bidule'")[0]

        rset = self.execute('Any Y where X name TMP, Y nom in (TMP, "bidule")')

        #self.assertEquals(rset.description, [('Personne',), ('Personne',)])

        self.assert_(('Personne',) in rset.description)

        rset = self.execute('DISTINCT Any Y where X name TMP, Y nom in (TMP, "bidule")')

        self.assert_(('Personne',) in rset.description)

    def test_select_not_attr(self):

        self.execute("INSERT Personne X: X nom 'bidule'")

        self.execute("INSERT Societe X: X nom 'chouette'")

        rset = self.execute('Personne X WHERE NOT X nom "bidule"')

        self.assertEquals(len(rset.rows), 0, rset.rows)

        rset = self.execute('Personne X WHERE NOT X nom "bid"')

        self.assertEquals(len(rset.rows), 1, rset.rows)

        self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")

        rset = self.execute('Personne X WHERE NOT X travaille S')

        self.assertEquals(len(rset.rows), 0, rset.rows)

    def test_select_is_in(self):

        self.execute("INSERT Personne X: X nom 'bidule'")

        self.execute("INSERT Societe X: X nom 'chouette'")

        self.assertEquals(len(self.execute("Any X WHERE X is IN (Personne, Societe)")),

                          2)

    def test_select_not_rel(self):

        self.execute("INSERT Personne X: X nom 'bidule'")

        self.execute("INSERT Societe X: X nom 'chouette'")

        self.execute("INSERT Personne X: X nom 'autre'")

        self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")

        rset = self.execute('Personne X WHERE NOT X travaille S')

        self.assertEquals(len(rset.rows), 1, rset.rows)

        rset = self.execute('Personne X WHERE NOT X travaille S, S nom "chouette"')

        self.assertEquals(len(rset.rows), 1, rset.rows)

    def test_select_nonregr_inlined(self):

        self.execute("INSERT Note X: X para 'bidule'")

        self.execute("INSERT Personne X: X nom 'chouette'")

        self.execute("INSERT Personne X: X nom 'autre'")

        self.execute("SET X ecrit_par P WHERE X para 'bidule', P nom 'chouette'")

                            'N ecrit_par U, N type T')#, {'x': self.ueid})

        self.assertEquals(len(rset.rows), 0)

    def test_select_nonregr_edition_not(self):

        groupeids = set((1, 2, 3))

        self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms))

        self.assertEquals(sorted(r[0] for r in rset.rows), sorted(groupeids - groupreadperms))

    def test_select_outer_join(self):

        peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]

        peid2 = self.execute("INSERT Personne X: X nom 'autre'")[0][0]

        seid1 = self.execute("INSERT Societe X: X nom 'chouette'")[0][0]

        seid2 = self.execute("INSERT Societe X: X nom 'chouetos'")[0][0]

        rset = self.execute('Any X,S ORDERBY X WHERE X travaille S?')

        self.assertEquals(rset.rows, [[peid1, None], [peid2, None]])

        self.execute("SET P travaille S WHERE P nom 'bidule', S nom 'chouette'")

        rset = self.execute('Any X,S ORDERBY X WHERE X travaille S?')

        self.assertEquals(rset.rows, [[peid1, seid1], [peid2, None]])

        rset = self.execute('Any S,X ORDERBY S WHERE X? travaille S')

        self.assertEquals(rset.rows, [[seid1, peid1], [seid2, None]])

    def test_select_outer_join_optimized(self):

        peid1 = self.execute("INSERT Personne X: X nom 'bidule'")[0][0]

        rset = self.execute('Any X WHERE X eid %(x)s, P? connait X', {'x':peid1}, 'x')

        self.assertEquals(rset.rows, [[peid1]])

        rset = self.execute('Any X WHERE X eid %(x)s, X require_permission P?', {'x':peid1}, 'x')