Changeset 4407

Timestamp:

03/28/09 23:00:51 (4 years ago)

Author:

gpolo

Message:

New tests, new utility functions.

Files:

• branch/merger/tests/test_sqlite_column_merge.py (modified) (3 diffs)

branch/merger/tests/test_sqlite_column_merge.py

@@ -5 +5 @@
 from sqlitedb import sqlite, merge, _DictCursor
 
+def conn_cursor(name):
+    conn = sqlite.connect(name)
+    return conn, conn.cursor()
+
+def _table_name():
+    """Generates an unique table name each time."""
+    start = 'a'
+    while True:
+        yield start
+        new_letter = ord(start[-1]) + 1
+        if new_letter > 122:
+            start = 'a' * (len(start) + 1)
+        else:
+            start = start[:-1] + chr(new_letter)
+
+
 class ColumnMergeTest(unittest.TestCase):
 
@@ -10 +26 @@
         self.new_db_file = tempfile.NamedTemporaryFile()
         self.old_db_file = tempfile.NamedTemporaryFile()
-        self.new_db = sqlite.connect(self.new_db_file.name)
-        self.old_db = sqlite.connect(self.old_db_file.name)
-        self.new_cursor = self.new_db.cursor(_DictCursor)
-        self.old_cursor = self.old_db.cursor(_DictCursor)
+        self.new_db, self.new_cursor = conn_cursor(self.new_db_file.name)
+        self.old_db, self.old_cursor = conn_cursor(self.old_db_file.name)
+
+        self.table_name = _table_name()
+
 
     def tearDown(self):
@@ -24 +41 @@
 
 
-    def test_column_merge(self):
-        self.new_cursor.execute("CREATE TABLE a (b INTEGER, c INTEGER)")
-        self.old_cursor.execute("CREATE TABLE a (b INTEGER)")
+    def _prepare_test(self, new, old):
+        table = self.table_name.next()
+        self.new_cursor.execute("CREATE TABLE %s (%s)" % (table, new))
+        self.old_cursor.execute("CREATE TABLE %s (%s)" % (table, old))
+        return table
 
-        # I don't care if the pragma table_info still exists or not, if it
-        # doesn't then let the test fail.
-
-        old_info = self.old_cursor.execute("pragma table_info('a')").fetchall()
-        self.failUnlessEqual(len(old_info), 1)
+    def _verify_merge(self, new, old):
+        table = self._prepare_test(new, old)
+        new_table_info = self.new_cursor.execute(
+                "pragma table_info(%s)" % table).fetchall()
 
         merge(self.new_db_file.name, self.old_db_file.name)
 
-        old_info = self.old_cursor.execute("pragma table_info('a')").fetchall()
-        self.failUnlessEqual(len(old_info), 2)
+        # Need to reconnect to get the updated table information.
+        conn, cursor = conn_cursor(self.old_db_file.name)
+        # I don't care if the pragma table_info still exists or not, if it
+        # doesn't then let the test fail.
+        old_table_info = cursor.execute(
+                "pragma table_info(%s)" % table).fetchall()
+        self.failUnlessEqual(len(old_table_info), len(new_table_info))
+        self.failUnlessEqual(old_table_info, new_table_info)
+        cursor.close()
+        conn.close()
+
+    def _verify_merge_fails(self, exc, table, new, old):
+        self._prepare_test(table, new, old)
+
+        self.failUnlessRaises(exc,
+                merge, self.new_db_file.name, self.old_db_file.name)
+
+
+    def test_column_merge(self):
+        # XXX not supported yet
+        #self.new_cursor.execute("CREATE TABLE a (b INTEGER, "
+        #        "c INTEGER CONSTRAINT fk REFERENCES d(id))")
+
+        self._verify_merge("a",
+                "b INTEGER, c INTEGER",
+                "b INTEGER")
+
+        self._verify_merge(
+                "b INTEGER, c INTEGER NOT NULL DEFAULT 3",
+                "b INTEGER")
+
+        self._verify_merge_fails(sqlite.OperationalError,
+                "b INTEGER, c INTEGER NOT NULL",
+                "b INTEGER")
+
+        self._verify_merge_fails(Exception,
+                "b INTEGER, c INTEGERY PRIMARY KEY",
+                "b INTEGER")