# -*- coding: utf-8 -*-

 """

     LDAPTestBase: LDAP testing support for py.test based unit tests

     Features

     --------

     * setup_class

       * automatic creation of a temporary LDAP server environment

       * automatic creation of a LDAP server process (slapd)

     * teardown_class

       * LDAP server process will be killed and termination will be waited for

       * temporary LDAP environment will be removed

     Usage

     -----

     Write your own test class and derive from LDAPTestBase:

     class TestLdap(LDAPTestBase):

         def testFunction(self):

             server_url = self.ldap_env.slapd.url

             lo = ldap.initialize(server_url)

             lo.simple_bind_s('', '')

     Notes

     -----

     On Ubuntu 8.04 there is apparmor imposing some restrictions on /usr/sbin/slapd,

     so you need to disable apparmor by invoking this as root:

     # /etc/init.d/apparmor stop

     Requires Python 2.4 (for subprocess module).

     @copyright: 2008 by Thomas Waldmann

     @license: GNU GPL, see COPYING for details.

 """

 SLAPD_EXECUTABLE = 'slapd'  # filename of LDAP server executable - if it is not

                             # in your PATH, you have to give full path/filename.

 import os, shutil, tempfile, time

 from StringIO import StringIO

 import signal

 try:

     import subprocess  # needs Python 2.4

 except ImportError:

     subprocess = None

 try:

     import ldap, ldif, ldap.modlist  # needs python-ldap

 except ImportError:

     ldap = None

 def check_environ():

     """ Check the system environment whether we are able to run.

         Either return some failure reason if we can't or None if everything

         looks OK.

     """

     if subprocess is None:

         return "You need at least python 2.4 to use ldap_testbase."

     if ldap is None:

         return "You need python-ldap installed to use ldap_testbase."

     slapd = False

     try:

         p = subprocess.Popen([SLAPD_EXECUTABLE, '-V'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

         pid = p.pid

         rc = p.wait()

         if pid and rc == 1:

             slapd = True  # it works

     except OSError, err:

         import errno

         if not (err.errno == errno.ENOENT or

                 (err.errno == 3 and os.name == 'nt')):

             raise

     if not slapd:

         return "Can't start %s (see SLAPD_EXECUTABLE)." % SLAPD_EXECUTABLE

     return None

 class Slapd(object):

     """ Manage a slapd process for testing purposes """

     def __init__(self,

                  config=None,  # config filename for -f

                  executable=SLAPD_EXECUTABLE,

                  debug_flags='', # None,  # for -d stats,acl,args,trace,sync,config

                  proto='ldap', ip='127.0.0.1', port=3890,  # use -h proto://ip:port

                  service_name=''  # defaults to -n executable:port, use None to not use -n

                 ):

         self.executable = executable

         self.config = config

         self.debug_flags = debug_flags

         self.proto = proto

         self.ip = ip

         self.port = port

         self.url = '%s://%s:%d' % (proto, ip, port) # can be used for ldap.initialize() call

         if service_name == '':

             self.service_name = '%s:%d' % (executable, port)

         else:

             self.service_name = service_name

     def start(self, timeout=0):

         """ start a slapd process and optionally wait up to timeout seconds until it responds """

         args = [self.executable, '-h', self.url, ]

         if self.config is not None:

             args.extend(['-f', self.config])

         if self.debug_flags is not None:

             args.extend(['-d', self.debug_flags])

         if self.service_name:

             args.extend(['-n', self.service_name])

         self.process = subprocess.Popen(args)

         started = None

         if timeout:

             lo = ldap.initialize(self.url)

             ldap.set_option(ldap.OPT_PROTOCOL_VERSION, ldap.VERSION3) # ldap v2 is outdated

             started = False

             wait_until = time.time() + timeout

             while time.time() < wait_until:

                 try:

                     lo.simple_bind_s('', '')

                     started = True

                 except ldap.SERVER_DOWN, err:

                     time.sleep(0.1)

                 else:

                     break

         return started

     def stop(self):

         """ stop this slapd process and wait until it has terminated """

         pid = self.process.pid

         os.kill(pid, signal.SIGTERM)

         os.waitpid(pid, 0)

 class LdapEnvironment(object):

     """ Manage a (temporary) environment for running a slapd in it """

     # default DB_CONFIG bdb configuration file contents

     DB_CONFIG = """\

 # STRANGE: if i use those settings, after the test slapd goes to 100% and doesn't terminate on SIGTERM

 # Set the database in memory cache size.

 #set_cachesize 0 10000000 1

 # Set log values.

 #set_lg_regionmax 262144

 #set_lg_bsize 262144

 #set_lg_max 10485760

 #set_tas_spins 0

 """

     def __init__(self,

                  basedn,

                  rootdn, rootpw,

                  instance=0,  # use different values when running multiple LdapEnvironments

                  schema_dir='/etc/ldap/schema',  # directory with schemas

                  coding='utf-8',  # coding used for config files

                  timeout=10,  # how long to wait for slapd starting [s]

                 ):

         self.basedn = basedn

         self.rootdn = rootdn

         self.rootpw = rootpw

         self.instance = instance

         self.schema_dir = schema_dir

         self.coding = coding

         self.ldap_dir = None

         self.slapd_conf = None

         self.timeout = timeout

     def create_env(self, slapd_config, db_config=DB_CONFIG):

         """ create a temporary LDAP server environment in a temp. directory,

             including writing a slapd.conf (see configure_slapd) and a

             DB_CONFIG there.

         """

         # create directories

         self.ldap_dir = tempfile.mkdtemp(prefix='LdapEnvironment-%d.' % self.instance)

         self.ldap_db_dir = os.path.join(self.ldap_dir, 'db')

         os.mkdir(self.ldap_db_dir)

         # create DB_CONFIG for bdb backend

         db_config_fname = os.path.join(self.ldap_db_dir, 'DB_CONFIG')

         f = open(db_config_fname, 'w')

         f.write(db_config)

         f.close()

         # create slapd.conf from content template in slapd_config

         slapd_config = slapd_config % {

             'ldap_dir': self.ldap_dir,

             'ldap_db_dir': self.ldap_db_dir,

             'schema_dir': self.schema_dir,

             'basedn': self.basedn,

             'rootdn': self.rootdn,

             'rootpw': self.rootpw,

         }

         if isinstance(slapd_config, unicode):

             slapd_config = slapd_config.encode(self.coding)

         self.slapd_conf = os.path.join(self.ldap_dir, "slapd.conf")

         f = open(self.slapd_conf, 'w')

         f.write(slapd_config)

         f.close()

     def start_slapd(self):

         """ start a slapd and optionally wait until it talks with us """

         self.slapd = Slapd(config=self.slapd_conf, port=3890+self.instance)

         started = self.slapd.start(timeout=self.timeout)

         return started

     def load_directory(self, ldif_content):

         """ load the directory with the ldif_content (str) """

         lo = ldap.initialize(self.slapd.url)

         ldap.set_option(ldap.OPT_PROTOCOL_VERSION, ldap.VERSION3) # ldap v2 is outdated

         lo.simple_bind_s(self.rootdn, self.rootpw)

         class LDIFLoader(ldif.LDIFParser):

             def handle(self, dn, entry):

                 lo.add_s(dn, ldap.modlist.addModlist(entry))

         loader = LDIFLoader(StringIO(ldif_content))

         loader.parse()

     def stop_slapd(self):

         """ stop a slapd """

         self.slapd.stop()

     def destroy_env(self):

         """ remove the temporary LDAP server environment """

         shutil.rmtree(self.ldap_dir)

 try:

     import py.test

     class LDAPTstBase:

         """ Test base class for py.test based tests which need a LDAP server to talk to.

             Inherit your test class from this base class to test LDAP stuff.

         """

         # You MUST define these in your derived class:

         slapd_config = None  # a string with your slapd.conf template

         ldif_content = None  # a string with your ldif contents

         basedn = None  # your base DN

         rootdn = None  # root DN

         rootpw = None  # root password

         def setup_class(self):

             """ Create LDAP server environment, start slapd """

             self.ldap_env = LdapEnvironment(self.basedn, self.rootdn, self.rootpw)

             self.ldap_env.create_env(slapd_config=self.slapd_config)

             started = self.ldap_env.start_slapd()

             if not started:

                 py.test.skip("Failed to start %s process, please see your syslog / log files"

                              " (and check if stopping apparmor helps, in case you use it)." % SLAPD_EXECUTABLE)

             self.ldap_env.load_directory(ldif_content=self.ldif_content)

         def teardown_class(self):

             """ Stop slapd, remove LDAP server environment """

             self.ldap_env.stop_slapd()

             self.ldap_env.destroy_env()

 except ImportError:

     pass  # obviously py.test not in use