tests/helpers/common.py

   1 #!/usr/bin/python
   2 #
   3 # Copyright (C) 2014  Simo Sorce <simo@redhat.com>
   4 #
   5 # see file 'COPYING' for use and warranty information
   6 #
   7 # This program is free software; you can redistribute it and/or modify
   8 # it under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation, either version 3 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # This program is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19
  20
  21 import ConfigParser
  22 import io
  23 import os
  24 import pwd
  25 import shutil
  26 import signal
  27 import random
  28 from string import Template
  29 import subprocess
  30
  31
  32 class IpsilonTestBase(object):
  33
  34     def __init__(self, name, execname):
  35         self.name = name
  36         self.execname = execname
  37         self.rootdir = os.getcwd()
  38         self.testdir = None
  39         self.testuser = pwd.getpwuid(os.getuid())[0]
  40         self.processes = []
  41
  42     def force_remove(self, op, name, info):
  43         os.chmod(name, 0700)
  44         os.remove(name)
  45
  46     def setup_base(self, path, test):
  47         self.testdir = os.path.join(path, test.name)
  48         if os.path.exists(self.testdir):
  49             shutil.rmtree(self.testdir, onerror=self.force_remove)
  50         os.makedirs(self.testdir)
  51         shutil.copytree(os.path.join(self.rootdir, 'templates'),
  52                         os.path.join(self.testdir, 'templates'))
  53         os.mkdir(os.path.join(self.testdir, 'etc'))
  54         os.mkdir(os.path.join(self.testdir, 'lib'))
  55         os.mkdir(os.path.join(self.testdir, 'lib', test.name))
  56         os.mkdir(os.path.join(self.testdir, 'log'))
  57
  58     def generate_profile(self, global_opts, args_opts, name, addr, port):
  59         newconf = ConfigParser.ConfigParser()
  60         newconf.add_section('globals')
  61         for k in global_opts.keys():
  62             newconf.set('globals', k, global_opts[k])
  63         newconf.add_section('arguments')
  64         for k in args_opts.keys():
  65             newconf.set('arguments', k, args_opts[k])
  66
  67         profile = io.BytesIO()
  68         newconf.write(profile)
  69
  70         t = Template(profile.getvalue())
  71         text = t.substitute({'NAME': name, 'ADDRESS': addr, 'PORT': port,
  72                              'TESTDIR': self.testdir,
  73                              'ROOTDIR': self.rootdir,
  74                              'TEST_USER': self.testuser})
  75
  76         filename = os.path.join(self.testdir, '%s_profile.cfg' % name)
  77         with open(filename, 'wb') as f:
  78             f.write(text)
  79
  80         return filename
  81
  82     def setup_http(self, name, addr, port):
  83         httpdir = os.path.join(self.testdir, name)
  84         os.mkdir(httpdir)
  85         os.mkdir(os.path.join(httpdir, 'conf.d'))
  86         os.mkdir(os.path.join(httpdir, 'html'))
  87         os.mkdir(os.path.join(httpdir, 'logs'))
  88         os.symlink('/etc/httpd/modules', os.path.join(httpdir, 'modules'))
  89
  90         with open(os.path.join(self.rootdir, 'tests/httpd.conf')) as f:
  91             t = Template(f.read())
  92             text = t.substitute({'HTTPROOT': httpdir,
  93                                  'HTTPADDR': addr,
  94                                  'HTTPPORT': port})
  95         filename = os.path.join(httpdir, 'httpd.conf')
  96         with open(filename, 'w+') as f:
  97             f.write(text)
  98
  99         return filename
 100
 101     def setup_idp_server(self, profile, name, addr, port, env):
 102         http_conf_file = self.setup_http(name, addr, port)
 103         cmd = [os.path.join(self.rootdir,
 104                             'ipsilon/install/ipsilon-server-install'),
 105                '--config-profile=%s' % profile]
 106         subprocess.check_call(cmd, env=env)
 107         os.symlink(os.path.join(self.rootdir, 'ipsilon'),
 108                    os.path.join(self.testdir, 'lib', name, 'ipsilon'))
 109
 110         return http_conf_file
 111
 112     def setup_sp_server(self, profile, name, addr, port, env):
 113         http_conf_file = self.setup_http(name, addr, port)
 114         cmd = [os.path.join(self.rootdir,
 115                             'ipsilon/install/ipsilon-client-install'),
 116                '--config-profile=%s' % profile]
 117         subprocess.check_call(cmd, env=env)
 118
 119         return http_conf_file
 120
 121     def setup_pgdb(self, datadir, env):
 122         cmd = ['/usr/bin/pg_ctl', 'initdb', '-D', datadir]
 123         subprocess.check_call(cmd, env=env)
 124         auth = 'host all all 127.0.0.1/24 trust\n'
 125         filename = os.path.join(datadir, 'pg_hba.conf')
 126         with open(filename, 'a') as f:
 127             f.write(auth)
 128
 129     def start_http_server(self, conf, env):
 130         env['MALLOC_CHECK_'] = '3'
 131         env['MALLOC_PERTURB_'] = str(random.randint(0, 32767) % 255 + 1)
 132         p = subprocess.Popen(['/usr/sbin/httpd', '-DFOREGROUND', '-f', conf],
 133                              env=env, preexec_fn=os.setsid)
 134         self.processes.append(p)
 135
 136     def start_pgdb_server(self, datadir, rundir, log, addr, port, env):
 137         p = subprocess.Popen(['/usr/bin/pg_ctl', 'start', '-D', datadir, '-o',
 138                               '-c unix_socket_directories=%s -c port=%s -c \
 139                                listen_addresses=%s' % (rundir, port, addr),
 140                               '-l', log, '-w'],
 141                              env=env, preexec_fn=os.setsid)
 142         self.processes.append(p)
 143         p.wait()
 144         for d in ['adminconfig', 'userprefs', 'transactions', 'sessions']:
 145             cmd = ['/usr/bin/createdb', '-h', addr, '-p', port, d]
 146             subprocess.check_call(cmd, env=env)
 147
 148     def setup_ldap(self, env):
 149         ldapdir = os.path.join(self.testdir, 'ldap')
 150         os.mkdir(ldapdir)
 151         with open(os.path.join(self.rootdir, 'tests/slapd.conf')) as f:
 152             t = Template(f.read())
 153             text = t.substitute({'ldapdir': ldapdir})
 154         filename = os.path.join(ldapdir, 'slapd.conf')
 155         with open(filename, 'w+') as f:
 156             f.write(text)
 157         subprocess.check_call(['/usr/sbin/slapadd', '-f', filename, '-l',
 158                                'tests/ldapdata.ldif'], env=env)
 159
 160         return filename
 161
 162     def start_ldap_server(self, conf, addr, port, env):
 163         p = subprocess.Popen(['/usr/sbin/slapd', '-d', '0', '-f', conf,
 164                              '-h', 'ldap://%s:%s' % (addr, port)],
 165                              env=env, preexec_fn=os.setsid)
 166         self.processes.append(p)
 167
 168     def wait(self):
 169         for p in self.processes:
 170             os.killpg(p.pid, signal.SIGTERM)
 171
 172     def setup_servers(self, env=None):
 173         raise NotImplementedError()
 174
 175     def run(self, env):
 176         exe = self.execname
 177         if exe.endswith('c'):
 178             exe = exe[:-1]
 179         return subprocess.call([exe], env=env)