tests/test_mymanager.py
changeset 75 39f777341db4
parent 49 08e4dfe1b0cb
child 104 d8ff52a0390f
equal deleted inserted replaced
74:d4306261ddfb 75:39f777341db4
     2 
     2 
     3 from tests.config import Config
     3 from tests.config import Config
     4 from mytoolkit import mymanager
     4 from mytoolkit import mymanager
     5 
     5 
     6 import unittest
     6 import unittest
       
     7 import logging
     7 
     8 
     8 
     9 
     9 class TestMyManager(unittest.TestCase):
    10 class TestMyManager(unittest.TestCase):
    10 
    11 
    11     def setUp(self):
    12     def setUp(self):
    12         cfg = Config('pgtoolkit.conf')
    13         cfg = Config('pgtoolkit.conf')
    13         test_db_conn_params = cfg['databases']['test_mysql']
    14         test_db_conn_params = cfg['databases']['test_mysql']
    14         params = self.params_to_mapping(test_db_conn_params)
    15         params = self.params_to_mapping(test_db_conn_params)
    15         self.m = mymanager.get_instance()
    16         self.m = mymanager.get_instance()
    16         self.m.create_conn(**params)
    17         self.m.create_conn(**params)
       
    18         #self.setup_logging()
    17 
    19 
    18     def tearDown(self):
    20     def tearDown(self):
    19         self.m.destroy_conn()
    21         self.m.destroy_conn()
       
    22 
       
    23     def setup_logging(self):
       
    24         #FIXME: write test handler, check SQL log
       
    25         log = logging.getLogger('mymanager_sql')
       
    26         handler = logging.StreamHandler()
       
    27         log.addHandler(handler)
       
    28         log.setLevel(logging.DEBUG)
    20 
    29 
    21     def params_to_mapping(self, params):
    30     def params_to_mapping(self, params):
    22         return dict([param.split('=') for param in params.split(' ')])
    31         return dict([param.split('=') for param in params.split(' ')])
    23 
    32 
    24     def test_mysql_query(self):
    33     def test_mysql_query(self):
    26             ajaj = 1
    35             ajaj = 1
    27             curs.execute('SELECT %(ajaj)s AS ajaj', locals())
    36             curs.execute('SELECT %(ajaj)s AS ajaj', locals())
    28             row = curs.fetchone_dict()
    37             row = curs.fetchone_dict()
    29             self.assertEqual(row.ajaj, ajaj)
    38             self.assertEqual(row.ajaj, ajaj)
    30 
    39 
       
    40     def test_mysql_error(self):
       
    41         with self.m.cursor() as curs:
       
    42             self.assertRaises(mymanager.OperationalError,
       
    43                 curs.execute, 'SELECT give_me_some_error;')
       
    44 
    31 
    45 
    32 if __name__ == '__main__':
    46 if __name__ == '__main__':
    33     unittest.main()
    47     unittest.main()
    34 
    48