# HG changeset patch # User Sandro Knauß # Date 1331402582 -3600 # Node ID c8615310da30e7180831da3419dc963877e1d3ab # Parent 602720314930b3b56a862e7b405b813e6ef62882 adding install functions like checkDatabase, checkConfig etc. + tests diff -r 602720314930 -r c8615310da30 iro.conf.inst --- a/iro.conf.inst Sat Mar 10 19:01:31 2012 +0100 +++ b/iro.conf.inst Sat Mar 10 19:03:02 2012 +0100 @@ -1,37 +1,39 @@ -[geonet] -host = localhost -port = 25 -send_from = -user = -password = - - -[FAX_de] -host = localhost -port = 25 -send_from = -user = -password = - -[sipgate] -user = -password = - -[smtp] -host = localhost -port = 25 -send_from = -user = -password = -TLS=No -SSL=No - -[server] -key= -cert= - -[smstrade] -key= -route=basic -debug=1 -from= +[main] +# Connection URL to database +dburl = + +# Port under that twisted is running +port = + +[smstrade] +# One available provider typ. +typ = smstrade + +# smstrade Gateway Key https://login.smstrade.de/index.php?gateway +key = + +[smtp] +# One available provider typ. +typ = smtp + +# Hostname of MTA +host = + +# Port of the MTA +# port = 25 + +# username to login into MTA. +# user = + +# password to login into MTA. +# password = + +# use SSL for connection to MTA +# SSL = False + +# use TLS for connection to MTA +# TLS = False + +# Emailaddress from which mail will be sended. +send_from = + diff -r 602720314930 -r c8615310da30 iro/install.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/iro/install.py Sat Mar 10 19:03:02 2012 +0100 @@ -0,0 +1,60 @@ +from twisted.python import log +from sqlalchemy import create_engine +from sqlalchemy.exc import DatabaseError +import os + +from config import configParser, confFiles, main +from error import NeededOption +from offer.provider import providers + +from model.schema import Base +import config + +def checkConfig(): + try: + l = configParser.read(confFiles) + if len(l) > 0: + return True + return False + except NeededOption as e: + log.msg("Error while processing config file: %s"%e) + return False + +def checkDatabase(): + engine = create_engine(config.main.dburl) + for t in Base.metadata.sorted_tables: + if not t.exists(engine): + return False + return True + +def checkDatabaseConnection(): + try: + engine = create_engine(config.main.dburl) + con = engine.connect() + con.close() + return True + except DatabaseError as e: + log.msg("Error while trying to connect to database\n%s"%e) + return False + +def createDatabase(): + engine = create_engine(config.main.dburl) + Base.metadata.create_all(engine) + +def createSampleConfig(): + if not os.path.exists("iro.conf"): + with open("iro.conf",'w') as fp: + fp.write("\n".join(main.sampleConf())) + fp.write("\n") + k = providers.keys() + k.sort() + for p in k: + fp.write("\n".join(providers[p](p).sampleConf())) + fp.write("\n") + else: + log.msg("iro.conf exists and will not be overwritten.") + +def check(): + if checkConfig() and checkDatabaseConnection() and checkDatabase(): + return True + return False diff -r 602720314930 -r c8615310da30 tests/install.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/tests/install.py Sat Mar 10 19:03:02 2012 +0100 @@ -0,0 +1,110 @@ +#from mock import patch, Mock +from twisted.trial import unittest +from twisted.python import log +from sqlalchemy import create_engine +import os + +from iro import install +from iro import config +from iro.model.schema import Base + +from .dbtestcase import md, SampleDatabase + +class DummyObserver(object): + def __init__(self): + self.e=[] + + def start(self): + log.addObserver(self.emit) + + def stop(self): + log.removeObserver(self.emit) + + def emit(self, eventDict): + self.e.append(eventDict) + + +class TestInstallation(unittest.TestCase): + '''test install script''' + + def setUp(self): + md.setUp() + if not hasattr(md,"db2"): + md.db2=SampleDatabase("test2","test2",'%s/my.cnf'%md.tdir) + md.dburl2='mysql://test2:test@localhost/test2?unix_socket=%s/socket'%md.tdir + md.db2.create() + self.log = DummyObserver() + self.log.start() + self.engine = None + + def tearDown(self): + self.log.stop() + if self.engine: + Base.metadata.drop_all(self.engine) + self.engine = None + try: + os.remove("iro.conf") + except OSError as e: + if e.errno != 2: + raise + + + + def testCheckConfig(self): + self.assertEqual(install.checkConfig(),False) + with open("iro.conf",'w') as fp: + fp.write("""[main] +dburl=foo""") + self.assertEqual(install.checkConfig(),False) + self.assertEqual(len(self.log.e),1) + self.assertEqual(self.log.e[0]['message'], ("Error while processing config file: Option 'port' in section 'main' is missing.",)) + with open("iro.conf",'w') as fp: + fp.write("""[main] +dburl=foo +port=123456 +""") + self.assertEqual(install.checkConfig(),True) + + def testCheckDatabase(self): + config.main.dburl=md.dburl2 + self.assertTrue(install.checkDatabaseConnection()) + self.assertFalse(install.checkDatabase()) + self.engine = create_engine(md.dburl2) + Base.metadata.create_all(self.engine) + self.assertTrue(install.checkDatabase()) + + def testCheckDatabaseConnection(self): + config.main.dburl="mysql://t@localhost/test" + self.assertFalse(install.checkDatabaseConnection()) + self.assertEqual(len(self.log.e),1) + self.assertTrue(self.log.e[0]['message'][0].startswith("Error while trying to connect to database\n")) + config.main.dburl="sqlite://" + self.assertTrue(install.checkDatabaseConnection()) + + + def testCreateDatabase(self): + config.main.dburl=md.dburl2 + self.assertTrue(install.checkDatabaseConnection()) + self.assertFalse(install.checkDatabase()) + install.createDatabase() + self.assertTrue(install.checkDatabase()) + + def testCreateSampleConfig(self): + install.createSampleConfig() + with open("iro.conf",'r') as fp: + c = fp.read() + with open(os.path.abspath("../iro.conf.inst"),"r") as fp: + self.assertEqual(c,fp.read()) + + def testNoOverwrite(self): + with open("iro.conf","w") as fp: + fp.write("muhaha") + install.createSampleConfig() + with open("iro.conf",'r') as fp: + self.assertEqual(fp.read(),"muhaha") + self.assertEqual(len(self.log.e),1) + self.assertEqual(self.log.e[0]['message'], ("iro.conf exists and will not be overwritten.",)) + + def testCheck(self): + pass + testCheck.todo = "to implement"