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                          nameid='unspecified'):
  60         newconf = ConfigParser.ConfigParser()
  61         newconf.add_section('globals')
  62         for k in global_opts.keys():
  63             newconf.set('globals', k, global_opts[k])
  64         newconf.add_section('arguments')
  65         for k in args_opts.keys():
  66             newconf.set('arguments', k, args_opts[k])
  67
  68         profile = io.BytesIO()
  69         newconf.write(profile)
  70
  71         t = Template(profile.getvalue())
  72         text = t.substitute({'NAME': name, 'ADDRESS': addr, 'PORT': port,
  73                              'TESTDIR': self.testdir,
  74                              'ROOTDIR': self.rootdir,
  75                              'NAMEID': nameid,
  76                              'TEST_USER': self.testuser})
  77
  78         filename = os.path.join(self.testdir, '%s_profile.cfg' % name)
  79         with open(filename, 'wb') as f:
  80             f.write(text)
  81
  82         return filename
  83
  84     def setup_http(self, name, addr, port):
  85         httpdir = os.path.join(self.testdir, name)
  86         os.mkdir(httpdir)
  87         os.mkdir(os.path.join(httpdir, 'conf.d'))
  88         os.mkdir(os.path.join(httpdir, 'html'))
  89         os.mkdir(os.path.join(httpdir, 'logs'))
  90         os.symlink('/etc/httpd/modules', os.path.join(httpdir, 'modules'))
  91
  92         with open(os.path.join(self.rootdir, 'tests/httpd.conf')) as f:
  93             t = Template(f.read())
  94             text = t.substitute({'HTTPROOT': httpdir,
  95                                  'HTTPADDR': addr,
  96                                  'HTTPPORT': port})
  97         filename = os.path.join(httpdir, 'httpd.conf')
  98         with open(filename, 'w+') as f:
  99             f.write(text)
 100
 101         return filename
 102
 103     def setup_idp_server(self, profile, name, addr, port, env):
 104         http_conf_file = self.setup_http(name, addr, port)
 105         cmd = [os.path.join(self.rootdir,
 106                             'ipsilon/install/ipsilon-server-install'),
 107                '--config-profile=%s' % profile]
 108         subprocess.check_call(cmd, env=env)
 109         os.symlink(os.path.join(self.rootdir, 'ipsilon'),
 110                    os.path.join(self.testdir, 'lib', name, 'ipsilon'))
 111
 112         return http_conf_file
 113
 114     def setup_sp_server(self, profile, name, addr, port, env):
 115         http_conf_file = self.setup_http(name, addr, port)
 116         cmd = [os.path.join(self.rootdir,
 117                             'ipsilon/install/ipsilon-client-install'),
 118                '--config-profile=%s' % profile]
 119         subprocess.check_call(cmd, env=env)
 120
 121         return http_conf_file
 122
 123     def setup_pgdb(self, datadir, env):
 124         cmd = ['/usr/bin/pg_ctl', 'initdb', '-D', datadir]
 125         subprocess.check_call(cmd, env=env)
 126         auth = 'host all all 127.0.0.1/24 trust\n'
 127         filename = os.path.join(datadir, 'pg_hba.conf')
 128         with open(filename, 'a') as f:
 129             f.write(auth)
 130
 131     def start_http_server(self, conf, env):
 132         env['MALLOC_CHECK_'] = '3'
 133         env['MALLOC_PERTURB_'] = str(random.randint(0, 32767) % 255 + 1)
 134         p = subprocess.Popen(['/usr/sbin/httpd', '-DFOREGROUND', '-f', conf],
 135                              env=env, preexec_fn=os.setsid)
 136         self.processes.append(p)
 137
 138     def start_pgdb_server(self, datadir, rundir, log, addr, port, env):
 139         p = subprocess.Popen(['/usr/bin/pg_ctl', 'start', '-D', datadir, '-o',
 140                               '-c unix_socket_directories=%s -c port=%s -c \
 141                                listen_addresses=%s' % (rundir, port, addr),
 142                               '-l', log, '-w'],
 143                              env=env, preexec_fn=os.setsid)
 144         self.processes.append(p)
 145         p.wait()
 146         for d in ['adminconfig', 'userprefs', 'transactions', 'sessions']:
 147             cmd = ['/usr/bin/createdb', '-h', addr, '-p', port, d]
 148             subprocess.check_call(cmd, env=env)
 149
 150     def setup_ldap(self, env):
 151         ldapdir = os.path.join(self.testdir, 'ldap')
 152         os.mkdir(ldapdir)
 153         with open(os.path.join(self.rootdir, 'tests/slapd.conf')) as f:
 154             t = Template(f.read())
 155             text = t.substitute({'ldapdir': ldapdir})
 156         filename = os.path.join(ldapdir, 'slapd.conf')
 157         with open(filename, 'w+') as f:
 158             f.write(text)
 159         subprocess.check_call(['/usr/sbin/slapadd', '-f', filename, '-l',
 160                                'tests/ldapdata.ldif'], env=env)
 161
 162         return filename
 163
 164     def start_ldap_server(self, conf, addr, port, env):
 165         p = subprocess.Popen(['/usr/sbin/slapd', '-d', '0', '-f', conf,
 166                              '-h', 'ldap://%s:%s' % (addr, port)],
 167                              env=env, preexec_fn=os.setsid)
 168         self.processes.append(p)
 169
 170     def wait(self):
 171         for p in self.processes:
 172             os.killpg(p.pid, signal.SIGTERM)
 173
 174     def setup_servers(self, env=None):
 175         raise NotImplementedError()
 176
 177     def run(self, env):
 178         exe = self.execname
 179         if exe.endswith('c'):
 180             exe = exe[:-1]
 181         return subprocess.call([exe], env=env)