Permalink
Fetching contributors…
Cannot retrieve contributors at this time
72 lines (63 sloc) 2.41 KB
import glob
import os
import sys
import unittest
import codecs
from simple_db_migrate.config import *
try:
from StringIO import StringIO
except ImportError:
from io import StringIO
from mock import patch
def create_file(file_name, content=None, encoding='utf-8'):
f = codecs.open(file_name, 'wb', encoding)
if content:
if (sys.version_info > (3, 0)):
f.write(content)
else:
f.write(content.decode(encoding))
f.close()
return file_name
def create_migration_file(file_name, sql_up='', sql_down=''):
create_file(file_name, 'SQL_UP=u"%s"\nSQL_DOWN=u"%s"' % (sql_up, sql_down))
return file_name
def delete_files(pattern):
filelist=glob.glob(pattern)
for _file in filelist:
os.remove(_file)
def create_config(host='localhost', username='root', password='', database='migration_example', migrations_dir='.', utc_timestamp=False, script_encoding='utf-8'):
config_file = '''
DATABASE_HOST = '%s'
DATABASE_USER = '%s'
DATABASE_PASSWORD = '%s'
DATABASE_NAME = '%s'
DATABASE_MIGRATIONS_DIR = '%s'
UTC_TIMESTAMP = %s
DATABASE_SCRIPT_ENCODING = '%s'
''' % (host, username, password, database, migrations_dir, utc_timestamp, script_encoding)
create_file('test_config_file.conf', config_file)
return FileConfig('test_config_file.conf')
class BaseTest(unittest.TestCase):
def setUp(self):
self.stdout_mock = patch('sys.stdout', new_callable=StringIO)
self.stdout_mock.start()
def tearDown(self):
self.stdout_mock.stop()
delete_files('*.log')
delete_files('*test_migration.migration')
delete_files('migrations/*test_migration.migration')
if os.path.exists(os.path.abspath('migrations')):
os.rmdir(os.path.abspath('migrations'))
if os.path.exists(os.path.abspath('test_config_file.conf')):
os.remove(os.path.abspath('test_config_file.conf'))
def assertRaisesWithMessage(self, excClass, excMessage, callableObj, *args, **kwargs):
raisedMessage = ''
try:
callableObj(*args, **kwargs)
except excClass as e:
raisedMessage = str(e)
if excMessage == raisedMessage:
return
if hasattr(excClass,'__name__'): excName = excClass.__name__
else: excName = str(excClass)
raise self.failureException("%s not raised with message '%s', the message was '%s'" % (excName, excMessage, raisedMessage))