3 # Copyright (C) 2014 Simo Sorce <simo@redhat.com>
5 # see file 'COPYING' for use and warranty information
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.
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.
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/>.
28 from string import Template
32 class IpsilonTestBase(object):
34 def __init__(self, name, execname):
36 self.execname = execname
37 self.rootdir = os.getcwd()
39 self.testuser = pwd.getpwuid(os.getuid())[0]
42 def force_remove(self, op, name, info):
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'))
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])
68 profile = io.BytesIO()
69 newconf.write(profile)
71 t = Template(profile.getvalue())
72 text = t.substitute({'NAME': name, 'ADDRESS': addr, 'PORT': port,
73 'TESTDIR': self.testdir,
74 'ROOTDIR': self.rootdir,
76 'TEST_USER': self.testuser})
78 filename = os.path.join(self.testdir, '%s_profile.cfg' % name)
79 with open(filename, 'wb') as f:
84 def setup_http(self, name, addr, port):
85 httpdir = os.path.join(self.testdir, name)
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'))
92 with open(os.path.join(self.rootdir, 'tests/httpd.conf')) as f:
93 t = Template(f.read())
94 text = t.substitute({'HTTPROOT': httpdir,
97 filename = os.path.join(httpdir, 'httpd.conf')
98 with open(filename, 'w+') as f:
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'))
112 return http_conf_file
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)
121 return http_conf_file
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:
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)
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),
143 env=env, preexec_fn=os.setsid)
144 self.processes.append(p)
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)
150 def setup_ldap(self, env):
151 ldapdir = os.path.join(self.testdir, 'ldap')
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:
159 subprocess.check_call(['/usr/sbin/slapadd', '-f', filename, '-l',
160 'tests/ldapdata.ldif'], env=env)
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)
171 for p in self.processes:
172 os.killpg(p.pid, signal.SIGTERM)
174 def setup_servers(self, env=None):
175 raise NotImplementedError()
179 if exe.endswith('c'):
181 return subprocess.call([exe], env=env)