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 from ipsilon.util.plugin import PluginObject
  24 import os
  25 import pwd
  26 import shutil
  27 import signal
  28 from string import Template
  29 import subprocess
  30
  31
  32 class IpsilonTestBase(PluginObject):
  33
  34     def __init__(self, name, execname):
  35         super(IpsilonTestBase, self).__init__()
  36         self.name = name
  37         self.execname = execname
  38         self.rootdir = os.getcwd()
  39         self.testdir = None
  40         self.testuser = pwd.getpwuid(os.getuid())[0]
  41         self.processes = []
  42
  43     def force_remove(self, op, name, info):
  44         os.chmod(name, 0700)
  45         os.remove(name)
  46
  47     def setup_base(self, path, test):
  48         self.testdir = os.path.join(path, test.name)
  49         if os.path.exists(self.testdir):
  50             shutil.rmtree(self.testdir, onerror=self.force_remove)
  51         os.makedirs(self.testdir)
  52         shutil.copytree(os.path.join(self.rootdir, 'templates'),
  53                         os.path.join(self.testdir, 'templates'))
  54         os.mkdir(os.path.join(self.testdir, 'etc'))
  55         os.mkdir(os.path.join(self.testdir, 'lib'))
  56         os.mkdir(os.path.join(self.testdir, 'lib', test.name))
  57         os.mkdir(os.path.join(self.testdir, 'log'))
  58
  59     def generate_profile(self, global_opts, args_opts, name, addr, port):
  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                              'TEST_USER': self.testuser})
  76
  77         filename = os.path.join(self.testdir, '%s_profile.cfg' % name)
  78         with open(filename, 'wb') as f:
  79             f.write(text)
  80
  81         return filename
  82
  83     def setup_http(self, name, addr, port):
  84         httpdir = os.path.join(self.testdir, name)
  85         os.mkdir(httpdir)
  86         os.mkdir(os.path.join(httpdir, 'conf.d'))
  87         os.mkdir(os.path.join(httpdir, 'html'))
  88         os.mkdir(os.path.join(httpdir, 'logs'))
  89         os.symlink('/etc/httpd/modules', os.path.join(httpdir, 'modules'))
  90
  91         with open(os.path.join(self.rootdir, 'tests/httpd.conf')) as f:
  92             t = Template(f.read())
  93             text = t.substitute({'HTTPROOT': httpdir,
  94                                  'HTTPADDR': addr,
  95                                  'HTTPPORT': port})
  96         filename = os.path.join(httpdir, 'httpd.conf')
  97         with open(filename, 'w+') as f:
  98             f.write(text)
  99
 100         return filename
 101
 102     def setup_idp_server(self, profile, name, addr, port, env):
 103         http_conf_file = self.setup_http(name, addr, port)
 104         cmd = [os.path.join(self.rootdir,
 105                             'ipsilon/install/ipsilon-server-install'),
 106                '--config-profile=%s' % profile]
 107         subprocess.check_call(cmd, env=env)
 108         os.symlink(os.path.join(self.rootdir, 'ipsilon'),
 109                    os.path.join(self.testdir, 'lib', name, 'ipsilon'))
 110
 111         return http_conf_file
 112
 113     def setup_sp_server(self, profile, name, addr, port, env):
 114         http_conf_file = self.setup_http(name, addr, port)
 115         cmd = [os.path.join(self.rootdir,
 116                             'ipsilon/install/ipsilon-client-install'),
 117                '--config-profile=%s' % profile]
 118         subprocess.check_call(cmd, env=env)
 119
 120         return http_conf_file
 121
 122     def start_http_server(self, conf, env):
 123         p = subprocess.Popen(['/usr/sbin/httpd', '-DFOREGROUND', '-f', conf],
 124                              env=env, preexec_fn=os.setsid)
 125         self.processes.append(p)
 126
 127     def wait(self):
 128         for p in self.processes:
 129             os.killpg(p.pid, signal.SIGTERM)
 130
 131     def setup_servers(self, env=None):
 132         raise NotImplementedError()
 133
 134     def run(self, env):
 135         exe = self.execname
 136         if exe.endswith('c'):
 137             exe = exe[:-1]
 138         return subprocess.call([exe], env=env)