diff MoinMoin/_tests/ldap_testbase.py @ 3650:23851c20e53f

add ldap testing framework, add ldap_login tests
author Thomas Waldmann <tw AT waldmann-edv DOT de>
date Sun, 01 Jun 2008 02:18:38 +0200
parents
children df024fd0a129
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/MoinMoin/_tests/ldap_testbase.py	Sun Jun 01 02:18:38 2008 +0200
@@ -0,0 +1,219 @@
+# -*- 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
+
+    @copyright: 2008 by Thomas Waldmann
+    @license: GNU GPL, see COPYING for details.
+"""
+
+import os, shutil, tempfile, time
+from StringIO import StringIO
+import signal
+import subprocess  # needs Python 2.4
+
+import ldap, ldif, ldap.modlist
+
+class Slapd(object):
+    """ Manage a slapd process for testing purposes """
+    def __init__(self,
+                 config=None,  # config filename for -f
+                 executable='slapd',  # slapd executable filename
+                 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)
+        self.slapd.start(timeout=self.timeout)
+
+    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)
+
+
+class LDAPTestBase:
+    """ 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)
+        self.ldap_env.start_slapd()
+        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()
+
+