From 2d653653cf9086b186fb92f1d4fbd4a8437a5677 Mon Sep 17 00:00:00 2001
From: Boris Pavlovic <boris@pavlovic.me>
Date: Mon, 28 Jan 2013 17:16:15 +0400
Subject: [PATCH] Fix missing wrap_db_error for Session.execute() method

We should add wrap_db_error for Session.execute() method, because
Query.update() method will call Session.execute() and it is not inside
Session.flush(), so exceptions from Query.update() wouldn't be wrapped.

Add test for Session.flush() wrapper
Add test for Session.execute() wrapper

Fixes bug 1107890

Change-Id: I96894e502f9f279999b61a1e66469e34a07a022d
---
 nova/tests/test_sqlalchemy.py | 63 +++++++++++++++++++++++++++++++++++
 1 file changed, 63 insertions(+)

diff --git a/nova/tests/test_sqlalchemy.py b/nova/tests/test_sqlalchemy.py
index f79d607f..5c7f4450 100644
--- a/nova/tests/test_sqlalchemy.py
+++ b/nova/tests/test_sqlalchemy.py
@@ -22,8 +22,14 @@ try:
 except ImportError:
     MySQLdb = None
 
+from sqlalchemy import Column, MetaData, Table, UniqueConstraint
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy import DateTime, Integer
+
 from nova import context
+from nova.db.sqlalchemy import models
 from nova.db.sqlalchemy import session
+from nova import exception
 from nova import test
 
 
@@ -64,3 +70,60 @@ class DbPoolTestCase(test.TestCase):
         self.assertEqual(info['kwargs']['max_idle'], 11)
         self.assertEqual(info['kwargs']['min_size'], 21)
         self.assertEqual(info['kwargs']['max_size'], 42)
+
+
+BASE = declarative_base()
+_TABLE_NAME = '__tmp__test__tmp__'
+
+
+class TmpTable(BASE, models.NovaBase):
+    __tablename__ = _TABLE_NAME
+    id = Column(Integer, primary_key=True)
+    foo = Column(Integer)
+
+
+class SessionErrorWrapperTestCase(test.TestCase):
+    def setUp(self):
+        super(SessionErrorWrapperTestCase, self).setUp()
+        meta = MetaData()
+        meta.bind = session.get_engine()
+        test_table = Table(_TABLE_NAME, meta,
+                           Column('id', Integer, primary_key=True,
+                                  nullable=False),
+                           Column('deleted', Integer, default=0),
+                           Column('deleted_at', DateTime),
+                           Column('updated_at', DateTime),
+                           Column('created_at', DateTime),
+                           Column('foo', Integer),
+                           UniqueConstraint('foo', name='uniq_foo'))
+        test_table.create()
+
+    def tearDown(self):
+        super(SessionErrorWrapperTestCase, self).tearDown()
+        meta = MetaData()
+        meta.bind = session.get_engine()
+        test_table = Table(_TABLE_NAME, meta, autoload=True)
+        test_table.drop()
+
+    def test_flush_wrapper(self):
+        tbl = TmpTable()
+        tbl.update({'foo': 10})
+        tbl.save()
+
+        tbl2 = TmpTable()
+        tbl2.update({'foo': 10})
+        self.assertRaises(exception.DBDuplicateEntry, tbl2.save)
+
+    def test_execute_wrapper(self):
+        _session = session.get_session()
+        with _session.begin():
+            for i in [10, 20]:
+                tbl = TmpTable()
+                tbl.update({'foo': i})
+                tbl.save(session=_session)
+
+            method = _session.query(TmpTable).\
+                            filter_by(foo=10).\
+                            update
+            self.assertRaises(exception.DBDuplicateEntry,
+                              method, {'foo': 20})