db.py 文件源码

def setUp(self):
        super(TestBase, self).setUp()
        db_url = os.environ.get('PIFPAF_URL', "sqlite://").replace(
            "mysql://", "mysql+pymysql://")

        engine = urlparse.urlparse(db_url).scheme
        # in case some drivers have additional specification, for example:
        # PyMySQL will have scheme mysql+pymysql
        engine = engine.split('+')[0]

        # NOTE(Alexei_987) Shortcut to skip expensive db setUp
        test_method = self._get_test_method()
        if (hasattr(test_method, '_run_with')
                and engine not in test_method._run_with):
            raise testcase.TestSkipped(
                'Test is not applicable for %s' % engine)

        self.CONF = service.prepare_service([], [])

        manager = self.DRIVER_MANAGERS.get(engine)
        if not manager:
            self.skipTest("missing driver manager: %s" % engine)

        self.db_manager = manager(db_url, self.CONF)

        self.useFixture(self.db_manager)

        self.conn = self.db_manager.connection
        self.conn.upgrade()

        self.useFixture(fixtures.MockPatch('panko.storage.get_connection',
                                           side_effect=self._get_connection))