1 # copyright 2003-2012 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

     2 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr

     3 #

     4 # This file is part of CubicWeb.

     5 #

     6 # CubicWeb is free software: you can redistribute it and/or modify it under the

     7 # terms of the GNU Lesser General Public License as published by the Free

     8 # Software Foundation, either version 2.1 of the License, or (at your option)

     9 # any later version.

    10 #

    11 # CubicWeb is distributed in the hope that it will be useful, but WITHOUT

    12 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

    13 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more

    14 # details.

    15 #

    16 # You should have received a copy of the GNU Lesser General Public License along

    17 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.

    18 """cubicweb.server.sources.ldapusers unit and functional tests"""

    19 

    20 import os

    21 import shutil

    22 import time

    23 from os.path import join, exists

    24 import subprocess

    25 import tempfile

    26 

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

    28 

    29 from cubicweb import AuthenticationError

    30 from cubicweb.devtools.testlib import CubicWebTC

    31 from cubicweb.devtools.repotest import RQLGeneratorTC

    32 from cubicweb.devtools.httptest import get_available_port

    33 from cubicweb.devtools import get_test_db_handler

    34 

    35 from cubicweb.server.sources.ldapuser import GlobTrFunc, UnknownEid, RQL2LDAPFilter

    36 

    37 CONFIG = u'user-base-dn=ou=People,dc=cubicweb,dc=test'

    38 URL = None

    39 

    40 def create_slapd_configuration(cls):

    41     global URL

    42     slapddir = tempfile.mkdtemp('cw-unittest-ldap')

    43     config = cls.config

    44     slapdconf = join(config.apphome, "slapd.conf")

    45     confin = file(join(config.apphome, "slapd.conf.in")).read()

    46     confstream = file(slapdconf, 'w')

    47     confstream.write(confin % {'apphome': config.apphome, 'testdir': slapddir})

    48     confstream.close()

    49     # fill ldap server with some data

    50     ldiffile = join(config.apphome, "ldap_test.ldif")

    51     config.info('Initing ldap database')

    52     cmdline = "/usr/sbin/slapadd -f %s -l %s -c" % (slapdconf, ldiffile)

    53     subprocess.call(cmdline, shell=True)

    54 

    55     #ldapuri = 'ldapi://' + join(basedir, "ldapi").replace('/', '%2f')

    56     port = get_available_port(xrange(9000, 9100))

    57     host = 'localhost:%s' % port

    58     ldapuri = 'ldap://%s' % host

    59     cmdline = ["/usr/sbin/slapd", "-f",  slapdconf,  "-h",  ldapuri, "-d", "0"]

    60     config.info('Starting slapd:', ' '.join(cmdline))

    61     cls.slapd_process = subprocess.Popen(cmdline)

    62     time.sleep(0.2)

    63     if cls.slapd_process.poll() is None:

    64         config.info('slapd started with pid %s' % cls.slapd_process.pid)

    65     else:

    66         raise EnvironmentError('Cannot start slapd with cmdline="%s" (from directory "%s")' %

    67                                (" ".join(cmdline), os.getcwd()))

    68     URL = u'ldap://%s' % host

    69     return slapddir

    70 

    71 def terminate_slapd(cls):

    72     config = cls.config

    73     if cls.slapd_process and cls.slapd_process.returncode is None:

    74         config.info('terminating slapd')

    75         if hasattr(cls.slapd_process, 'terminate'):

    76             cls.slapd_process.terminate()

    77         else:

    78             import os, signal

    79             os.kill(cls.slapd_process.pid, signal.SIGTERM)

    80         cls.slapd_process.wait()

    81         config.info('DONE')

    82 

    83 class LDAPTestBase(CubicWebTC):

    84     loglevel = 'ERROR'

    85 

    86     @classmethod

    87     def setUpClass(cls):

    88         from cubicweb.cwctl import init_cmdline_log_threshold

    89         init_cmdline_log_threshold(cls.config, cls.loglevel)

    90         cls._tmpdir = create_slapd_configuration(cls)

    91 

    92     @classmethod

    93     def tearDownClass(cls):

    94         terminate_slapd(cls)

    95         try:

    96             shutil.rmtree(cls._tmpdir)

    97         except:

    98             pass

    99 

   100 class CheckWrongGroup(LDAPTestBase):

   101 

   102     def test_wrong_group(self):

   103         self.session.create_entity('CWSource', name=u'ldapuser', type=u'ldapfeed', parser=u'ldapfeed',

   104                                    url=URL, config=CONFIG)

   105         self.commit()

   106         with self.session.repo.internal_session(safe=True) as session:

   107             source = self.session.execute('CWSource S WHERE S type="ldapfeed"').get_entity(0,0)

   108             config = source.repo_source.check_config(source)

   109             # inject a bogus group here, along with at least a valid one

   110             config['user-default-group'] = ('thisgroupdoesnotexists','users')

   111             source.repo_source.update_config(source, config)

   112             session.commit(free_cnxset=False)

   113             # here we emitted an error log entry

   114             stats = source.repo_source.pull_data(session, force=True, raise_on_error=True)

   115             session.commit()

   116 

   117 class DeleteStuffFromLDAPFeedSourceTC(LDAPTestBase):

   118     test_db_id = 'ldap-feed'

   119 

   120     @classmethod

   121     def pre_setup_database(cls, session, config):

   122         session.create_entity('CWSource', name=u'ldapuser', type=u'ldapfeed', parser=u'ldapfeed',

   123                               url=URL, config=CONFIG)

   124         session.commit()

   125         with session.repo.internal_session(safe=True) as isession:

   126             lfsource = isession.repo.sources_by_uri['ldapuser']

   127             stats = lfsource.pull_data(isession, force=True, raise_on_error=True)

   128 

   129     def _pull(self):

   130         with self.session.repo.internal_session() as isession:

   131             lfsource = isession.repo.sources_by_uri['ldapuser']

   132             stats = lfsource.pull_data(isession, force=True, raise_on_error=True)

   133             isession.commit()

   134 

   135     def test_a_filter_inactivate(self):

   136         """ filtered out people should be deactivated, unable to authenticate """

   137         source = self.session.execute('CWSource S WHERE S type="ldapfeed"').get_entity(0,0)

   138         config = source.repo_source.check_config(source)

   139         # filter with adim's phone number

   140         config['user-filter'] = u'(%s=%s)' % ('telephoneNumber', '109')

   141         source.repo_source.update_config(source, config)

   142         self.commit()

   143         self._pull()

   144         self.assertRaises(AuthenticationError, self.repo.connect, 'syt', password='syt')

   145         self.assertEqual(self.execute('Any N WHERE U login "syt", '

   146                                       'U in_state S, S name N').rows[0][0],

   147                          'deactivated')

   148         self.assertEqual(self.execute('Any N WHERE U login "adim", '

   149                                       'U in_state S, S name N').rows[0][0],

   150                          'activated')

   151         # unfilter, syt should be activated again

   152         config['user-filter'] = u''

   153         source.repo_source.update_config(source, config)

   154         self.commit()

   155         self._pull()

   156         self.assertEqual(self.execute('Any N WHERE U login "syt", '

   157                                       'U in_state S, S name N').rows[0][0],

   158                          'activated')

   159         self.assertEqual(self.execute('Any N WHERE U login "adim", '

   160                                       'U in_state S, S name N').rows[0][0],

   161                          'activated')

   162 

   163     def test_delete(self):

   164         """ delete syt, pull, check deactivation, repull,

   165         readd syt, pull, check activation

   166         """

   167         uri = self.repo.sources_by_uri['ldapuser'].urls[0]

   168         deletecmd = ("ldapdelete -H %s 'uid=syt,ou=People,dc=cubicweb,dc=test' "

   169                      "-v -x -D cn=admin,dc=cubicweb,dc=test -w'cw'" % uri)

   170         os.system(deletecmd)

   171         self._pull()

   172         self.assertRaises(AuthenticationError, self.repo.connect, 'syt', password='syt')

   173         self.assertEqual(self.execute('Any N WHERE U login "syt", '

   174                                       'U in_state S, S name N').rows[0][0],

   175                          'deactivated')

   176         # check that it doesn't choke

   177         self._pull()

   178         # reset the fscking ldap thing

   179         self.tearDownClass()

   180         self.setUpClass()

   181         self._pull()

   182         self.assertEqual(self.execute('Any N WHERE U login "syt", '

   183                                       'U in_state S, S name N').rows[0][0],

   184                          'activated')

   185         # test reactivating the user isn't enough to authenticate, as the native source

   186         # refuse to authenticate user from other sources

   187         os.system(deletecmd)

   188         self._pull()

   189         user = self.execute('CWUser U WHERE U login "syt"').get_entity(0, 0)

   190         user.cw_adapt_to('IWorkflowable').fire_transition('activate')

   191         self.commit()

   192         self.assertRaises(AuthenticationError, self.repo.connect, 'syt', password='syt')

   193 

   194 class LDAPFeedSourceTC(LDAPTestBase):

   195     test_db_id = 'ldap-feed'

   196 

   197     @classmethod

   198     def pre_setup_database(cls, session, config):

   199         session.create_entity('CWSource', name=u'ldapuser', type=u'ldapfeed', parser=u'ldapfeed',

   200                               url=URL, config=CONFIG)

   201         session.commit()

   202         isession = session.repo.internal_session(safe=True)

   203         lfsource = isession.repo.sources_by_uri['ldapuser']

   204         stats = lfsource.pull_data(isession, force=True, raise_on_error=True)

   205 

   206     def setUp(self):

   207         super(LDAPFeedSourceTC, self).setUp()

   208         # ldap source url in the database may use a different port as the one

   209         # just attributed

   210         lfsource = self.repo.sources_by_uri['ldapuser']

   211         lfsource.urls = [URL]

   212 

   213     def assertMetadata(self, entity):

   214         self.assertTrue(entity.creation_date)

   215         self.assertTrue(entity.modification_date)

   216 

   217     def test_authenticate(self):

   218         source = self.repo.sources_by_uri['ldapuser']

   219         self.session.set_cnxset()

   220         # ensure we won't be logged against

   221         self.assertRaises(AuthenticationError,

   222                           source.authenticate, self.session, 'toto', 'toto')

   223         self.assertTrue(source.authenticate(self.session, 'syt', 'syt'))

   224         self.assertTrue(self.repo.connect('syt', password='syt'))

   225 

   226     def test_base(self):

   227         # check a known one

   228         rset = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'})

   229         e = rset.get_entity(0, 0)

   230         self.assertEqual(e.login, 'syt')

   231         e.complete()

   232         self.assertMetadata(e)

   233         self.assertEqual(e.firstname, None)

   234         self.assertEqual(e.surname, None)

   235         self.assertEqual(e.in_group[0].name, 'users')

   236         self.assertEqual(e.owned_by[0].login, 'syt')

   237         self.assertEqual(e.created_by, ())

   238         self.assertEqual(e.primary_email[0].address, 'Sylvain Thenault')

   239         # email content should be indexed on the user

   240         rset = self.sexecute('CWUser X WHERE X has_text "thenault"')

   241         self.assertEqual(rset.rows, [[e.eid]])

   242 

   243     def test_copy_to_system_source(self):

   244         source = self.repo.sources_by_uri['ldapuser']

   245         eid = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'})[0][0]

   246         self.sexecute('SET X cw_source S WHERE X eid %(x)s, S name "system"', {'x': eid})

   247         self.commit()

   248         source.reset_caches()

   249         rset = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'})

   250         self.assertEqual(len(rset), 1)

   251         e = rset.get_entity(0, 0)

   252         self.assertEqual(e.eid, eid)

   253         self.assertEqual(e.cw_metainformation(), {'source': {'type': u'native',

   254                                                              'uri': u'system',

   255                                                              'use-cwuri-as-url': False},

   256                                                   'type': 'CWUser',

   257                                                   'extid': None})

   258         self.assertEqual(e.cw_source[0].name, 'system')

   259         self.assertTrue(e.creation_date)

   260         self.assertTrue(e.modification_date)

   261         source.pull_data(self.session)

   262         rset = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'})

   263         self.assertEqual(len(rset), 1)

   264         self.assertTrue(self.repo.system_source.authenticate(

   265                 self.session, 'syt', password='syt'))

   266 

   267 

   268 class LDAPUserSourceTC(LDAPFeedSourceTC):

   269     test_db_id = 'ldap-user'

   270     tags = CubicWebTC.tags | Tags(('ldap'))

   271 

   272     @classmethod

   273     def pre_setup_database(cls, session, config):

   274         session.create_entity('CWSource', name=u'ldapuser', type=u'ldapuser',

   275                               url=URL, config=CONFIG)

   276         session.commit()

   277         # XXX keep it there

   278         session.execute('CWUser U')

   279 

   280     def assertMetadata(self, entity):

   281         self.assertEqual(entity.creation_date, None)

   282         self.assertEqual(entity.modification_date, None)

   283 

   284     def test_synchronize(self):

   285         source = self.repo.sources_by_uri['ldapuser']

   286         source.synchronize()

   287 

   288     def test_base(self):

   289         # check a known one

   290         rset = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'})

   291         e = rset.get_entity(0, 0)

   292         self.assertEqual(e.login, 'syt')

   293         e.complete()

   294         self.assertMetadata(e)

   295         self.assertEqual(e.firstname, None)

   296         self.assertEqual(e.surname, None)

   297         self.assertEqual(e.in_group[0].name, 'users')

   298         self.assertEqual(e.owned_by[0].login, 'syt')

   299         self.assertEqual(e.created_by, ())

   300         self.assertEqual(e.primary_email[0].address, 'Sylvain Thenault')

   301         # email content should be indexed on the user

   302         rset = self.sexecute('CWUser X WHERE X has_text "thenault"')

   303         self.assertEqual(rset.rows, [[e.eid]])

   304 

   305     def test_not(self):

   306         eid = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'})[0][0]

   307         rset = self.sexecute('CWUser X WHERE NOT X eid %s' % eid)

   308         self.assert_(rset)

   309         self.assert_(not eid in (r[0] for r in rset))

   310 

   311     def test_multiple(self):

   312         seid = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'})[0][0]

   313         aeid = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'adim'})[0][0]

   314         rset = self.sexecute('CWUser X, Y WHERE X login %(syt)s, Y login %(adim)s',

   315                             {'syt': 'syt', 'adim': 'adim'})

   316         self.assertEqual(rset.rows, [[seid, aeid]])

   317         rset = self.sexecute('Any X,Y,L WHERE X login L, X login %(syt)s, Y login %(adim)s',

   318                             {'syt': 'syt', 'adim': 'adim'})

   319         self.assertEqual(rset.rows, [[seid, aeid, 'syt']])

   320 

   321     def test_in(self):

   322         seid = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'})[0][0]

   323         aeid = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'adim'})[0][0]

   324         rset = self.sexecute('Any X,L ORDERBY L WHERE X login IN("%s", "%s"), X login L' % ('syt', 'adim'))

   325         self.assertEqual(rset.rows, [[aeid, 'adim'], [seid, 'syt']])

   326 

   327     def test_relations(self):

   328         eid = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'})[0][0]

   329         rset = self.sexecute('Any X,E WHERE X is CWUser, X login L, X primary_email E')

   330         self.assert_(eid in (r[0] for r in rset))

   331         rset = self.sexecute('Any X,L,E WHERE X is CWUser, X login L, X primary_email E')

   332         self.assert_('syt' in (r[1] for r in rset))

   333 

   334     def test_count(self):

   335         nbusers = self.sexecute('Any COUNT(X) WHERE X is CWUser')[0][0]

   336         # just check this is a possible number

   337         self.assert_(nbusers > 1, nbusers)

   338         self.assert_(nbusers < 30, nbusers)

   339 

   340     def test_upper(self):

   341         eid = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'})[0][0]

   342         rset = self.sexecute('Any UPPER(L) WHERE X eid %s, X login L' % eid)

   343         self.assertEqual(rset[0][0], 'syt'.upper())

   344 

   345     def test_unknown_attr(self):

   346         eid = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'})[0][0]

   347         rset = self.sexecute('Any L,C,M WHERE X eid %s, X login L, '

   348                             'X creation_date C, X modification_date M' % eid)

   349         self.assertEqual(rset[0][0], 'syt')

   350         self.assertEqual(rset[0][1], None)

   351         self.assertEqual(rset[0][2], None)

   352 

   353     def test_sort(self):

   354         logins = [l for l, in self.sexecute('Any L ORDERBY L WHERE X login L')]

   355         self.assertEqual(logins, sorted(logins))

   356 

   357     def test_lower_sort(self):

   358         logins = [l for l, in self.sexecute('Any L ORDERBY lower(L) WHERE X login L')]

   359         self.assertEqual(logins, sorted(logins))

   360 

   361     def test_or(self):

   362         rset = self.sexecute('DISTINCT Any X WHERE X login %(login)s OR (X in_group G, G name "managers")',

   363                             {'login': 'syt'})

   364         self.assertEqual(len(rset), 2, rset.rows) # syt + admin

   365 

   366     def test_nonregr_set_owned_by(self):

   367         # test that when a user coming from ldap is triggering a transition

   368         # the related TrInfo has correct owner information

   369         self.sexecute('SET X in_group G WHERE X login %(syt)s, G name "managers"', {'syt': 'syt'})

   370         self.commit()

   371         syt = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'syt'}).get_entity(0, 0)

   372         self.assertEqual([g.name for g in syt.in_group], ['managers', 'users'])

   373         cnx = self.login('syt', password='syt')

   374         cu = cnx.cursor()

   375         adim = cu.execute('CWUser X WHERE X login %(login)s', {'login': 'adim'}).get_entity(0, 0)

   376         iworkflowable = adim.cw_adapt_to('IWorkflowable')

   377         iworkflowable.fire_transition('deactivate')

   378         try:

   379             cnx.commit()

   380             adim.cw_clear_all_caches()

   381             self.assertEqual(adim.in_state[0].name, 'deactivated')

   382             trinfo = iworkflowable.latest_trinfo()

   383             self.assertEqual(trinfo.owned_by[0].login, 'syt')

   384             # select from_state to skip the user's creation TrInfo

   385             rset = self.sexecute('Any U ORDERBY D DESC WHERE WF wf_info_for X,'

   386                                 'WF creation_date D, WF from_state FS,'

   387                                 'WF owned_by U?, X eid %(x)s',

   388                                 {'x': adim.eid})

   389             self.assertEqual(rset.rows, [[syt.eid]])

   390         finally:

   391             # restore db state

   392             self.restore_connection()

   393             adim = self.sexecute('CWUser X WHERE X login %(login)s', {'login': 'adim'}).get_entity(0, 0)

   394             adim.cw_adapt_to('IWorkflowable').fire_transition('activate')

   395             self.sexecute('DELETE X in_group G WHERE X login %(syt)s, G name "managers"', {'syt': 'syt'})

   396 

   397     def test_same_column_names(self):

   398         self.sexecute('Any X, Y WHERE X copain Y, X login "comme", Y login "cochon"')

   399 

   400     def test_multiple_entities_from_different_sources(self):

   401         req = self.request()

   402         self.create_user(req, 'cochon')

   403         self.assertTrue(self.sexecute('Any X,Y WHERE X login %(syt)s, Y login "cochon"', {'syt': 'syt'}))

   404 

   405     def test_exists1(self):

   406         self.session.set_cnxset()

   407         self.session.create_entity('CWGroup', name=u'bougloup1')

   408         self.session.create_entity('CWGroup', name=u'bougloup2')

   409         self.sexecute('SET U in_group G WHERE G name ~= "bougloup%", U login "admin"')

   410         self.sexecute('SET U in_group G WHERE G name = "bougloup1", U login %(syt)s', {'syt': 'syt'})

   411         rset = self.sexecute('Any L,SN ORDERBY L WHERE X in_state S, '

   412                              'S name SN, X login L, EXISTS(X in_group G, G name ~= "bougloup%")')

   413         self.assertEqual(rset.rows, [['admin', 'activated'], ['syt', 'activated']])

   414 

   415     def test_exists2(self):

   416         req = self.request()

   417         self.create_user(req, 'comme')

   418         self.create_user(req, 'cochon')

   419         self.sexecute('SET X copain Y WHERE X login "comme", Y login "cochon"')

   420         rset = self.sexecute('Any GN ORDERBY GN WHERE X in_group G, G name GN, '

   421                              '(G name "managers" OR EXISTS(X copain T, T login in ("comme", "cochon")))')

   422         self.assertEqual(rset.rows, [['managers'], ['users']])

   423 

   424     def test_exists3(self):

   425         req = self.request()

   426         self.create_user(req, 'comme')

   427         self.create_user(req, 'cochon')

   428         self.sexecute('SET X copain Y WHERE X login "comme", Y login "cochon"')

   429         self.assertTrue(self.sexecute('Any X, Y WHERE X copain Y, X login "comme", Y login "cochon"'))

   430         self.sexecute('SET X copain Y WHERE X login %(syt)s, Y login "cochon"', {'syt': 'syt'})

   431         self.assertTrue(self.sexecute('Any X, Y WHERE X copain Y, X login %(syt)s, Y login "cochon"', {'syt': 'syt'}))

   432         rset = self.sexecute('Any GN,L WHERE X in_group G, X login L, G name GN, G name "managers" '

   433                              'OR EXISTS(X copain T, T login in ("comme", "cochon"))')

   434         self.assertEqual(sorted(rset.rows), [['managers', 'admin'], ['users', 'comme'], ['users', 'syt']])

   435 

   436     def test_exists4(self):

   437         req = self.request()

   438         self.create_user(req, 'comme')

   439         self.create_user(req, 'cochon', groups=('users', 'guests'))

   440         self.create_user(req, 'billy')

   441         self.sexecute('SET X copain Y WHERE X login "comme", Y login "cochon"')

   442         self.sexecute('SET X copain Y WHERE X login "cochon", Y login "cochon"')

   443         self.sexecute('SET X copain Y WHERE X login "comme", Y login "billy"')

   444         self.sexecute('SET X copain Y WHERE X login %(syt)s, Y login "billy"', {'syt': 'syt'})

   445         # search for group name, login where

   446         #   CWUser copain with "comme" or "cochon" AND same login as the copain

   447         # OR

   448         #   CWUser in_state activated AND not copain with billy

   449         #

   450         # SO we expect everybody but "comme" and "syt"

   451         rset= self.sexecute('Any GN,L WHERE X in_group G, X login L, G name GN, '

   452                            'EXISTS(X copain T, T login L, T login in ("comme", "cochon")) OR '

   453                            'EXISTS(X in_state S, S name "activated", NOT X copain T2, T2 login "billy")')

   454         all = self.sexecute('Any GN, L WHERE X in_group G, X login L, G name GN')

   455         all.rows.remove(['users', 'comme'])

   456         all.rows.remove(['users', 'syt'])

   457         self.assertEqual(sorted(rset.rows), sorted(all.rows))

   458 

   459     def test_exists5(self):

   460         req = self.request()

   461         self.create_user(req, 'comme')

   462         self.create_user(req, 'cochon', groups=('users', 'guests'))

   463         self.create_user(req, 'billy')

   464         self.sexecute('SET X copain Y WHERE X login "comme", Y login "cochon"')

   465         self.sexecute('SET X copain Y WHERE X login "cochon", Y login "cochon"')

   466         self.sexecute('SET X copain Y WHERE X login "comme", Y login "billy"')

   467         self.sexecute('SET X copain Y WHERE X login %(syt)s, Y login "cochon"', {'syt': 'syt'})

   468         rset= self.sexecute('Any L WHERE X login L, '

   469                            'EXISTS(X copain T, T login in ("comme", "cochon")) AND '

   470                            'NOT EXISTS(X copain T2, T2 login "billy")')

   471         self.assertEqual(sorted(rset.rows), [['cochon'], ['syt']])

   472         rset= self.sexecute('Any GN,L WHERE X in_group G, X login L, G name GN, '

   473                            'EXISTS(X copain T, T login in ("comme", "cochon")) AND '

   474                            'NOT EXISTS(X copain T2, T2 login "billy")')

   475         self.assertEqual(sorted(rset.rows), [['guests', 'cochon'],

   476                                               ['users', 'cochon'],

   477                                               ['users', 'syt']])

   478 

   479     def test_cd_restriction(self):

   480         rset = self.sexecute('CWUser X WHERE X creation_date > "2009-02-01"')

   481         # admin/anon but no ldap user since it doesn't support creation_date

   482         self.assertEqual(sorted(e.login for e in rset.entities()),

   483                           ['admin', 'anon'])

   484 

   485     def test_union(self):

   486         afeids = self.sexecute('State X')

   487         ueids = self.sexecute('CWUser X')

   488         rset = self.sexecute('(Any X WHERE X is State) UNION (Any X WHERE X is CWUser)')

   489         self.assertEqual(sorted(r[0] for r in rset.rows),

   490                           sorted(r[0] for r in afeids + ueids))

   491 

   492     def _init_security_test(self):

   493         req = self.request()

   494         self.create_user(req, 'iaminguestsgrouponly', groups=('guests',))

   495         cnx = self.login('iaminguestsgrouponly')

   496         return cnx.cursor()

   497 

   498     def test_security1(self):

   499         cu = self._init_security_test()

   500         rset = cu.execute('CWUser X WHERE X login %(login)s', {'login': 'syt'})

   501         self.assertEqual(rset.rows, [])

   502         rset = cu.execute('Any X WHERE X login "iaminguestsgrouponly"')

   503         self.assertEqual(len(rset.rows), 1)

   504 

   505     def test_security2(self):

   506         cu = self._init_security_test()

   507         rset = cu.execute('Any X WHERE X has_text %(syt)s', {'syt': 'syt'})

   508         self.assertEqual(rset.rows, [])

   509         rset = cu.execute('Any X WHERE X has_text "iaminguestsgrouponly"')

   510         self.assertEqual(len(rset.rows), 1)

   511 

   512     def test_security3(self):

   513         cu = self._init_security_test()

   514         rset = cu.execute('Any F WHERE X has_text %(syt)s, X firstname F', {'syt': 'syt'})

   515         self.assertEqual(rset.rows, [])

   516         rset = cu.execute('Any F WHERE X has_text "iaminguestsgrouponly", X firstname F')

   517         self.assertEqual(rset.rows, [[None]])

   518 

   519     def test_nonregr1(self):

   520         self.sexecute('Any X,AA ORDERBY AA DESC WHERE E eid %(x)s, E owned_by X, '

   521                      'X modification_date AA',

   522                      {'x': self.session.user.eid})

   523 

   524     def test_nonregr2(self):

   525         self.sexecute('Any X,L,AA WHERE E eid %(x)s, E owned_by X, '

   526                      'X login L, X modification_date AA',

   527                      {'x': self.session.user.eid})

   528 

   529     def test_nonregr3(self):

   530         self.sexecute('Any X,AA ORDERBY AA DESC WHERE E eid %(x)s, '

   531                      'X modification_date AA',

   532                      {'x': self.session.user.eid})

   533 

   534     def test_nonregr4(self):

   535         emaileid = self.sexecute('INSERT EmailAddress X: X address "toto@logilab.org"')[0][0]

   536         self.sexecute('Any X,AA WHERE X use_email Y, Y eid %(x)s, X modification_date AA',

   537                      {'x': emaileid})

   538 

   539     def test_nonregr5(self):

   540         # original jpl query:

   541         # Any X, NOW - CD, P WHERE P is Project, U interested_in P, U is CWUser,

   542         # U login "sthenault", X concerns P, X creation_date CD ORDERBY CD DESC LIMIT 5

   543         rql = ('Any X, NOW - CD, P ORDERBY CD DESC LIMIT 5 WHERE P bookmarked_by U, '

   544                'U login "%s", P is X, X creation_date CD') % self.session.user.login

   545         self.sexecute(rql, )#{'x': })

   546 

   547     def test_nonregr6(self):

   548         self.sexecute('Any B,U,UL GROUPBY B,U,UL WHERE B created_by U?, B is File '

   549                      'WITH U,UL BEING (Any U,UL WHERE ME eid %(x)s, (EXISTS(U identity ME) '

   550                      'OR (EXISTS(U in_group G, G name IN("managers", "staff")))) '

   551                      'OR (EXISTS(U in_group H, ME in_group H, NOT H name "users")), U login UL, U is CWUser)',

   552                      {'x': self.session.user.eid})

   553 

   554 class GlobTrFuncTC(TestCase):

   555 

   556     def test_count(self):

   557         trfunc = GlobTrFunc('count', 0)

   558         res = trfunc.apply([[1], [2], [3], [4]])

   559         self.assertEqual(res, [[4]])

   560         trfunc = GlobTrFunc('count', 1)

   561         res = trfunc.apply([[1, 2], [2, 4], [3, 6], [1, 5]])

   562         self.assertEqual(res, [[1, 2], [2, 1], [3, 1]])

   563 

   564     def test_sum(self):

   565         trfunc = GlobTrFunc('sum', 0)

   566         res = trfunc.apply([[1], [2], [3], [4]])

   567         self.assertEqual(res, [[10]])

   568         trfunc = GlobTrFunc('sum', 1)

   569         res = trfunc.apply([[1, 2], [2, 4], [3, 6], [1, 5]])

   570         self.assertEqual(res, [[1, 7], [2, 4], [3, 6]])

   571 

   572     def test_min(self):

   573         trfunc = GlobTrFunc('min', 0)

   574         res = trfunc.apply([[1], [2], [3], [4]])

   575         self.assertEqual(res, [[1]])

   576         trfunc = GlobTrFunc('min', 1)

   577         res = trfunc.apply([[1, 2], [2, 4], [3, 6], [1, 5]])

   578         self.assertEqual(res, [[1, 2], [2, 4], [3, 6]])

   579 

   580     def test_max(self):

   581         trfunc = GlobTrFunc('max', 0)

   582         res = trfunc.apply([[1], [2], [3], [4]])

   583         self.assertEqual(res, [[4]])

   584         trfunc = GlobTrFunc('max', 1)

   585         res = trfunc.apply([[1, 2], [2, 4], [3, 6], [1, 5]])

   586         self.assertEqual(res, [[1, 5], [2, 4], [3, 6]])

   587 

   588 

   589 class RQL2LDAPFilterTC(RQLGeneratorTC):

   590 

   591     tags = RQLGeneratorTC.tags | Tags(('ldap'))

   592 

   593     @property

   594     def schema(self):

   595         """return the application schema"""

   596         return self._schema

   597 

   598     def setUp(self):

   599         self.handler = get_test_db_handler(LDAPUserSourceTC.config)

   600         self.handler.build_db_cache('ldap-user', LDAPUserSourceTC.pre_setup_database)

   601         self.handler.restore_database('ldap-user')

   602         self._repo = repo = self.handler.get_repo()

   603         self._schema = repo.schema

   604         super(RQL2LDAPFilterTC, self).setUp()

   605         ldapsource = repo.sources[-1]

   606         self.cnxset = repo._get_cnxset()

   607         session = mock_object(cnxset=self.cnxset)

   608         self.o = RQL2LDAPFilter(ldapsource, session)

   609         self.ldapclasses = ''.join(ldapsource.base_filters)

   610 

   611     def tearDown(self):

   612         self._repo.turn_repo_off()

   613         super(RQL2LDAPFilterTC, self).tearDown()

   614 

   615     def test_base(self):

   616         rqlst = self._prepare('CWUser X WHERE X login "toto"').children[0]

   617         self.assertEqual(self.o.generate(rqlst, 'X')[1],

   618                           '(&%s(uid=toto))' % self.ldapclasses)

   619 

   620     def test_kwargs(self):

   621         rqlst = self._prepare('CWUser X WHERE X login %(x)s').children[0]

   622         self.o._args = {'x': "toto"}

   623         self.assertEqual(self.o.generate(rqlst, 'X')[1],

   624                           '(&%s(uid=toto))' % self.ldapclasses)

   625 

   626     def test_get_attr(self):

   627         rqlst = self._prepare('Any X WHERE E firstname X, E eid 12').children[0]

   628         self.assertRaises(UnknownEid, self.o.generate, rqlst, 'E')

   629 

   630 

   631 if __name__ == '__main__':

   632     unittest_main()