source: indico/indico/tests/core.py @ 58c67a

burotelhello-world-walkthroughipv6v0.98-seriesv0.98.2v0.98.3v0.98b1v0.98b2v0.99v1.0v1.1
Last change on this file since 58c67a was 58c67a, checked in by Pedro Ferreira <jose.pedro.ferreira@…>, 3 years ago

[IMP] TF: Pylinting and interface elements

  • Supressed some unnecessary warnings, loosened some regexp;
  • Pylinted all the test framework code;
  • Made some interface elements look nicer;
  • Pylinted the fossilize module;
  • Property mode set to 100644
File size: 8.8 KB
Line 
1# -*- coding: utf-8 -*-
2##
3## $Id$
4##
5## This file is part of CDS Indico.
6## Copyright (C) 2002, 2003, 2004, 2005, 2006, 2007 CERN.
7##
8## CDS Indico is free software; you can redistribute it and/or
9## modify it under the terms of the GNU General Public License as
10## published by the Free Software Foundation; either version 2 of the
11## License, or (at your option) any later version.
12##
13## CDS Indico is distributed in the hope that it will be useful, but
14## WITHOUT ANY WARRANTY; without even the implied warranty of
15## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
16## General Public License for more details.
17##
18## You should have received a copy of the GNU General Public License
19## along with CDS Indico; if not, write to the Free Software Foundation, Inc.,
20## 59 Temple Place, Suite 330, Boston, MA 02111-1307, USA.
21
22
23# System modules
24import os, sys, shutil, signal, commands, tempfile
25
26# Database
27import transaction
28from MaKaC.common.db import DBMgr
29
30# Indico
31from MaKaC.common.Configuration import Config
32
33from indico.tests.util import colored
34from indico.tests.config import TestConfig
35from indico.tests.runners import *
36
37TEST_RUNNERS = {'unit': UnitTestRunner,
38                'functional': FunctionalTestRunner,
39                'pylint': PylintTestRunner,
40                'jsunit': JSUnitTestRunner,
41                'jslint': JSLintTestRunner,
42                'grid': GridTestRunner}
43
44
45class TestManager(object):
46    """
47    Main class, the heart of the test API.
48    Launches all the tests according to the parameters that are passed
49    """
50
51    __instance = None
52
53    def __init__(self):
54        self.dbmgr = None
55        self.zeoServer = None
56        self.tempDirs = {}
57        self.dbFolder = None
58
59    def main(self, fakeDBPolicy, testsToRun, options):
60
61        returnString = "\n\n%s\n\n" % colored('** Results', 'blue', attrs = ['bold'])
62
63        #To not pollute the installation of Indico
64        self.configureTempFolders()
65
66
67        self.startManageDB(fakeDBPolicy)
68
69        if options['coverage']:
70            CoverageTestRunner.instantiate()
71
72        #specified test can either be unit or functional.
73        if options['specify']:
74            returnString += SpecificFunctionalTestRunner(options['specify']).run()
75        else:
76            for test in testsToRun:
77                if test in TEST_RUNNERS:
78                    returnString += TEST_RUNNERS[test](**options).run()
79                else:
80                    returnString += colored("[ERR] Test set '%s' does not exist. "
81                      "It has to be added in the TEST_RUNNERS variable\n", 'red') \
82                    % test
83
84
85        self.stopManageDB(fakeDBPolicy)
86
87        return returnString
88
89    def configureTempFolders(self):
90        keyNames = [#'LogDir',
91                    'ArchiveDir',
92                    'UploadedFilesTempDir']
93
94        for key in keyNames:
95            self.tempDirs[key] = tempfile.mkdtemp()
96
97        Config.getInstance().updateValues(self.tempDirs)
98
99    def deleteTempFolders(self):
100        for k in self.tempDirs:
101            shutil.rmtree(self.tempDirs[k])
102
103################## Start of DB Managing functions ##################
104    def startManageDB(self, fakeDBPolicy):
105        """
106        fakeDBPolicy == 0, the tests to run do not need any DB
107        fakeDBPolicy == 1, unit tests need a fake DB that can be run in parallel
108        with the production DB
109        fakeDBPolicy == 2, production DB is not running and functional tests
110        need fake DB which is going to be run on production port.
111        fakeDBPolicy == 3, production DB is running, we need to stop it and
112        and start a fake DB on the production port. we will restart production DB
113        """
114
115        if fakeDBPolicy == 1:
116            self.startFakeDB(TestConfig.getInstance().getFakeDBPort())
117            TestManager.createDummyUser()
118        elif fakeDBPolicy == 2:
119            self.startFakeDB(Config.getInstance().getDBConnectionParams()[1])
120            TestManager.createDummyUser()
121        elif fakeDBPolicy == 3:
122            TestManager.stopProductionDB()
123            self.startFakeDB(Config.getInstance().getDBConnectionParams()[1])
124            TestManager.createDummyUser()
125
126    def stopManageDB(self, fakeDBPolicy):
127        if fakeDBPolicy == 1 or fakeDBPolicy == 2:
128            self.stopFakeDB()
129            TestManager.restoreDBInstance()
130        elif fakeDBPolicy == 3:
131            self.stopFakeDB()
132            TestManager.startProductionDB()
133            TestManager.restoreDBInstance()
134
135    def startFakeDB(self, zeoPort):
136        self.createNewDBFile()
137        self.zeoServer = TestManager.createDBServer(
138            os.path.join(self.dbFolder, "Data.fs"),
139            zeoPort)
140
141        DBMgr.setInstance(DBMgr(hostname="localhost", port=zeoPort))
142
143    def stopFakeDB(self):
144        try:
145            os.kill(self.zeoServer, signal.SIGINT)
146        except OSError, e:
147            print ("Problem sending kill signal: " + str(e))
148
149        try:
150            os.waitpid(self.zeoServer, 0)
151            self.removeDBFile()
152        except OSError, e:
153            print ("Problem waiting for ZEO Server: " + str(e))
154
155    @staticmethod
156    def restoreDBInstance():
157        DBMgr.setInstance(None)
158
159    @staticmethod
160    def startProductionDB():
161        try:
162            commands.getstatusoutput(TestConfig.getInstance().getStartDBCmd())
163        except KeyError:
164            print "[ERR] Not found in tests.conf: command to start production DB\n"
165            sys.exit(1)
166
167    @staticmethod
168    def stopProductionDB():
169        try:
170            commands.getstatusoutput(TestConfig.getInstance().getStopDBCmd())
171        except KeyError:
172            print "[ERR] Not found in tests.conf: command to stop production DB\n"
173            sys.exit(1)
174
175    def createNewDBFile(self):
176        from ZODB import FileStorage, DB
177        savedDir = os.getcwd()
178        self.dbFolder = tempfile.mkdtemp()
179        os.chdir(self.dbFolder)
180
181        storage = FileStorage.FileStorage("Data.fs")
182        db = DB(storage)
183        connection = db.open()
184
185        transaction.commit()
186
187        connection.close()
188        db.close()
189        storage.close()
190        os.chdir(savedDir)
191
192    def removeDBFile(self):
193        shutil.rmtree(self.dbFolder)
194
195    @staticmethod
196    def createDBServer(dbFile, port):
197        """
198        Creates a fake DB server for testing
199        """
200
201        pid = os.fork()
202        if pid:
203            return pid
204        else:
205            # run a DB in a child process
206            from indico.tests.util import TestZEOServer
207            server = TestZEOServer(port, dbFile)
208            server.start()
209
210################## End of DB Managing functions ##################
211
212    @staticmethod
213    def createDummyUser():
214        from MaKaC import user
215        from MaKaC.authentication import AuthenticatorMgr
216        from MaKaC.common import HelperMaKaCInfo
217
218        DBMgr.getInstance().startRequest()
219
220        #filling info to new user
221        avatar = user.Avatar()
222        avatar.setName( "fake" )
223        avatar.setSurName( "fake" )
224        avatar.setOrganisation( "fake" )
225        avatar.setLang( "en_US" )
226        avatar.setEmail( "fake@fake.fake" )
227
228        #registering user
229        ah = user.AvatarHolder()
230        ah.add(avatar)
231
232        #setting up the login info
233        li = user.LoginInfo( "dummyuser", "dummyuser" )
234        ih = AuthenticatorMgr()
235        userid = ih.createIdentity( li, avatar, "Local" )
236        ih.add( userid )
237
238        #activate the account
239        avatar.activateAccount()
240
241        #since the DB is empty, we have to add dummy user as admin
242        minfo = HelperMaKaCInfo.getMaKaCInfoInstance()
243        al = minfo.getAdminList()
244        al.grant( avatar )
245
246        DBMgr.getInstance().endRequest()
247
248    @staticmethod
249    def deleteDummyUser():
250        from MaKaC import user
251        from MaKaC.authentication import AuthenticatorMgr
252        from MaKaC.common import HelperMaKaCInfo
253        from MaKaC.common import indexes
254        DBMgr.getInstance().startRequest()
255
256        #removing user from admin list
257        minfo = HelperMaKaCInfo.getMaKaCInfoInstance()
258        al = minfo.getAdminList()
259        ah = user.AvatarHolder()
260        avatar = ah.match({'email':'fake@fake.fake'})[0]
261        al.revoke( avatar )
262
263        #remove the login info
264        userid = avatar.getIdentityList()[0]
265        ih = AuthenticatorMgr()
266        ih.removeIdentity(userid)
267
268        #unregistering the user info
269        index = indexes.IndexesHolder().getById("email")
270        index.unindexUser(avatar)
271        index = indexes.IndexesHolder().getById("name")
272        index.unindexUser(avatar)
273        index = indexes.IndexesHolder().getById("surName")
274        index.unindexUser(avatar)
275        index = indexes.IndexesHolder().getById("organisation")
276        index.unindexUser(avatar)
277        index = indexes.IndexesHolder().getById("status")
278        index.unindexUser(avatar)
279
280        DBMgr.getInstance().endRequest()
Note: See TracBrowser for help on using the repository browser.