openstack/deb-python-oslo.db
Code Issues Proposed changes

Fix DBReferenceError on MySQL and SQLite

Browse Source 
It turns out the test was wrong as the error message used was not a good
one. Real world tests showed that, so I've updated the test and the code
to work.

Refactored tests to use real backends, fix error message for SQLite.

Co-Authored-By: Victor Sergeyev <vsergeyev@mirantis.com>

Change-Id: Ifa64478a9ac9ed0a3c317a8974615b770c85f313
This commit is contained in:
Julien Danjou
2014-08-22 18:17:20 +02:00
committed by Victor Sergeyev
parent be6f9a0ff7
commit 4ed63fac85
2 changed files with 103 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
oslo/db/sqlalchemy/exc_filters.py
View File

@@ -169,7 +169,7 @@ def _sqlite_dupe_key_error(integrity_error, match, engine_name, is_disconnect):
@filters("sqlite", sqla_exc.IntegrityError,
r".*SQL error: foreign key constraint failed")
r"(?i).*foreign key constraint failed")
@filters("postgresql", sqla_exc.IntegrityError,
r".*on table \"(?P<table>[^\"]+)\" violates "
"foreign key constraint \"(?P<constraint>[^\"]+)\"\s*\n"
@@ -177,13 +177,13 @@ def _sqlite_dupe_key_error(integrity_error, match, engine_name, is_disconnect):
"is not present in table "
"\"(?P<key_table>[^\"]+)\".")
@filters("mysql", sqla_exc.IntegrityError,
r".* Cannot add or update a child row: "
"a foreign key constraint fails "
"\((?P<table>.+), CONSTRAINT (?P<constraint>.+) "
"FOREIGN KEY \((?P<key>.+)\) "
"REFERENCES (?P<key_table>.+) \(.+\)\)")
r".* 'Cannot add or update a child row: "
'a foreign key constraint fails \([`"].+[`"]\.[`"](?P<table>.+)[`"], '
'CONSTRAINT [`"](?P<constraint>.+)[`"] FOREIGN KEY '
'\([`"](?P<key>.+)[`"]\) REFERENCES [`"](?P<key_table>.+)[`"] ')
def _foreign_key_error(integrity_error, match, engine_name, is_disconnect):
"""Filter for foreign key errors."""
try:
table = match.group("table")
except IndexError:

169
tests/sqlalchemy/test_exc_filters.py
View File

@@ -195,87 +195,112 @@ class TestFallthroughsAndNonDBAPI(TestsExceptionFilter):
self.assertEqual("mysqldb has an attribute error", matched.args[0])
class TestRaiseReferenceError(TestsExceptionFilter):
def test_postgresql(self):
e = self._run_test(
"postgresql",
"INSERT SOMETHING",
self.IntegrityError(
"insert or update on table "
"\"resource_entity\" "
"violates foreign key constraint "
"\"resource_entity_entity_id_fkey\"\n"
"DETAIL: Key "
"(entity_id)=(74b5da71-5a9c-4f89-a8e9-4a2d856e6c29) "
"is not present in table \"entity\".\n"
"'INSERT INTO resource_entity (resource_id, entity_id, name) "
"VALUES (%(resource_id)s, "
"%(entity_id)s, %(name)s)' "
"{'entity_id': '74b5da71-5a9c-4f89-a8e9-4a2d856e6c29', "
"'name': u'foo', "
"'resource_id': 'ffb12cb4-d955-4d96-a315-5f48ea161eef'}"),
exception.DBReferenceError,
)
self.assertEqual("resource_entity", e.table)
self.assertEqual("resource_entity_entity_id_fkey", e.constraint)
self.assertEqual("entity_id", e.key)
self.assertEqual("entity", e.key_table)
self.assertEqual(
"(IntegrityError) insert or update on table "
"\"resource_entity\" violates foreign key constraint "
"\"resource_entity_entity_id_fkey\"\n"
"DETAIL: Key (entity_id)=(74b5da71-5a9c-4f89-a8e9-4a2d856e6c29) "
"is not present in table \"entity\".\n"
"'INSERT INTO resource_entity (resource_id, entity_id, name) "
"VALUES (%(resource_id)s, %(entity_id)s, %(name)s)' "
"{'entity_id': '74b5da71-5a9c-4f89-a8e9-4a2d856e6c29', "
"'name': u'foo', "
"'resource_id': 'ffb12cb4-d955-4d96-a315-5f48ea161eef'} "
"'INSERT SOMETHING' ()",
str(e))
class TestReferenceErrorSQLite(test_base.DbTestCase):
def test_mysql(self):
e = self._run_test(
"mysql",
"INSERT SOMETHING",
self.IntegrityError(
"Cannot add or update a child row: "
"a foreign key constraint fails "
"(resource_entity, CONSTRAINT resource_entity_entity_id_fkey "
"FOREIGN KEY (entity_id) "
"REFERENCES entity (entity_id))"
),
exception.DBReferenceError,
)
self.assertEqual("resource_entity", e.table)
self.assertEqual("resource_entity_entity_id_fkey", e.constraint)
self.assertEqual("entity_id", e.key)
self.assertEqual("entity", e.key_table)
self.assertEqual(
"(IntegrityError) Cannot add or update a child row: "
"a foreign key constraint fails "
"(resource_entity, CONSTRAINT resource_entity_entity_id_fkey "
"FOREIGN KEY (entity_id) REFERENCES entity (entity_id)) "
"'INSERT SOMETHING' ()",
str(e))
def setUp(self):
super(TestReferenceErrorSQLite, self).setUp()
def test_sqlite(self):
e = self._run_test(
"sqlite",
"INSERT SOMETHING",
self.IntegrityError(
"SQL error: foreign key constraint failed"
),
exception.DBReferenceError,
meta = sqla.MetaData(bind=self.engine)
table_1 = sqla.Table(
"resource_foo", meta,
sqla.Column("id", sqla.Integer, primary_key=True),
sqla.Column("foo", sqla.Integer),
mysql_engine='InnoDB',
mysql_charset='utf8',
)
table_1.create()
self.table_2 = sqla.Table(
"resource_entity", meta,
sqla.Column("id", sqla.Integer, primary_key=True),
sqla.Column("foo_id", sqla.Integer,
sqla.ForeignKey("resource_foo.id", name="foo_fkey")),
mysql_engine='InnoDB',
mysql_charset='utf8',
)
self.table_2.create()
def test_raise(self):
self.engine.execute("PRAGMA foreign_keys = ON;")
e = self.assertRaises(
exception.DBReferenceError,
self.engine.execute,
self.table_2.insert({'id': 1, 'foo_id': 2})
)
self.assertEqual(
"(IntegrityError) FOREIGN KEY constraint failed u'INSERT INTO "
"resource_entity (id, foo_id) VALUES (?, ?)' (1, 2)".lower(),
str(e).lower())
self.assertIsNone(e.table)
self.assertIsNone(e.constraint)
self.assertIsNone(e.key)
self.assertIsNone(e.key_table)
class TestReferenceErrorPostgreSQL(TestReferenceErrorSQLite,
test_base.PostgreSQLOpportunisticTestCase):
def test_raise(self):
e = self.assertRaises(
exception.DBReferenceError,
self.engine.execute,
self.table_2.insert({'id': 1, 'foo_id': 2})
)
self.assertIn(
"(IntegrityError) insert or update on table \"resource_entity\" "
"violates foreign key constraint \"foo_fkey\"\nDETAIL: Key "
"(foo_id)=(2) is not present in table \"resource_foo\".\n"
" 'INSERT INTO resource_entity (id, foo_id) VALUES (%(id)s, "
"%(foo_id)s)'", str(e))
self.assertEqual("resource_entity", e.table)
self.assertEqual("foo_fkey", e.constraint)
self.assertEqual("foo_id", e.key)
self.assertEqual("resource_foo", e.key_table)
class TestReferenceErrorMySQL(TestReferenceErrorSQLite,
test_base.MySQLOpportunisticTestCase):
def test_raise(self):
e = self.assertRaises(
exception.DBReferenceError,
self.engine.execute,
self.table_2.insert({'id': 1, 'foo_id': 2})
)
self.assertEqual(
"(IntegrityError) SQL error: foreign key "
"constraint failed 'INSERT SOMETHING' ()",
"(IntegrityError) (1452, 'Cannot add or update a child row: a "
"foreign key constraint fails (`{0}`.`resource_entity`, "
"CONSTRAINT `foo_fkey` FOREIGN KEY (`foo_id`) REFERENCES "
"`resource_foo` (`id`))') 'INSERT INTO resource_entity (id, foo_id"
") VALUES (%s, %s)' (1, 2)".format(self.engine.url.database),
str(e))
self.assertEqual("resource_entity", e.table)
self.assertEqual("foo_fkey", e.constraint)
self.assertEqual("foo_id", e.key)
self.assertEqual("resource_foo", e.key_table)
def test_raise_ansi_quotes(self):
self.engine.execute("SET SESSION sql_mode = 'ANSI';")
e = self.assertRaises(
exception.DBReferenceError,
self.engine.execute,
self.table_2.insert({'id': 1, 'foo_id': 2})
)
self.assertEqual(
"(IntegrityError) (1452, 'Cannot add or update a child row: a "
'foreign key constraint fails ("{0}"."resource_entity", '
'CONSTRAINT "foo_fkey" FOREIGN KEY ("foo_id") REFERENCES '
'"resource_foo" ("id"))\') \'INSERT INTO resource_entity (id, '
"foo_id) VALUES (%s, %s)' (1, 2)".format(self.engine.url.database),
str(e))
self.assertEqual("resource_entity", e.table)
self.assertEqual("foo_fkey", e.constraint)
self.assertEqual("foo_id", e.key)
self.assertEqual("resource_foo", e.key_table)
class TestDuplicate(TestsExceptionFilter):