Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- class FixtureHelper(object):
- """Underlying helper object for a StorageFixture to hold driver methods"""
- def __init__(self, fixture):
- """
- :param fixture: The fixture object
- """
- self.fixture = fixture
- def setUp(self):
- """Runs pr test, typically a db reset or similar"""
- def pre_migrate(self):
- """Run before migrations"""
- def migrate(self):
- """Migrate the storage"""
- def post_migrate(self):
- """This is executed after migrations"""
- def post_init(self):
- """Runs at the end of the object initialization"""
- class SQLAlchemyHelper(FixtureHelper):
- def __init__(self, fixture):
- super(SQLAlchemyHelper, self).__init__(fixture)
- self.sqlite_db = fixture.kw.get('sqlite_db')
- self.sqlite_clean_db = fixture.kw.get('sqlite_clean_db')
- self.testdb = None
- def setUp(self):
- if self.fixture.database_connection == "sqlite://":
- conn = self.fixture.connection.engine.connect()
- conn.connection.executescript(self._DB)
- self.addCleanup(self.fixture.connection.engine.dispose)
- else:
- shutil.copyfile(paths.state_path_rel(self.sqlite_clean_db),
- paths.state_path_rel(self.sqlite_db))
- def pre_migrate(self):
- self.fixture.connection.engine.dispose()
- self.fixture.connection.engine.connect()
- if self.fixture.database_connection == "sqlite://":
- #https://github.com/openstack/nova/blob/master/nova/test.py#L82-L84
- pass
- else:
- testdb = paths.state_path_rel(self.sqlite_db)
- if os.path.exists(testdb):
- return
- def migrate(self):
- self.fixture.connection.setup_schema()
- def post_init(self):
- if self.fixture.database_connection == "sqlite://":
- conn = self.fixture.connection.engine.connect()
- self._as_string = "".join(
- l for l in conn.iterdump())
- self.fixture.connection.engine.dispose()
- else:
- cleandb = paths.state_path_rel(self.sqlite_clean_db)
- shutil.copyfile(self.testdb, cleandb)
- class StorageFixture(fixtures.Fixture):
- """
- Storage fixture that for now just supports SQLAlchemy
- """
- def __init__(self, svc, **kw):
- self.svc = svc
- self.kw = kw
- self.driver = kw.get('storage_driver', 'sqlalchemy')
- self.database_connection = kw.get('database_connection', 'sqlite://')
- self.svc_group = 'service:%s' % self.svc
- self.driver_group = '%s:%s' % (self.svc, self.driver)
- self.module = self.import_module()
- self.helper = SQLAlchemyHelper(self)
- set_config(storage_driver=self.driver, group=self.svc_group)
- set_config(database_connection=self.database_connection,
- group=self.driver_group)
- self.connection = self.get_storage_connection(**kw)
- self.helper.pre_migrate()
- self.helper.migrate()
- self.helper.post_migrate()
- self.helper.post_init()
- def setUp(self):
- super(StorageFixture, self).setUp()
- self.helper.setUp()
- def import_module(self):
- """
- Setup the storage module on self.
- """
- return importutils.import_module('billingstack.%s.storage' % self.svc)
- def get_storage_connection(self, **kw):
- """
- Import the storage module for the service that we are going to act on,
- then return a connection object for that storage module.
- :param service: The service.
- """
- engine = self.module.get_engine(self.driver)
- return engine.get_connection()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement