# copyright 2003-2012 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/>.

"""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 import AuthenticationError

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 *

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

URL = None

def create_slapd_configuration(config):

    global slapd_process, URL

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

    #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"]

    slapd_process = subprocess.Popen(cmdline)

    time.sleep(0.2)

    if slapd_process.poll() is None:

    else:

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

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

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

    global slapd_process

    if slapd_process.returncode is None:

        if hasattr(slapd_process, 'terminate'):

            slapd_process.terminate()

        else:

            import os, signal

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

        slapd_process.wait()

    del slapd_process

    test_db_id = 'ldap-feed'

    @classmethod

    def pre_setup_database(cls, session, config):

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

                              url=URL, config=CONFIG)

        session.commit()

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

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

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

    def setUp(self):

        super(LDAPFeedSourceTC, self).setUp()

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

        # just attributed

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

        lfsource.urls = [URL]

    def assertMetadata(self, entity):

        self.assertTrue(entity.creation_date)

        self.assertTrue(entity.modification_date)

    def test_authenticate(self):

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

        self.session.set_cnxset()

        # ensure we won't be logged against

        self.assertRaises(AuthenticationError,

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

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

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

    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.assertMetadata(e)

        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, 'Sylvain Thenault')

        # 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_copy_to_system_source(self):

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

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

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

        self.commit()

        source.reset_caches()

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

        self.assertEqual(len(rset), 1)

        e = rset.get_entity(0, 0)

        self.assertEqual(e.eid, eid)

                                                  'type': 'CWUser',

                                                  'extid': None})

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

        self.assertTrue(e.creation_date)

        self.assertTrue(e.modification_date)

        source.pull_data(self.session)

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

        self.assertEqual(len(rset), 1)

        # test some password has been set

        cu = self.session.system_sql('SELECT cw_upassword FROM cw_CWUser WHERE cw_eid=%s' % rset[0][0])

        value = str(cu.fetchall()[0][0])

        self.assertEqual(value, '{SSHA}v/8xJQP3uoaTBZz1T7Y0B3qOxRN1cj7D')

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

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

class LDAPUserSourceTC(LDAPFeedSourceTC):

    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',

                              url=URL, config=CONFIG)

        session.commit()

        # XXX keep it there

        session.execute('CWUser U')

    def assertMetadata(self, entity):

        self.assertEqual(entity.creation_date, None)

        self.assertEqual(entity.modification_date, None)

    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.assertMetadata(e)

        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, 'Sylvain Thenault')

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

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

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

        req = self.request()