# coding: utf-8

# copyright 2003-2010 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 urlparse import urlsplit

import pickle

from rql import parse

from logilab.common.testlib import TestCase, unittest_main, mock_object

from cubicweb.devtools.testlib import CubicWebTC

from cubicweb.rset import NotAnEntity, ResultSet, attr_desc_iterator

def pprelcachedict(d):

    res = {}

    for k, (rset, related) in d.items():

        res[k] = sorted(v.eid for v in related)

    return sorted(res.items())

class AttrDescIteratorTC(TestCase):

    """TestCase for cubicweb.rset.attr_desc_iterator"""

    def test_relations_description(self):

        """tests relations_description() function"""

        queries = {

            'Any U,L,M where U is CWUser, U login L, U mail M' : [(1, 'login', 'subject'), (2, 'mail', 'subject')],

            'Any U,L,M where U is CWUser, L is Foo, U mail M' : [(2, 'mail', 'subject')],

            'Any C,P where C is Company, C employs P' : [(1, 'employs', 'subject')],

            'Any C,P where C is Company, P employed_by P' : [],

            'Any C where C is Company, C employs P' : [],

            }

        for rql, relations in queries.items():

            result = list(attr_desc_iterator(parse(rql).children[0]))

            self.assertEqual((rql, result), (rql, relations))

    def test_relations_description_indexed(self):

        """tests relations_description() function"""

        queries = {

            'Any C,U,P,L,M where C is Company, C employs P, U is CWUser, U login L, U mail M' :

            {0: [(2,'employs', 'subject')], 1: [(3,'login', 'subject'), (4,'mail', 'subject')]},

            }

        for rql, results in queries.items():

            for var_index, relations in results.items():

                result = list(attr_desc_iterator(parse(rql).children[0], var_index))

                self.assertEqual(result, relations)

class ResultSetTC(CubicWebTC):

    def setUp(self):

        super(ResultSetTC, self).setUp()

        self.rset = ResultSet([[12, 'adim'], [13, 'syt']],

                              'Any U,L where U is CWUser, U login L',

                              description=[['CWUser', 'String'], ['Bar', 'String']])

        self.rset.req = mock_object(vreg=self.vreg)

    def compare_urls(self, url1, url2):

        info1 = urlsplit(url1)

        info2 = urlsplit(url2)

        self.assertEqual(info1[:3], info2[:3])

        if info1[3] != info2[3]:

            params1 = dict(pair.split('=') for pair in info1[3].split('&'))

            params2 = dict(pair.split('=') for pair in info1[3].split('&'))

            self.assertDictEqual(params1, params2)

    def test_pickle(self):

        del self.rset.req

        self.assertEqual(len(pickle.dumps(self.rset)), 392)

    def test_build_url(self):

        req = self.request()

        baseurl = req.base_url()

        self.compare_urls(req.build_url('view', vid='foo', rql='yo'),

                          '%sview?vid=foo&rql=yo' % baseurl)

        self.compare_urls(req.build_url('view', _restpath='task/title/go'),

                          '%stask/title/go' % baseurl)

        #self.compare_urls(req.build_url('view', _restpath='/task/title/go'),

        #                  '%stask/title/go' % baseurl)

        # empty _restpath should not crash

        self.compare_urls(req.build_url('view', _restpath=''), baseurl)

    def test_resultset_build(self):

        """test basic build of a ResultSet"""

        rs = ResultSet([1,2,3], 'CWGroup X', description=['CWGroup', 'CWGroup', 'CWGroup'])

        self.assertEqual(rs.rowcount, 3)

        self.assertEqual(rs.rows, [1,2,3])

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

    def test_resultset_limit(self):

        rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],

                       'Any U,L where U is CWUser, U login L',

                       description=[['CWUser', 'String']] * 3)

        rs.req = self.request()

        rs.vreg = self.vreg

        self.assertEqual(rs.limit(2).rows, [[12000, 'adim'], [13000, 'syt']])

        rs2 = rs.limit(2, offset=1)

        self.assertEqual(rs2.rows, [[13000, 'syt'], [14000, 'nico']])

        self.assertEqual(rs2.get_entity(0, 0).cw_row, 0)

        self.assertEqual(rs.limit(2, offset=2).rows, [[14000, 'nico']])

        self.assertEqual(rs.limit(2, offset=3).rows, [])

    def test_resultset_filter(self):

        rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],

                       'Any U,L where U is CWUser, U login L',

                       description=[['CWUser', 'String']] * 3)

        rs.req = self.request()

        rs.vreg = self.vreg

        def test_filter(entity):

            return entity.login != 'nico'

        rs2 = rs.filtered_rset(test_filter)

        self.assertEqual(len(rs2), 2)

        self.assertEqual([login for _, login in rs2], ['adim', 'syt'])

    def test_resultset_transform(self):

        rs = ResultSet([[12, 'adim'], [13, 'syt'], [14, 'nico']],

                       'Any U,L where U is CWUser, U login L',

                       description=[['CWUser', 'String']] * 3)

        rs.req = self.request()

        def test_transform(row, desc):

            return row[1:], desc[1:]

        rs2 = rs.transformed_rset(test_transform)

        self.assertEqual(len(rs2), 3)

        self.assertEqual(list(rs2), [['adim'],['syt'],['nico']])

    def test_resultset_sort(self):

        rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],

                       'Any U,L where U is CWUser, U login L',

                       description=[['CWUser', 'String']] * 3)

        rs.req = self.request()

        rs.vreg = self.vreg

        rs2 = rs.sorted_rset(lambda e:e.cw_attr_cache['login'])

        self.assertEquals(len(rs2), 3)

        self.assertEquals([login for _, login in rs2], ['adim', 'nico', 'syt'])

        rs2 = rs.sorted_rset(lambda e:e['login'])

        self.assertEqual(len(rs2), 3)

        self.assertEqual([login for _, login in rs2], ['adim', 'nico', 'syt'])

        # make sure rs is unchanged

        self.assertEqual([login for _, login in rs], ['adim', 'syt', 'nico'])

        rs2 = rs.sorted_rset(lambda e:e.cw_attr_cache['login'], reverse=True)

        self.assertEquals(len(rs2), 3)

        self.assertEquals([login for _, login in rs2], ['syt', 'nico', 'adim'])

        rs2 = rs.sorted_rset(lambda e:e['login'], reverse=True)

        self.assertEqual(len(rs2), 3)

        self.assertEqual([login for _, login in rs2], ['syt', 'nico', 'adim'])

        # make sure rs is unchanged

        self.assertEqual([login for _, login in rs], ['adim', 'syt', 'nico'])

        rs3 = rs.sorted_rset(lambda row: row[1], col=-1)

        self.assertEqual(len(rs3), 3)

        self.assertEqual([login for _, login in rs3], ['adim', 'nico', 'syt'])

        # make sure rs is unchanged

        self.assertEqual([login for _, login in rs], ['adim', 'syt', 'nico'])

    def test_resultset_split(self):

        rs = ResultSet([[12000, 'adim', u'Adim chez les pinguins'],

                        [12000, 'adim', u'Jardiner facile'],

                        [13000, 'syt',  u'Le carrelage en 42 leçons'],

                        [14000, 'nico', u'La tarte tatin en 15 minutes'],

                        [14000, 'nico', u"L'épluchage du castor commun"]],

                       'Any U, L, T WHERE U is CWUser, U login L,'\

                       'D created_by U, D title T',

                       description=[['CWUser', 'String', 'String']] * 5)

        rs.req = self.request()

        rs.vreg = self.vreg

        rsets = rs.split_rset(lambda e:e.cw_attr_cache['login'])

        self.assertEquals(len(rsets), 3)

        self.assertEquals([login for _, login,_ in rsets[0]], ['adim', 'adim'])

        self.assertEquals([login for _, login,_ in rsets[1]], ['syt'])

        self.assertEquals([login for _, login,_ in rsets[2]], ['nico', 'nico'])

        rsets = rs.split_rset(lambda e:e['login'])

        self.assertEqual(len(rsets), 3)

        self.assertEqual([login for _, login,_ in rsets[0]], ['adim', 'adim'])

        self.assertEqual([login for _, login,_ in rsets[1]], ['syt'])

        self.assertEqual([login for _, login,_ in rsets[2]], ['nico', 'nico'])

        # make sure rs is unchanged

        self.assertEqual([login for _, login,_ in rs], ['adim', 'adim', 'syt', 'nico', 'nico'])

        rsets = rs.split_rset(lambda e:e.cw_attr_cache['login'], return_dict=True)

        self.assertEquals(len(rsets), 3)

        self.assertEquals([login for _, login,_ in rsets['nico']], ['nico', 'nico'])

        self.assertEquals([login for _, login,_ in rsets['adim']], ['adim', 'adim'])

        self.assertEquals([login for _, login,_ in rsets['syt']], ['syt'])

        rsets = rs.split_rset(lambda e:e['login'], return_dict=True)

        self.assertEqual(len(rsets), 3)

        self.assertEqual([login for _, login,_ in rsets['nico']], ['nico', 'nico'])

        self.assertEqual([login for _, login,_ in rsets['adim']], ['adim', 'adim'])

        self.assertEqual([login for _, login,_ in rsets['syt']], ['syt'])

        # make sure rs is unchanged

        self.assertEqual([login for _, login,_ in rs], ['adim', 'adim', 'syt', 'nico', 'nico'])

        rsets = rs.split_rset(lambda s: s.count('d'), col=2)

        self.assertEqual(len(rsets), 2)

        self.assertEqual([title for _, _, title in rsets[0]],

                          [u"Adim chez les pinguins",

                           u"Jardiner facile",

                           u"L'épluchage du castor commun",])

        self.assertEqual([title for _, _, title in rsets[1]],

                          [u"Le carrelage en 42 leçons",

                           u"La tarte tatin en 15 minutes",])

        # make sure rs is unchanged

        self.assertEqual([title for _, _, title in rs],

                          [u'Adim chez les pinguins',

                           u'Jardiner facile',

                           u'Le carrelage en 42 leçons',

                           u'La tarte tatin en 15 minutes',

                           u"L'épluchage du castor commun"])

    def test_cached_syntax_tree(self):

        """make sure syntax tree is cached"""

        rqlst1 = self.rset.syntax_tree()

        rqlst2 = self.rset.syntax_tree()

        self.assert_(rqlst1 is rqlst2)

    def test_get_entity_simple(self):

        self.request().create_entity('CWUser', login=u'adim', upassword='adim',

                                     surname=u'di mascio', firstname=u'adrien')

        e = self.execute('Any X,T WHERE X login "adim", X surname T').get_entity(0, 0)

        self.assertEquals(e.cw_attr_cache['surname'], 'di mascio')

        self.assertRaises(KeyError, e.cw_attr_cache.__getitem__, 'firstname')

        self.assertRaises(KeyError, e.cw_attr_cache.__getitem__, 'creation_date')

        self.assertEquals(pprelcachedict(e._cw_related_cache), [])

        self.assertEqual(e['surname'], 'di mascio')

        self.assertRaises(KeyError, e.__getitem__, 'firstname')

        self.assertRaises(KeyError, e.__getitem__, 'creation_date')

        self.assertEqual(pprelcachedict(e._cw_related_cache), [])

        e.complete()

        self.assertEquals(e.cw_attr_cache['firstname'], 'adrien')

        self.assertEquals(pprelcachedict(e._cw_related_cache), [])

        self.assertEqual(e['firstname'], 'adrien')

        self.assertEqual(pprelcachedict(e._cw_related_cache), [])

    def test_get_entity_advanced(self):

        self.request().create_entity('Bookmark', title=u'zou', path=u'/view')

        self.execute('SET X bookmarked_by Y WHERE X is Bookmark, Y login "anon"')

        rset = self.execute('Any X,Y,XT,YN WHERE X bookmarked_by Y, X title XT, Y login YN')

        e = rset.get_entity(0, 0)

        self.assertEquals(e.cw_row, 0)

        self.assertEquals(e.cw_col, 0)

        self.assertEquals(e.cw_attr_cache['title'], 'zou')

        self.assertRaises(KeyError, e.cw_attr_cache.__getitem__, 'path')

        self.assertEquals(e.view('text'), 'zou')

        self.assertEquals(pprelcachedict(e._cw_related_cache), [])

        self.assertEqual(e.cw_row, 0)

        self.assertEqual(e.cw_col, 0)

        self.assertEqual(e['title'], 'zou')

        self.assertRaises(KeyError, e.__getitem__, 'path')

        self.assertEqual(e.view('text'), 'zou')

        self.assertEqual(pprelcachedict(e._cw_related_cache), [])

        e = rset.get_entity(0, 1)

        self.assertEquals(e.cw_row, 0)

        self.assertEquals(e.cw_col, 1)

        self.assertEquals(e.cw_attr_cache['login'], 'anon')

        self.assertRaises(KeyError, e.cw_attr_cache.__getitem__, 'firstname')

        self.assertEquals(pprelcachedict(e._cw_related_cache),

        self.assertEqual(e.cw_row, 0)

        self.assertEqual(e.cw_col, 1)

        self.assertEqual(e['login'], 'anon')

        self.assertRaises(KeyError, e.__getitem__, 'firstname')

        self.assertEqual(pprelcachedict(e._cw_related_cache),

                          [])

        e.complete()

        self.assertEquals(e.cw_attr_cache['firstname'], None)

        self.assertEquals(e.view('text'), 'anon')

        self.assertEquals(pprelcachedict(e._cw_related_cache),

        self.assertEqual(e['firstname'], None)

        self.assertEqual(e.view('text'), 'anon')