import ConfigParser
import io
-from ipsilon.util.plugin import PluginObject
import os
import pwd
import shutil
import signal
+import random
from string import Template
import subprocess
-class IpsilonTestBase(PluginObject):
+class IpsilonTestBase(object):
def __init__(self, name, execname):
- super(IpsilonTestBase, self).__init__()
self.name = name
self.execname = execname
self.rootdir = os.getcwd()
os.mkdir(os.path.join(self.testdir, 'lib', test.name))
os.mkdir(os.path.join(self.testdir, 'log'))
- def generate_profile(self, global_opts, args_opts, name, addr, port):
+ def generate_profile(self, global_opts, args_opts, name, addr, port,
+ nameid='unspecified'):
newconf = ConfigParser.ConfigParser()
newconf.add_section('globals')
for k in global_opts.keys():
text = t.substitute({'NAME': name, 'ADDRESS': addr, 'PORT': port,
'TESTDIR': self.testdir,
'ROOTDIR': self.rootdir,
+ 'NAMEID': nameid,
'TEST_USER': self.testuser})
filename = os.path.join(self.testdir, '%s_profile.cfg' % name)
return http_conf_file
+ def setup_pgdb(self, datadir, env):
+ cmd = ['/usr/bin/pg_ctl', 'initdb', '-D', datadir]
+ subprocess.check_call(cmd, env=env)
+ auth = 'host all all 127.0.0.1/24 trust\n'
+ filename = os.path.join(datadir, 'pg_hba.conf')
+ with open(filename, 'a') as f:
+ f.write(auth)
+
def start_http_server(self, conf, env):
+ env['MALLOC_CHECK_'] = '3'
+ env['MALLOC_PERTURB_'] = str(random.randint(0, 32767) % 255 + 1)
p = subprocess.Popen(['/usr/sbin/httpd', '-DFOREGROUND', '-f', conf],
env=env, preexec_fn=os.setsid)
self.processes.append(p)
+ def start_pgdb_server(self, datadir, rundir, log, addr, port, env):
+ p = subprocess.Popen(['/usr/bin/pg_ctl', 'start', '-D', datadir, '-o',
+ '-c unix_socket_directories=%s -c port=%s -c \
+ listen_addresses=%s' % (rundir, port, addr),
+ '-l', log, '-w'],
+ env=env, preexec_fn=os.setsid)
+ self.processes.append(p)
+ p.wait()
+ for d in ['adminconfig', 'users', 'transactions', 'sessions']:
+ cmd = ['/usr/bin/createdb', '-h', addr, '-p', port, d]
+ subprocess.check_call(cmd, env=env)
+
+ def setup_ldap(self, env):
+ ldapdir = os.path.join(self.testdir, 'ldap')
+ os.mkdir(ldapdir)
+ with open(os.path.join(self.rootdir, 'tests/slapd.conf')) as f:
+ t = Template(f.read())
+ text = t.substitute({'ldapdir': ldapdir})
+ filename = os.path.join(ldapdir, 'slapd.conf')
+ with open(filename, 'w+') as f:
+ f.write(text)
+ subprocess.check_call(['/usr/sbin/slapadd', '-f', filename, '-l',
+ 'tests/ldapdata.ldif'], env=env)
+
+ return filename
+
+ def start_ldap_server(self, conf, addr, port, env):
+ p = subprocess.Popen(['/usr/sbin/slapd', '-d', '0', '-f', conf,
+ '-h', 'ldap://%s:%s' % (addr, port)],
+ env=env, preexec_fn=os.setsid)
+ self.processes.append(p)
+
def wait(self):
for p in self.processes:
os.killpg(p.pid, signal.SIGTERM)