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 from string import Template
  28 import subprocess
  29
  30
  31 class IpsilonTestBase(object):
  32
  33     def __init__(self, name, execname):
  34         self.name = name
  35         self.execname = execname
  36         self.rootdir = os.getcwd()
  37         self.testdir = None
  38         self.testuser = pwd.getpwuid(os.getuid())[0]
  39         self.processes = []
  40
  41     def force_remove(self, op, name, info):
  42         os.chmod(name, 0700)
  43         os.remove(name)
  44
  45     def setup_base(self, path, test):
  46         self.testdir = os.path.join(path, test.name)
  47         if os.path.exists(self.testdir):
  48             shutil.rmtree(self.testdir, onerror=self.force_remove)
  49         os.makedirs(self.testdir)
  50         shutil.copytree(os.path.join(self.rootdir, 'templates'),
  51                         os.path.join(self.testdir, 'templates'))
  52         os.mkdir(os.path.join(self.testdir, 'etc'))
  53         os.mkdir(os.path.join(self.testdir, 'lib'))
  54         os.mkdir(os.path.join(self.testdir, 'lib', test.name))
  55         os.mkdir(os.path.join(self.testdir, 'log'))
  56
  57     def generate_profile(self, global_opts, args_opts, name, addr, port):
  58         newconf = ConfigParser.ConfigParser()
  59         newconf.add_section('globals')
  60         for k in global_opts.keys():
  61             newconf.set('globals', k, global_opts[k])
  62         newconf.add_section('arguments')
  63         for k in args_opts.keys():
  64             newconf.set('arguments', k, args_opts[k])
  65
  66         profile = io.BytesIO()
  67         newconf.write(profile)
  68
  69         t = Template(profile.getvalue())
  70         text = t.substitute({'NAME': name, 'ADDRESS': addr, 'PORT': port,
  71                              'TESTDIR': self.testdir,
  72                              'ROOTDIR': self.rootdir,
  73                              'TEST_USER': self.testuser})
  74
  75         filename = os.path.join(self.testdir, '%s_profile.cfg' % name)
  76         with open(filename, 'wb') as f:
  77             f.write(text)
  78
  79         return filename
  80
  81     def setup_http(self, name, addr, port):
  82         httpdir = os.path.join(self.testdir, name)
  83         os.mkdir(httpdir)
  84         os.mkdir(os.path.join(httpdir, 'conf.d'))
  85         os.mkdir(os.path.join(httpdir, 'html'))
  86         os.mkdir(os.path.join(httpdir, 'logs'))
  87         os.symlink('/etc/httpd/modules', os.path.join(httpdir, 'modules'))
  88
  89         with open(os.path.join(self.rootdir, 'tests/httpd.conf')) as f:
  90             t = Template(f.read())
  91             text = t.substitute({'HTTPROOT': httpdir,
  92                                  'HTTPADDR': addr,
  93                                  'HTTPPORT': port})
  94         filename = os.path.join(httpdir, 'httpd.conf')
  95         with open(filename, 'w+') as f:
  96             f.write(text)
  97
  98         return filename
  99
 100     def setup_idp_server(self, profile, name, addr, port, env):
 101         http_conf_file = self.setup_http(name, addr, port)
 102         cmd = [os.path.join(self.rootdir,
 103                             'ipsilon/install/ipsilon-server-install'),
 104                '--config-profile=%s' % profile]
 105         subprocess.check_call(cmd, env=env)
 106         os.symlink(os.path.join(self.rootdir, 'ipsilon'),
 107                    os.path.join(self.testdir, 'lib', name, 'ipsilon'))
 108
 109         return http_conf_file
 110
 111     def setup_sp_server(self, profile, name, addr, port, env):
 112         http_conf_file = self.setup_http(name, addr, port)
 113         cmd = [os.path.join(self.rootdir,
 114                             'ipsilon/install/ipsilon-client-install'),
 115                '--config-profile=%s' % profile]
 116         subprocess.check_call(cmd, env=env)
 117
 118         return http_conf_file
 119
 120     def setup_pgdb(self, datadir, env):
 121         cmd = ['/usr/bin/pg_ctl', 'initdb', '-D', datadir]
 122         subprocess.check_call(cmd, env=env)
 123         auth = 'host all all 127.0.0.1/24 trust\n'
 124         filename = os.path.join(datadir, 'pg_hba.conf')
 125         with open(filename, 'a') as f:
 126             f.write(auth)
 127
 128     def start_http_server(self, conf, env):
 129         p = subprocess.Popen(['/usr/sbin/httpd', '-DFOREGROUND', '-f', conf],
 130                              env=env, preexec_fn=os.setsid)
 131         self.processes.append(p)
 132
 133     def start_pgdb_server(self, datadir, rundir, log, addr, port, env):
 134         p = subprocess.Popen(['/usr/bin/pg_ctl', 'start', '-D', datadir, '-o',
 135                               '-c unix_socket_directories=%s -c port=%s -c \
 136                                listen_addresses=%s' % (rundir, port, addr),
 137                               '-l', log, '-w'],
 138                              env=env, preexec_fn=os.setsid)
 139         self.processes.append(p)
 140         p.wait()
 141         for d in ['adminconfig', 'userprefs', 'transactions', 'sessions']:
 142             cmd = ['/usr/bin/createdb', '-h', addr, '-p', port, d]
 143             subprocess.check_call(cmd, env=env)
 144
 145     def wait(self):
 146         for p in self.processes:
 147             os.killpg(p.pid, signal.SIGTERM)
 148
 149     def setup_servers(self, env=None):
 150         raise NotImplementedError()
 151
 152     def run(self, env):
 153         exe = self.execname
 154         if exe.endswith('c'):
 155             exe = exe[:-1]
 156         return subprocess.call([exe], env=env)