3 # Copyright (C) 2014 Ipsilon project Contributors, for license see COPYING
5 from helpers.common import IpsilonTestBase # pylint: disable=relative-import
6 from helpers.http import HttpSessions # pylint: disable=relative-import
12 import ipsilon.util.data
14 idp_g = {'TEMPLATES': '${TESTDIR}/templates/install',
15 'CONFDIR': '${TESTDIR}/etc',
16 'DATADIR': '${TESTDIR}/lib',
17 'HTTPDCONFD': '${TESTDIR}/${NAME}/conf.d',
18 'STATICDIR': '${ROOTDIR}',
19 'BINDIR': '${ROOTDIR}/ipsilon',
20 'WSGI_SOCKET_PREFIX': '${TESTDIR}/${NAME}/logs/wsgi'}
23 idp_a = {'hostname': '${ADDRESS}:${PORT}',
24 'admin_user': '${TEST_USER}',
25 'system_user': '${TEST_USER}',
26 'instance': '${NAME}',
32 'server_debugging': 'True'}
35 class IpsilonTest(IpsilonTestBase):
38 super(IpsilonTest, self).__init__('dbupgrades', __file__)
40 def setup_servers(self, env=None):
43 def test_upgrade_from(self, env, old_version):
45 print "Installing IDP server to test upgrade from %i" % old_version
46 name = 'idp_v%i' % old_version
47 addr = '127.0.0.%i' % (10 + old_version)
48 port = str(45080 + old_version)
49 idp = self.generate_profile(idp_g, idp_a, name, addr, port)
50 conf = self.setup_idp_server(idp, name, addr, port, env)
52 # Move database of old_version into place
53 cfgfile = os.path.join(self.testdir, 'etc', name, 'ipsilon.conf')
54 db_indir = os.path.join(self.rootdir, 'tests', 'blobs', 'old_dbs',
56 db_outdir = os.path.join(self.testdir, 'lib', name)
59 for database in ['adminconfig',
64 db_in = os.path.join(db_indir, '%s.sqlite.dump' % database)
65 db_out = os.path.join(db_outdir, '%s.sqlite' % database)
67 cmd = ['/bin/sqlite3', db_out, '.read %s' % db_in]
68 subprocess.check_call(cmd)
70 # Upgrade that database
71 cmd = [os.path.join(self.rootdir,
72 'ipsilon/install/ipsilon-upgrade-database'),
74 subprocess.check_call(cmd,
75 cwd=os.path.join(self.testdir, 'lib', name),
78 # Check some version-specific changes, to see if the upgrade went OK
80 # Check all features in a newly created database
81 # Let's verify if at least one index was created
82 test_db = os.path.join(db_outdir, 'adminconfig.sqlite')
83 p = subprocess.Popen(['/bin/sqlite3', test_db, '.dump'],
84 stdout=subprocess.PIPE)
85 output, _ = p.communicate()
87 print 'Sqlite dump failed'
89 if 'CREATE INDEX' not in output:
90 raise Exception('Database upgrade did not introduce index')
91 if 'PRIMARY KEY' not in output:
92 raise Exception('Database upgrade did not introduce primary ' +
94 elif old_version == 1:
95 # In 1 -> 2, we added indexes and primary keys
96 # Let's verify if at least one index was created
97 test_db = os.path.join(db_outdir, 'adminconfig.sqlite')
98 p = subprocess.Popen(['/bin/sqlite3', test_db, '.dump'],
99 stdout=subprocess.PIPE)
100 output, _ = p.communicate()
102 print 'Sqlite dump failed'
104 if 'CREATE INDEX' not in output:
105 raise Exception('Database upgrade did not introduce index')
106 # SQLite did not support creating primary keys, so we can't test
108 # Start the httpd server
109 http_server = self.start_http_server(conf, env)
111 # Now attempt to use the upgraded database
113 if exe.endswith('c'):
116 exe.append(str(old_version))
118 exe.append('%s:%s' % (addr, port))
119 exit_code = subprocess.call(exe, env=env)
123 # Now kill the last http server
124 os.killpg(http_server.pid, signal.SIGTERM)
125 self.processes.remove(http_server)
128 for version in range(ipsilon.util.data.CURRENT_SCHEMA_VERSION):
129 self.test_upgrade_from(env, version)
132 if __name__ == '__main__':
133 from_version = sys.argv[1]
134 idpname = sys.argv[2]
137 user = pwd.getpwuid(os.getuid())[0]
139 sess = HttpSessions()
140 sess.add_server(idpname, 'http://%s' % url, user,
143 print "dbupgrades: From v%s: Authenticate to IDP ..." % from_version,
145 sess.auth_to_idp(idpname)
146 except Exception, e: # pylint: disable=broad-except
147 print >> sys.stderr, " ERROR: %s" % repr(e)