""" This runs the RMS scenari as unit test for the DB. For that, it replaces the normal MySQL connection with an inmemory SQLite db """ # pylint: disable=invalid-name,wrong-import-position from __future__ import print_function from __future__ import absolute_import from __future__ import division from mock import patch from pytest import fixture from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from DIRAC import gLogger from DIRAC.RequestManagementSystem.DB import RequestDB from DIRAC.RequestManagementSystem.DB.test.RMSTestScenari import test_dirty,\ test_scheduled,\ test_stress,\ test_stressBulk @fixture(scope="function") def reqDB(request): """ This fixture instanciate a RequestDB with an in memory sqlite backend """ def mock_requestDB__init__(self): """ This mock creates the RequestDB with an in memory sqlite backend""" self.log = gLogger.getSubLogger('RequestDB') # Initialize the connection info self.engine = create_engine('sqlite:///:memory:', echo=False, pool_recycle=3600) RequestDB.metadata.bind = self.engine self.DBSession = sessionmaker(bind=self.engine) with patch.object(RequestDB.RequestDB, "__init__", mock_requestDB__init__): db = RequestDB.RequestDB() db.createTables() yield db