# 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/>.

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

import os

import shutil

import time

from os.path import abspath, join, exists

import subprocess

from socket import socket, error as socketerror

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

from cubicweb.devtools.testlib import CubicWebTC

from cubicweb.devtools.repotest import RQLGeneratorTC

from cubicweb.devtools.httptest import get_available_port

from cubicweb.devtools import get_test_db_handler

from cubicweb.server.sources.ldapuser import *

SYT = 'syt'

SYT_EMAIL = 'Sylvain Thenault'

ADIM = 'adim'

CONFIG = u'''host=%s

user-base-dn=ou=People,dc=cubicweb,dc=test

user-scope=ONELEVEL

user-classes=top,posixAccount

user-login-attr=uid

user-default-group=users

user-attrs-map=gecos:email,uid:login

'''

def nopwd_authenticate(self, session, login, password):

    """used to monkey patch the source to get successful authentication without

    upassword checking

    """

    assert login, 'no login!'

    searchfilter = [filter_format('(%s=%s)', (self.user_login_attr, login))]

    searchfilter.extend(self.base_filters)

    searchstr = '(&%s)' % ''.join(searchfilter)

    # first search the user

    try:

        user = self._search(session, self.user_base_dn, self.user_base_scope,

                            searchstr)[0]

    except IndexError:

        # no such user

        raise AuthenticationError()

    # don't check upassword !

    return self.extid2eid(user['dn'], 'CWUser', session)

def setUpModule(*args):

    create_slapd_configuration(LDAPUserSourceTC.config)

def tearDownModule(*args):

    terminate_slapd()

def create_slapd_configuration(config):

    global slapd_process, CONFIG

    basedir = join(config.apphome, "ldapdb")

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

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

    confstream = file(slapdconf, 'w')

    confstream.write(confin % {'apphome': config.apphome})

    confstream.close()

    if not exists(basedir):

        os.makedirs(basedir)

        # fill ldap server with some data

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

        print "Initing ldap database"

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

        subprocess.call(cmdline, shell=True)

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

    port = get_available_port(xrange(9000, 9100))

    host = 'localhost:%s' % port

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

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

    print "Starting slapd on", ldapuri

    slapd_process = subprocess.Popen(cmdline)

    time.sleep(0.2)

    if slapd_process.poll() is None:

        print "slapd started with pid %s" % slapd_process.pid

    else:

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

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

    CONFIG = CONFIG % host

def terminate_slapd():

    global slapd_process

    if slapd_process.returncode is None:

        print "terminating slapd"

        if hasattr(slapd_process, 'terminate'):

            slapd_process.terminate()

        else:

            import os, signal

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

        slapd_process.wait()

        print "DONE"

    del slapd_process

class LDAPUserSourceTC(CubicWebTC):

    test_db_id = 'ldap-user'

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

    @classmethod

    def pre_setup_database(cls, session, config):

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

                                    config=CONFIG)

        session.commit()

        # XXX keep it there

        session.execute('CWUser U')

    def patch_authenticate(self):

        self._orig_authenticate = LDAPUserSource.authenticate

        LDAPUserSource.authenticate = nopwd_authenticate

    def tearDown(self):

        if hasattr(self, '_orig_authenticate'):

            LDAPUserSource.authenticate = self._orig_authenticate

        CubicWebTC.tearDown(self)

    def test_authenticate(self):

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

        self.session.set_pool()

        self.assertRaises(AuthenticationError,

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

    def test_synchronize(self):

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

        source.synchronize()

    def test_base(self):

        # check a known one

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

        e = rset.get_entity(0, 0)

        self.assertEqual(e.login, SYT)

        e.complete()

        self.assertEqual(e.creation_date, None)

        self.assertEqual(e.modification_date, None)

        self.assertEqual(e.firstname, None)

        self.assertEqual(e.surname, None)

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

        self.assertEqual(e.owned_by[0].login, SYT)

        self.assertEqual(e.created_by, ())

        self.assertEqual(e.primary_email[0].address, SYT_EMAIL)

        # email content should be indexed on the user

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

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

    def test_not(self):

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

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

        self.assert_(rset)

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

    def test_multiple(self):

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

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

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

                            {'syt': SYT, 'adim': ADIM})

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

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

                            {'syt': SYT, 'adim': ADIM})

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

    def test_in(self):

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

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

        rset = self.sexecute('Any X,L ORDERBY L WHERE X login IN("%s", "%s"), X login L' % (SYT, ADIM))

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

    def test_relations(self):

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

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

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

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

        self.assert_(SYT in (r[1] for r in rset))

    def test_count(self):

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

        # just check this is a possible number

        self.assert_(nbusers > 1, nbusers)

        self.assert_(nbusers < 30, nbusers)

    def test_upper(self):

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

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

        self.assertEqual(rset[0][0], SYT.upper())

    def test_unknown_attr(self):

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

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

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

        self.assertEqual(rset[0][0], SYT)

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

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

    def test_sort(self):

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

        self.assertEqual(logins, sorted(logins))

    def test_lower_sort(self):

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

        self.assertEqual(logins, sorted(logins))

    def test_or(self):

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

                            {'login': SYT})

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

    def test_nonregr_set_owned_by(self):

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

        # the related TrInfo has correct owner information

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

        self.commit()

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

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

        self.patch_authenticate()

        cnx = self.login(SYT, password='dummypassword')

        cu = cnx.cursor()

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

        iworkflowable = adim.cw_adapt_to('IWorkflowable')

        iworkflowable.fire_transition('deactivate')

        try:

            cnx.commit()

            adim.clear_all_caches()

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

            trinfo = iworkflowable.latest_trinfo()

            self.assertEqual(trinfo.owned_by[0].login, SYT)

            # select from_state to skip the user's creation TrInfo

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

                                'WF creation_date D, WF from_state FS,'

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

                                {'x': adim.eid})

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

        finally:

            # restore db state

            self.restore_connection()

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

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

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

    def test_same_column_names(self):

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

    def test_multiple_entities_from_different_sources(self):

        self.create_user('cochon')

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

    def test_exists1(self):

        self.session.set_pool()

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

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

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

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

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

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

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

    def test_exists2(self):

        self.create_user('comme')

        self.create_user('cochon')

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

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

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

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

    def test_exists3(self):

        self.create_user('comme')

        self.create_user('cochon')

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

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

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

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

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

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

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

    def test_exists4(self):

        self.create_user('comme')

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

        self.create_user('billy')

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

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

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

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

        # search for group name, login where

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

        # OR

        #   CWUser in_state activated AND not copain with billy

        #

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

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

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

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

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

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

        all.rows.remove(['users', SYT])

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

    def test_exists5(self):

        self.create_user('comme')

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

        self.create_user('billy')

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

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

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

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

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

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

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

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

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

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

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

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

                                              ['users', 'cochon'],

                                              ['users', SYT]])

    def test_cd_restriction(self):

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

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

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

                          ['admin', 'anon'])

    def test_union(self):

        afeids = self.sexecute('State X')

        ueids = self.sexecute('CWUser X')

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

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

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

    def _init_security_test(self):