[dataimport] check that MassiveObjectStore restores the db schema properly
Ideally we'd use "pg_dump --schema-only" to compare before/after, but when
restoring a constraint such as:
cstr48f9ceae537d68859b62c601681c8d3e CHECK (((cw_type)::text = ANY ((ARRAY['normal'::character varying, 'auto'::character varying])::text[])))
postgres will convert the text cast on array to a cast on each cell of array
cstr48f9ceae537d68859b62c601681c8d3e CHECK (cw_type::text = ANY (ARRAY['normal'::character varying::text, 'auto'::character varying::text]))
--- a/dataimport/test/test_massive_store.py Mon Dec 21 12:41:30 2015 +0100
+++ b/dataimport/test/test_massive_store.py Mon Dec 21 12:29:50 2015 +0100
@@ -21,7 +21,7 @@
from cubicweb.dataimport import ucsvreader
from cubicweb.devtools import testlib, PostgresApptestConfiguration
from cubicweb.devtools import startpgcluster, stoppgcluster
-from cubicweb.dataimport.massive_store import MassiveObjectStore
+from cubicweb.dataimport.massive_store import MassiveObjectStore, PGHelper
def setUpModule():
@@ -165,6 +165,33 @@
store.flush()
self.assertEqual(eid, 10000)
+ @staticmethod
+ def get_db_descr(cnx):
+ pg_schema = (
+ cnx.repo.config.system_source_config.get('db-namespace')
+ or 'public')
+ pgh = PGHelper(cnx, pg_schema)
+ all_tables = cnx.system_sql('''
+SELECT table_name
+FROM information_schema.tables
+where table_schema = %(s)s''', {'s': pg_schema}).fetchall()
+ all_tables_descr = {}
+ for tablename, in all_tables:
+ all_tables_descr[tablename] = set(pgh.index_list(tablename)).union(
+ set(pgh.constraint_list(tablename)))
+ return all_tables_descr
+
+ def test_identical_schema(self):
+ with self.admin_access.repo_cnx() as cnx:
+ init_descr = self.get_db_descr(cnx)
+ with self.admin_access.repo_cnx() as cnx:
+ store = MassiveObjectStore(cnx)
+ store.init_etype_table('CWUser')
+ store.finish()
+ with self.admin_access.repo_cnx() as cnx:
+ final_descr = self.get_db_descr(cnx)
+ self.assertEqual(init_descr, final_descr)
+
def test_on_commit_callback(self):
counter = itertools.count()
with self.admin_access.repo_cnx() as cnx: