diff -r eecb7bbb6795 -r 1d0f4064a87e server/test/unittest_migractions.py --- a/server/test/unittest_migractions.py Tue Jun 23 10:26:03 2015 +0200 +++ b/server/test/unittest_migractions.py Fri May 15 18:11:08 2015 +0200 @@ -26,6 +26,7 @@ from yams.constraints import UniqueConstraint from cubicweb import ConfigurationError, ValidationError, ExecutionError +from cubicweb.devtools import startpgcluster, stoppgcluster from cubicweb.devtools.testlib import CubicWebTC from cubicweb.server.sqlutils import SQL_PREFIX from cubicweb.server.migractions import ServerMigrationHelper @@ -35,6 +36,11 @@ HERE = osp.dirname(osp.abspath(__file__)) + +def setUpModule(): + startpgcluster(__file__) + + migrschema = None def tearDownModule(*args): global migrschema @@ -43,10 +49,16 @@ del MigrationCommandsTC.origschema if hasattr(MigrationCommandsComputedTC, 'origschema'): del MigrationCommandsComputedTC.origschema + stoppgcluster(__file__) + + +class MigrationConfig(cubicweb.devtools.TestServerConfiguration): + default_sources = cubicweb.devtools.DEFAULT_PSQL_SOURCES + class MigrationTC(CubicWebTC): - configcls = cubicweb.devtools.TestServerConfiguration + configcls = MigrationConfig tags = CubicWebTC.tags | Tags(('server', 'migration', 'migractions')) @@ -78,17 +90,17 @@ interactive=False) def table_sql(self, mh, tablename): - result = mh.sqlexec("SELECT sql FROM sqlite_master WHERE type='table' " - "and name=%(table)s", {'table': tablename}) + result = mh.sqlexec("SELECT table_name FROM information_schema.tables WHERE LOWER(table_name)=%(table)s", + {'table': tablename.lower()}) if result: return result[0][0] return None # no such table def table_schema(self, mh, tablename): - sql = self.table_sql(mh, tablename) - assert sql, 'no table %s' % tablename - return dict(x.split()[:2] - for x in sql.split('(', 1)[1].rsplit(')', 1)[0].split(',')) + result = mh.sqlexec("SELECT column_name, data_type, character_maximum_length FROM information_schema.columns " + "WHERE LOWER(table_name) = %(table)s", {'table': tablename.lower()}) + assert result, 'no table %s' % tablename + return dict((x[0], (x[1], x[2])) for x in result) class MigrationCommandsTC(MigrationTC): @@ -160,7 +172,7 @@ self.assertEqual(self.schema['shortpara'].objects(), ('String', )) # test created column is actually a varchar(64) fields = self.table_schema(mh, '%sNote' % SQL_PREFIX) - self.assertEqual(fields['%sshortpara' % SQL_PREFIX], 'varchar(64)') + self.assertEqual(fields['%sshortpara' % SQL_PREFIX], ('character varying', 64)) # test default value set on existing entities self.assertEqual(cnx.execute('Note X').get_entity(0, 0).shortpara, 'hop') # test default value set for next entities @@ -212,6 +224,7 @@ droprequired=True): mh.cmd_add_attribute('Note', 'unique_id') mh.rqlexec('INSERT Note N') + mh.rqlexec('SET N unique_id "x"') # make sure the required=True was restored self.assertRaises(ValidationError, mh.rqlexec, 'INSERT Note N') mh.rollback() @@ -779,7 +792,7 @@ self.assertEqual(self.schema['score'].rdefs['Company', 'Float'].formula, 'Any AVG(NN) WHERE X employees E, N concerns E, N note NN') fields = self.table_schema(mh, '%sCompany' % SQL_PREFIX) - self.assertEqual(fields['%sscore' % SQL_PREFIX], 'float') + self.assertEqual(fields['%sscore' % SQL_PREFIX], ('double precision', None)) self.assertEqual([[3.0]], mh.rqlexec('Any CS WHERE C score CS, C is Company').rows) @@ -803,10 +816,9 @@ def assert_computed_attribute_dropped(self): self.assertNotIn('note20', self.schema) - # DROP COLUMN not supported by sqlite - #with self.mh() as (cnx, mh): - # fields = self.table_schema(mh, '%sNote' % SQL_PREFIX) - #self.assertNotIn('%snote20' % SQL_PREFIX, fields) + with self.mh() as (cnx, mh): + fields = self.table_schema(mh, '%sNote' % SQL_PREFIX) + self.assertNotIn('%snote20' % SQL_PREFIX, fields) def test_computed_attribute_drop_type(self): self.assertIn('note20', self.schema)