Я пытаюсь смоделировать функцию, чтобы вызвать исключение, но она не работает.
class TestGetPos(unittest.TestCase):
@patch('db.apis._create_engine')
def test_get_all_pos(self, mock_sqlalchemy_engine):
mock_sqlalchemy_engine.return_value.execute.return_value.fetchall.return_value = [1,2,3]
assert get_all_pos(None) == [1,2,3]
mock_sqlalchemy_engine.return_value.execute('testqry').side_effect = DBAPIError
print(get_all_pos(None))
with self.assertRaises(DBAPIError):
print(get_all_pos(None))
import sqlalchemy
def _create_engine(app):
impac_engine = sqlalchemy.create_engine(
app.config['DB'],
poolclass=sqlalchemy.pool.NullPool # this setting enables NOT to use Pooling, preventing from timeout issues.
)
return impac_engine
def get_all_pos(app):
engine = _create_engine(app)
qry = """SELECT DISTINCT id, name FROM po_t ORDER BY name ASC"""
try:
cursor = engine.execute(qry)
rows = cursor.fetchall()
return rows
except Exception as re:
raise re