test/unittest_rset.py
author Rémi Cardona <remi.cardona@logilab.fr>
Mon, 21 Sep 2015 18:04:22 +0200
changeset 10722 1d0b5aef7523
parent 10691 af266f27c4d5
child 10997 da712d3f0601
permissions -rw-r--r--
[tests] Stop counting pickle bytes in ResultSet tests Between python 2's pickle and cPickle implementations, the various protocol versions added in python 3, the length of the pickled string is completely unpredictable. Instead, do a simple round trip and test a few attributes.

# coding: utf-8
# copyright 2003-2014 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 module cubicweb.utils"""

from six import string_types
from six.moves import cPickle as pickle
from six.moves.urllib.parse import urlsplit

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
from cubicweb import NoResultError, MultipleResultsError


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], 0, 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 idx, relations in results.items():
                result = list(attr_desc_iterator(parse(rql).children[0], idx, idx))
                self.assertEqual(result, relations)

    def test_subquery_callfunc(self):
        rql = ('Any A,B,C,COUNT(D) GROUPBY A,B,C WITH A,B,C,D BEING '
               '(Any YEAR(CD), MONTH(CD), S, X WHERE X is CWUser, X creation_date CD, X in_state S)')
        rqlst = parse(rql)
        select, col = rqlst.locate_subquery(2, 'CWUser', None)
        result = list(attr_desc_iterator(select, col, 2))
        self.assertEqual(result, [])

    def test_subquery_callfunc_2(self):
        rql = ('Any X,S,L WHERE X in_state S WITH X, L BEING (Any X,MAX(L) GROUPBY X WHERE X is CWUser, T wf_info_for X, T creation_date L)')
        rqlst = parse(rql)
        select, col = rqlst.locate_subquery(0, 'CWUser', None)
        result = list(attr_desc_iterator(select, col, 0))
        self.assertEqual(result, [(1, 'in_state', 'subject')])


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
        rs2 = pickle.loads(pickle.dumps(self.rset))
        self.assertEqual(self.rset.rows, rs2.rows)
        self.assertEqual(self.rset.rowcount, rs2.rowcount)
        self.assertEqual(self.rset.rql, rs2.rql)
        self.assertEqual(self.rset.description, rs2.description)

    def test_build_url(self):
        with self.admin_access.web_request() as req:
            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)
            self.assertNotIn('https', req.build_url('view', vid='foo', rql='yo',
                                                      __secure__=True))
            try:
                self.config.global_set_option('https-url', 'https://testing.fr/')
                self.assertTrue('https', req.build_url('view', vid='foo', rql='yo',
                                                         __secure__=True))
                self.compare_urls(req.build_url('view', vid='foo', rql='yo',
                                                __secure__=True),
                                  '%sview?vid=foo&rql=yo' % req.base_url(secure=True))
            finally:
                self.config.global_set_option('https-url', None)


    def test_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_limit(self):
        rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],
                       'Any U,L where U is CWUser, U login L',
                       description=[['CWUser', 'String']] * 3)
        with self.admin_access.web_request() as req:
            rs.req = req
            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_limit_2(self):
        with self.admin_access.web_request() as req:
            # drop user from cache for the sake of this test
            req.drop_entity_cache(req.user.eid)
            rs = req.execute('Any E,U WHERE E is CWEType, E created_by U')
            # get entity on row 9. This will fill its created_by relation cache,
            # with cwuser on row 9 as well
            e1 = rs.get_entity(9, 0)
            # get entity on row 10. This will fill its created_by relation cache,
            # with cwuser built on row 9
            e2 = rs.get_entity(10, 0)
            # limit result set from row 10
            rs.limit(1, 10, inplace=True)
            # get back eid
            e = rs.get_entity(0, 0)
            self.assertTrue(e2 is e)
            # rs.limit has properly removed cwuser for request cache, but it's
            # still referenced by e/e2 relation cache
            u = e.created_by[0]
            # now ensure this doesn't trigger IndexError because cwuser.cw_row is 9
            # while now rset has only one row
            u.cw_rset[u.cw_row]

    def test_filter(self):
        rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],
                       'Any U,L where U is CWUser, U login L',
                       description=[['CWUser', 'String']] * 3)
        with self.admin_access.web_request() as req:
            rs.req = req
            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'])
            self.assertEqual(rs2.description, rs.description[1:])

    def test_transform(self):
        rs = ResultSet([[12, 'adim'], [13, 'syt'], [14, 'nico']],
                       'Any U,L where U is CWUser, U login L',
                       description=[['CWUser', 'String']] * 3)
        with self.admin_access.web_request() as req:
            rs.req = req
            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_sort(self):
        rs = ResultSet([[12000, 'adim'], [13000, 'syt'], [14000, 'nico']],
                       'Any U,L where U is CWUser, U login L',
                       description=[['CWUser', 'String']] * 3)
        with self.admin_access.web_request() as req:
            rs.req = req
            rs.vreg = self.vreg

            rs2 = rs.sorted_rset(lambda e:e.cw_attr_cache['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.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_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)
        with self.admin_access.web_request() as req:
            rs.req = req
            rs.vreg = self.vreg
            rsets = rs.split_rset(lambda e:e.cw_attr_cache['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.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.assertIs(rqlst1, rqlst2)

    def test_get_entity_simple(self):
        with self.admin_access.web_request() as req:
            req.create_entity('CWUser', login=u'adim', upassword='adim',
                                         surname=u'di mascio', firstname=u'adrien')
            req.cnx.drop_entity_cache()
            e = req.execute('Any X,T WHERE X login "adim", X surname T').get_entity(0, 0)
            self.assertEqual(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.assertEqual(pprelcachedict(e._cw_related_cache), [])
            e.complete()
            self.assertEqual(e.cw_attr_cache['firstname'], 'adrien')
            self.assertEqual(pprelcachedict(e._cw_related_cache), [])

    def test_get_entity_advanced(self):
        with self.admin_access.web_request() as req:
            req.create_entity('Bookmark', title=u'zou', path=u'/view')
            req.cnx.drop_entity_cache()
            req.execute('SET X bookmarked_by Y WHERE X is Bookmark, Y login "anon"')
            rset = req.execute('Any X,Y,XT,YN WHERE X bookmarked_by Y, X title XT, Y login YN')

            e = rset.get_entity(0, 0)
            self.assertEqual(e.cw_row, 0)
            self.assertEqual(e.cw_col, 0)
            self.assertEqual(e.cw_attr_cache['title'], 'zou')
            self.assertRaises(KeyError, e.cw_attr_cache.__getitem__, 'path')
            self.assertEqual(e.view('text'), 'zou')
            self.assertEqual(pprelcachedict(e._cw_related_cache), [])

            e = rset.get_entity(0, 1)
            self.assertEqual(e.cw_row, 0)
            self.assertEqual(e.cw_col, 1)
            self.assertEqual(e.cw_attr_cache['login'], 'anon')
            self.assertRaises(KeyError, e.cw_attr_cache.__getitem__, 'firstname')
            self.assertEqual(pprelcachedict(e._cw_related_cache),
                              [])
            e.complete()
            self.assertEqual(e.cw_attr_cache['firstname'], None)
            self.assertEqual(e.view('text'), 'anon')
            self.assertEqual(pprelcachedict(e._cw_related_cache),
                              [])

            self.assertRaises(NotAnEntity, rset.get_entity, 0, 2)
            self.assertRaises(NotAnEntity, rset.get_entity, 0, 3)

    def test_get_entity_relation_cache_compt(self):
        with self.admin_access.web_request() as req:
            rset = req.execute('Any X,S WHERE X in_state S, X login "anon"')
            e = rset.get_entity(0, 0)
            seid = req.execute('State X WHERE X name "activated"')[0][0]
            # for_user / in_group are prefetched in CWUser __init__, in_state should
            # be filed from our query rset
            self.assertEqual(pprelcachedict(e._cw_related_cache),
                              [('in_state_subject', [seid])])

    def test_get_entity_advanced_prefilled_cache(self):
        with self.admin_access.web_request() as req:
            e = req.create_entity('Bookmark', title=u'zou', path=u'path')
            req.cnx.commit()
            rset = req.execute('Any X,U,S,XT,UL,SN WHERE X created_by U, U in_state S, '
                                'X title XT, S name SN, U login UL, X eid %s' % e.eid)
            e = rset.get_entity(0, 0)
            self.assertEqual(e.cw_attr_cache['title'], 'zou')
            self.assertEqual(pprelcachedict(e._cw_related_cache),
                              [('created_by_subject', [req.user.eid])])
            # first level of recursion
            u = e.created_by[0]
            self.assertEqual(u.cw_attr_cache['login'], 'admin')
            self.assertRaises(KeyError, u.cw_attr_cache.__getitem__, 'firstname')
            # second level of recursion
            s = u.in_state[0]
            self.assertEqual(s.cw_attr_cache['name'], 'activated')
            self.assertRaises(KeyError, s.cw_attr_cache.__getitem__, 'description')


    def test_get_entity_cache_with_left_outer_join(self):
        with self.admin_access.web_request() as req:
            eid = req.execute('INSERT CWUser E: E login "joe", E upassword "joe", E in_group G '
                               'WHERE G name "users"')[0][0]
            rset = req.execute('Any X,E WHERE X eid %(x)s, X primary_email E?', {'x': eid})
            e = rset.get_entity(0, 0)
            # if any of the assertion below fails with a KeyError, the relation is not cached
            # related entities should be an empty list
            self.assertEqual(e._cw_related_cache['primary_email_subject'][True], ())
            # related rset should be an empty rset
            cached = e._cw_related_cache['primary_email_subject'][False]
            self.assertIsInstance(cached, ResultSet)
            self.assertEqual(cached.rowcount, 0)


    def test_get_entity_union(self):
        with self.admin_access.web_request() as req:
            e = req.create_entity('Bookmark', title=u'manger', path=u'path')
            req.cnx.drop_entity_cache()
            rset = req.execute('Any X,N ORDERBY N WITH X,N BEING '
                                '((Any X,N WHERE X is Bookmark, X title N)'
                                ' UNION '
                                ' (Any X,N WHERE X is CWGroup, X name N))')
            expected = (('CWGroup', 'guests'), ('CWGroup', 'managers'),
                        ('Bookmark', 'manger'), ('CWGroup', 'owners'),
                        ('CWGroup', 'users'))
            for entity in rset.entities(): # test get_entity for each row actually
                etype, n = expected[entity.cw_row]
                self.assertEqual(entity.cw_etype, etype)
                attr = etype == 'Bookmark' and 'title' or 'name'
                self.assertEqual(entity.cw_attr_cache[attr], n)

    def test_one(self):
        with self.admin_access.web_request() as req:
            req.create_entity('CWUser', login=u'cdevienne',
                                         upassword=u'cdevienne',
                                         surname=u'de Vienne',
                                         firstname=u'Christophe')
            e = req.execute('Any X WHERE X login "cdevienne"').one()

            self.assertEqual(e.surname, u'de Vienne')

            e = req.execute(
                'Any X, N WHERE X login "cdevienne", X surname N').one()
            self.assertEqual(e.surname, u'de Vienne')

            e = req.execute(
                'Any N, X WHERE X login "cdevienne", X surname N').one(col=1)
            self.assertEqual(e.surname, u'de Vienne')

    def test_one_no_rows(self):
        with self.admin_access.web_request() as req:
            with self.assertRaises(NoResultError):
                req.execute('Any X WHERE X login "patanok"').one()

    def test_one_multiple_rows(self):
        with self.admin_access.web_request() as req:
            req.create_entity(
                'CWUser', login=u'cdevienne', upassword=u'cdevienne',
                surname=u'de Vienne', firstname=u'Christophe')

            req.create_entity(
                'CWUser', login=u'adim', upassword='adim', surname=u'di mascio',
                firstname=u'adrien')

            with self.assertRaises(MultipleResultsError):
                req.execute('Any X WHERE X is CWUser').one()

    def test_related_entity_optional(self):
        with self.admin_access.web_request() as req:
            e = req.create_entity('Bookmark', title=u'aaaa', path=u'path')
            rset = req.execute('Any B,U,L WHERE B bookmarked_by U?, U login L')
            entity, rtype = rset.related_entity(0, 2)
            self.assertEqual(entity, None)
            self.assertEqual(rtype, None)

    def test_related_entity_union_subquery_1(self):
        with self.admin_access.web_request() as req:
            e = req.create_entity('Bookmark', title=u'aaaa', path=u'path')
            rset = req.execute('Any X,N ORDERBY N WITH X,N BEING '
                                '((Any X,N WHERE X is CWGroup, X name N)'
                                ' UNION '
                                ' (Any X,N WHERE X is Bookmark, X title N))')
            entity, rtype = rset.related_entity(0, 1)
            self.assertEqual(entity.eid, e.eid)
            self.assertEqual(rtype, 'title')
            self.assertEqual(entity.title, 'aaaa')
            entity, rtype = rset.related_entity(1, 1)
            self.assertEqual(entity.cw_etype, 'CWGroup')
            self.assertEqual(rtype, 'name')
            self.assertEqual(entity.name, 'guests')

    def test_related_entity_union_subquery_2(self):
        with self.admin_access.web_request() as req:
            e = req.create_entity('Bookmark', title=u'aaaa', path=u'path')
            rset = req.execute('Any X,N ORDERBY N WHERE X is Bookmark WITH X,N BEING '
                                '((Any X,N WHERE X is CWGroup, X name N)'
                                ' UNION '
                                ' (Any X,N WHERE X is Bookmark, X title N))')
            entity, rtype = rset.related_entity(0, 1)
            self.assertEqual(entity.eid, e.eid)
            self.assertEqual(rtype, 'title')
            self.assertEqual(entity.title, 'aaaa')

    def test_related_entity_union_subquery_3(self):
        with self.admin_access.web_request() as req:
            e = req.create_entity('Bookmark', title=u'aaaa', path=u'path')
            rset = req.execute('Any X,N ORDERBY N WITH N,X BEING '
                                '((Any N,X WHERE X is CWGroup, X name N)'
                                ' UNION '
                                ' (Any N,X WHERE X is Bookmark, X title N))')
            entity, rtype = rset.related_entity(0, 1)
            self.assertEqual(entity.eid, e.eid)
            self.assertEqual(rtype, 'title')
            self.assertEqual(entity.title, 'aaaa')

    def test_related_entity_union_subquery_4(self):
        with self.admin_access.web_request() as req:
            e = req.create_entity('Bookmark', title=u'aaaa', path=u'path')
            rset = req.execute('Any X,X, N ORDERBY N WITH X,N BEING '
                                '((Any X,N WHERE X is CWGroup, X name N)'
                                ' UNION '
                                ' (Any X,N WHERE X is Bookmark, X title N))')
            entity, rtype = rset.related_entity(0, 2)
            self.assertEqual(entity.eid, e.eid)
            self.assertEqual(rtype, 'title')
            self.assertEqual(entity.title, 'aaaa')

    def test_related_entity_trap_subquery(self):
        with self.admin_access.web_request() as req:
            req.create_entity('Bookmark', title=u'test bookmark', path=u'')
            req.execute('SET B bookmarked_by U WHERE U login "admin"')
            rset = req.execute('Any B,T,L WHERE B bookmarked_by U, U login L '
                                'WITH B,T BEING (Any B,T WHERE B is Bookmark, B title T)')
            rset.related_entity(0, 2)

    def test_related_entity_subquery_outerjoin(self):
        with self.admin_access.web_request() as req:
            rset = req.execute('Any X,S,L WHERE X in_state S '
                                'WITH X, L BEING (Any X,MAX(L) GROUPBY X '
                                'WHERE X is CWUser, T? wf_info_for X, T creation_date L)')
            self.assertEqual(len(rset), 2)
            rset.related_entity(0, 1)
            rset.related_entity(0, 2)

    def test_entities(self):
        with self.admin_access.web_request() as req:
            rset = req.execute('Any U,G WHERE U in_group G')
            # make sure we have at least one element
            self.assertTrue(rset)
            self.assertEqual(set(e.e_schema.type for e in rset.entities(0)),
                              set(['CWUser',]))
            self.assertEqual(set(e.e_schema.type for e in rset.entities(1)),
                              set(['CWGroup',]))

    def test_iter_rows_with_entities(self):
        with self.admin_access.web_request() as req:
            rset = req.execute('Any U,UN,G,GN WHERE U in_group G, U login UN, G name GN')
            # make sure we have at least one element
            self.assertTrue(rset)
            out = list(rset.iter_rows_with_entities())[0]
            self.assertEqual( out[0].login, out[1] )
            self.assertEqual( out[2].name, out[3] )

    def test_printable_rql(self):
        with self.admin_access.web_request() as req:
            rset = req.execute(u'CWEType X WHERE X final FALSE')
            self.assertEqual(rset.printable_rql(),
                              'Any X WHERE X final FALSE, X is CWEType')

    def test_searched_text(self):
        with self.admin_access.web_request() as req:
            rset = req.execute(u'Any X WHERE X has_text "foobar"')
            self.assertEqual(rset.searched_text(), 'foobar')
            rset = req.execute(u'Any X WHERE X has_text %(text)s', {'text' : 'foo'})
            self.assertEqual(rset.searched_text(), 'foo')

    def test_union_limited_rql(self):
        with self.admin_access.web_request() as req:
            rset = req.execute('(Any X,N WHERE X is Bookmark, X title N)'
                                ' UNION '
                                '(Any X,N WHERE X is CWGroup, X name N)')
            rset.limit(2, 10, inplace=True)
            self.assertEqual(rset.limited_rql(),
                              'Any A,B LIMIT 2 OFFSET 10 '
                              'WITH A,B BEING ('
                              '(Any X,N WHERE X is Bookmark, X title N) '
                              'UNION '
                              '(Any X,N WHERE X is CWGroup, X name N)'
                              ')')

    def test_count_users_by_date(self):
        with self.admin_access.web_request() as req:
            rset = req.execute('Any D, COUNT(U) GROUPBY D WHERE U is CWUser, U creation_date D')
            self.assertEqual(rset.related_entity(0,0), (None, None))

    def test_str(self):
        with self.admin_access.web_request() as req:
            rset = req.execute('(Any X,N WHERE X is CWGroup, X name N)')
            self.assertIsInstance(str(rset), string_types)
            self.assertEqual(len(str(rset).splitlines()), 1)

    def test_repr(self):
        with self.admin_access.web_request() as req:
            rset = req.execute('(Any X,N WHERE X is CWGroup, X name N)')
            self.assertIsInstance(repr(rset), string_types)
            self.assertTrue(len(repr(rset).splitlines()) > 1)

            rset = req.execute('(Any X WHERE X is CWGroup, X name "managers")')
            self.assertIsInstance(str(rset), string_types)
            self.assertEqual(len(str(rset).splitlines()), 1)

    def test_slice(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)
        self.assertEqual(rs[1::2],
            [[12000, 'adim', u'Jardiner facile'],
             [14000, 'nico', u'La tarte tatin en 15 minutes']])

    def test_nonregr_symmetric_relation(self):
        # see https://www.cubicweb.org/ticket/4739253
        with self.admin_access.client_cnx() as cnx:
            p1 = cnx.create_entity('Personne', nom=u'sylvain')
            cnx.create_entity('Personne', nom=u'denis', connait=p1)
            cnx.commit()
            rset = cnx.execute('Any X,Y WHERE X connait Y')
            rset.get_entity(0, 1) # used to raise KeyError

if __name__ == '__main__':
    unittest_main()