1 #from mock import patch, Mock

     2 from twisted.trial import unittest

     3 from twisted.python import log

     4 from sqlalchemy import create_engine

     5 import os

     6 

     7 from iro import install

     8 from iro import config

     9 from iro.model.schema import Base

    10 

    11 from .dbtestcase import md, SampleDatabase

    12 

    13 class DummyObserver(object):

    14     def __init__(self):

    15         self.e=[]    

    16 

    17     def start(self):

    18         log.addObserver(self.emit)

    19 

    20     def stop(self):

    21         log.removeObserver(self.emit)

    22     

    23     def emit(self, eventDict):

    24         self.e.append(eventDict)

    25 

    26 

    27 class TestInstallation(unittest.TestCase):

    28     '''test install script'''

    29    

    30     def setUp(self):

    31         md.setUp()

    32         if not hasattr(md,"db2"):

    33             md.db2=SampleDatabase("test2","test2",'%s/my.cnf'%md.tdir)

    34             md.dburl2='mysql://test2:test@localhost/test2?unix_socket=%s/socket'%md.tdir

    35             md.db2.create()

    36         self.log = DummyObserver()

    37         self.log.start()

    38         self.engine = None

    39 

    40     def tearDown(self):

    41         self.log.stop()

    42         if self.engine:

    43             Base.metadata.drop_all(self.engine)

    44             self.engine = None

    45         try:

    46             os.remove("iro.conf")

    47         except OSError as e:

    48             if e.errno != 2:

    49                 raise

    50 

    51         

    52 

    53     def testCheckConfig(self):

    54         self.assertEqual(install.checkConfig(),False)

    55         with open("iro.conf",'w') as fp:

    56             fp.write("""[main]

    57 dburl=foo""")

    58         self.assertEqual(install.checkConfig(),False)

    59         self.assertEqual(len(self.log.e),1)

    60         self.assertEqual(self.log.e[0]['message'], ("Error while processing config file: Option 'port' in section 'main' is missing.",))

    61         with open("iro.conf",'w') as fp:

    62             fp.write("""[main]

    63 dburl=foo

    64 port=123456

    65 """)

    66         self.assertEqual(install.checkConfig(),True)

    67 

    68     def testCheckDatabase(self):

    69         config.main.dburl=md.dburl2

    70         self.assertTrue(install.checkDatabaseConnection())

    71         self.assertFalse(install.checkDatabase())

    72         self.engine = create_engine(md.dburl2)

    73         Base.metadata.create_all(self.engine)

    74         self.assertTrue(install.checkDatabase())

    75 

    76     def testCheckDatabaseConnection(self):

    77         config.main.dburl="mysql://t@localhost/test"

    78         self.assertFalse(install.checkDatabaseConnection())

    79         self.assertEqual(len(self.log.e),1)

    80         self.assertTrue(self.log.e[0]['message'][0].startswith("Error while trying to connect to database\n"))

    81         config.main.dburl="sqlite://"

    82         self.assertTrue(install.checkDatabaseConnection())

    83 

    84 

    85     def testCreateDatabase(self):

    86         config.main.dburl=md.dburl2

    87         self.assertTrue(install.checkDatabaseConnection())

    88         self.assertFalse(install.checkDatabase())

    89         install.createDatabase()

    90         self.assertTrue(install.checkDatabase())

    91    

    92     def testCreateSampleConfig(self):

    93         install.createSampleConfig()

    94         with open("iro.conf",'r') as fp:

    95             c = fp.read()

    96         with open(os.path.abspath("../iro.conf.inst"),"r") as fp:

    97             self.assertEqual(c,fp.read())

    98 

    99     def testNoOverwrite(self):

   100         with open("iro.conf","w") as fp:

   101             fp.write("muhaha")

   102         install.createSampleConfig()

   103         with open("iro.conf",'r') as fp:

   104             self.assertEqual(fp.read(),"muhaha")

   105         self.assertEqual(len(self.log.e),1)

   106         self.assertEqual(self.log.e[0]['message'], ("iro.conf exists and will not be overwritten.",))

   107 

   108     def testCheck(self):

   109         pass

   110     testCheck.todo = "to implement"