# Copyright (C) 2010 Leonard Thomas # # This file is part of Dodai. # # Dodai is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # Dodai is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with Dodai. If not, see . import sys import os import unittest import ConfigParser from dingus import Dingus path = os.path.realpath(os.path.join(os.path.dirname(__file__), '..', '..')) sys.path.append(path) from dodai.config.databases import ConfigDatabases from dodai.config.databases import DatabaseConnectionValidator from dodai.exception import DatabaseEmptyOptionException from dodai.exception import DatabasePortException from dodai.exception import DatabaseHostnameException from dodai.exception import DatabaseProtocolException from dodai.exception import DatabaseConnectionException from dodai.exception import UnknownDatabaseConnectionException from dodai.tools import list_to_english from dodai.tools.himo import String2Himo from dodai.config.sections import ConfigSections class TestConfigDatabases(unittest.TestCase): SECTIONS = {} # valid server connection SECTIONS['db_1'] = {} SECTIONS['db_1']['protocol'] = 'postgresql' SECTIONS['db_1']['hostname'] = '127.0.0.1' SECTIONS['db_1']['username'] = 'test' SECTIONS['db_1']['password'] = 'test' SECTIONS['db_1']['port'] = '12345' SECTIONS['db_1']['database'] = 'testing' SECTIONS['db_1']['schema'] = 'foo' # valid file connection SECTIONS['db_2'] = {} SECTIONS['db_2']['protocol'] = 'sqlite' SECTIONS['db_2']['filename'] = '/tmp/test' # missing protocol server SECTIONS['db_3'] = {} SECTIONS['db_3']['hostname'] = '127.0.0.1' SECTIONS['db_3']['username'] = 'test' SECTIONS['db_3']['password'] = 'test' SECTIONS['db_3']['port'] = '12345' SECTIONS['db_3']['database'] = 'testing' # missing protocol file SECTIONS['db_4'] = {} SECTIONS['db_4']['filename'] = '/tmp/test' # empty protocol server SECTIONS['db_5'] = {} SECTIONS['db_5']['protocol'] = '' SECTIONS['db_5']['hostname'] = '127.0.0.1' SECTIONS['db_5']['username'] = 'test' SECTIONS['db_5']['password'] = 'test' SECTIONS['db_5']['port'] = '12345' SECTIONS['db_5']['database'] = 'testing' # empty protocol file SECTIONS['db_6'] = {} SECTIONS['db_6']['protocol'] = None SECTIONS['db_6']['filename'] ='/tmp/test' # Missing port SECTIONS['db_7'] = {} SECTIONS['db_7']['protocol'] = 'postgresql' SECTIONS['db_7']['protocol_extra'] = 'psycopg2' SECTIONS['db_7']['hostname'] = '127.0.0.1' SECTIONS['db_7']['username'] = 'test' SECTIONS['db_7']['password'] = 'test' SECTIONS['db_7']['database'] = 'testing' # Empty port SECTIONS['db_8'] = {} SECTIONS['db_8']['protocol'] ='postgresql' SECTIONS['db_8']['hostname'] ='127.0.0.1' SECTIONS['db_8']['username'] ='test' SECTIONS['db_8']['password'] ='test' SECTIONS['db_8']['port'] ='' SECTIONS['db_8']['database'] ='testing' # Invalid port because it's a string SECTIONS['db_9'] = {} SECTIONS['db_9']['protocol'] ='postgresql' SECTIONS['db_9']['hostname'] ='127.0.0.1' SECTIONS['db_9']['username'] ='test' SECTIONS['db_9']['password'] ='test' SECTIONS['db_9']['port'] ='blah' SECTIONS['db_9']['database'] ='testing' # Invalid port low SECTIONS['db_10'] = {} SECTIONS['db_10']['protocol'] ='postgresql' SECTIONS['db_10']['hostname'] ='127.0.0.1' SECTIONS['db_10']['username'] ='test' SECTIONS['db_10']['password'] ='test' SECTIONS['db_10']['port'] ='0' SECTIONS['db_10']['database'] ='testing' # Invalid port high SECTIONS['db_11'] = {} SECTIONS['db_11']['protocol'] ='postgresql' SECTIONS['db_11']['hostname'] ='127.0.0.1' SECTIONS['db_11']['username'] ='test' SECTIONS['db_11']['password'] ='test' SECTIONS['db_11']['port'] ='655555' SECTIONS['db_11']['database'] ='testing' # missing hostname SECTIONS['db_12'] = {} SECTIONS['db_12']['protocol'] ='postgresql' SECTIONS['db_12']['username'] ='test' SECTIONS['db_12']['password'] ='test' SECTIONS['db_12']['port'] ='655' SECTIONS['db_12']['database'] ='testing' # empty hostname SECTIONS['db_13'] = {} SECTIONS['db_13']['protocol'] ='postgresql' SECTIONS['db_13']['hostname'] ='' SECTIONS['db_13']['username'] ='test' SECTIONS['db_13']['password'] ='test' SECTIONS['db_13']['port'] ='655' SECTIONS['db_13']['database'] ='testing' # missing username SECTIONS['db_14'] = {} SECTIONS['db_14']['protocol'] = 'postgresql' SECTIONS['db_14']['hostname'] = '127.0.0.1' SECTIONS['db_14']['password'] = 'test' SECTIONS['db_14']['port'] = '12345' SECTIONS['db_14']['database'] = 'testing' SECTIONS['db_14']['schema'] = 'foo' # empty username SECTIONS['db_15'] = {} SECTIONS['db_15']['protocol'] = 'postgresql' SECTIONS['db_15']['hostname'] = '127.0.0.1' SECTIONS['db_15']['username'] = '' SECTIONS['db_15']['password'] = 'test' SECTIONS['db_15']['port'] = '12345' SECTIONS['db_15']['database'] = 'testing' # missing password SECTIONS['db_16'] = {} SECTIONS['db_16']['protocol'] = 'postgresql' SECTIONS['db_16']['hostname'] = '127.0.0.1' SECTIONS['db_16']['username'] = 'test' SECTIONS['db_16']['port'] = '12345' SECTIONS['db_16']['database'] = 'testing' # empty password SECTIONS['db_17'] = {} SECTIONS['db_17']['protocol'] = 'postgresql' SECTIONS['db_17']['hostname'] = '127.0.0.1' SECTIONS['db_17']['username'] = 'test' SECTIONS['db_17']['password'] = '' SECTIONS['db_17']['port'] = '12345' SECTIONS['db_17']['database'] = 'testing' # missing database SECTIONS['db_18'] = {} SECTIONS['db_18']['protocol'] = 'postgresql' SECTIONS['db_18']['hostname'] = '127.0.0.1' SECTIONS['db_18']['username'] = 'test' SECTIONS['db_18']['password'] = 'test' SECTIONS['db_18']['port'] = '12345' # empty database SECTIONS['db_19'] = {} SECTIONS['db_19']['protocol'] = 'postgresql' SECTIONS['db_19']['hostname'] = '127.0.0.1' SECTIONS['db_19']['username'] = 'test' SECTIONS['db_19']['password'] = 'test' SECTIONS['db_19']['port'] = '12345' SECTIONS['db_19']['database'] = '' # missing filename SECTIONS['db_20'] = {} SECTIONS['db_20']['protocol'] = 'sqlite' # empty filename SECTIONS['db_21'] = {} SECTIONS['db_21']['protocol'] = 'sqlite' SECTIONS['db_21']['filename'] = '' # not a database SECTIONS['db_22'] = {} SECTIONS['db_22']['foo'] = 'bar' # valid server test handler SECTIONS['db_23'] = {} SECTIONS['db_23']['handler'] = 'test_handler' SECTIONS['db_23']['protocol'] = 'postgresql' SECTIONS['db_23']['hostname'] = '127.0.0.1' SECTIONS['db_23']['username'] = 'test' SECTIONS['db_23']['password'] = 'test' SECTIONS['db_23']['port'] = '12345' SECTIONS['db_23']['database'] = 'testing' # invalid protocol SECTIONS['db_24'] = {} SECTIONS['db_24']['protocol'] = 'foo' SECTIONS['db_24']['hostname'] = '127.0.0.1' SECTIONS['db_24']['username'] = 'test' SECTIONS['db_24']['password'] = 'test' SECTIONS['db_24']['port'] = '12345' SECTIONS['db_24']['database'] = 'testing' def setUp(self): sa = Dingus('sa') sa.name = 'test' self.db = ConfigDatabases(sa, 'sa') self.db.add(self.SECTIONS) def test_default_handler(self): self.assertTrue('sa' in self.db._handlers) val = self.db._handlers['sa'] self.assertTrue('test' == val.name) def test_sections(self): self.assertTrue(len(self.db.connections) == len(self.SECTIONS)) def test_load_server(self): obj = self.db.load('db_1') def test_load_file(self): obj = self.db.load('db_2') def test_unable_to_load(self): self.failUnlessRaises(UnknownDatabaseConnectionException, self.db.load, 'db_twenty_seven') def test_missing_protocol_server(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_3') def test_missing_protocol_file(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_4') def test_empty_protocol_server(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_5') def test_empty_protocol_file(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_6') def test_missing_port(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_7') def test_empty_port(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_8') def test_invalid_port_because_its_a_string(self): self.failUnlessRaises(DatabasePortException, self.db.load, 'db_9') def test_invalid_port_because_its_to_low(self): self.failUnlessRaises(DatabasePortException, self.db.load, 'db_10') def test_invalid_port_because_its_to_high(self): self.failUnlessRaises(DatabasePortException, self.db.load, 'db_11') def test_missing_hostname(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_12') def test_empty_hostname(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_13') def test_missing_username(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_14') def test_empty_username(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_15') def test_missing_password(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_16') def test_empty_password(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_17') def test_missing_database(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_18') def test_empty_database(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_19') def test_missing_filename(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_20') def test_empty_filename(self): self.failUnlessRaises(DatabaseEmptyOptionException, self.db.load, 'db_21') def test_not_a_connection(self): self.failUnlessRaises(DatabaseConnectionException, self.db.load, 'db_22') def test_non_default_handler(self): test_handler = Dingus('test_handler') self.db.add_handler('test_handler', test_handler) self.db.load('db_23') def test_invalid_protocol(self): self.failUnlessRaises(DatabaseProtocolException, self.db.load, 'db_24')