cubicweb/server/test/unittest_migractions.py
changeset 12238 6ed86e0b0222
parent 12237 2dd0dcb2e5f9
child 12239 19aae64c4010
equal deleted inserted replaced
12237:2dd0dcb2e5f9 12238:6ed86e0b0222
    22 import sys
    22 import sys
    23 from datetime import date
    23 from datetime import date
    24 from contextlib import contextmanager
    24 from contextlib import contextmanager
    25 import tempfile
    25 import tempfile
    26 
    26 
    27 from logilab.common.testlib import unittest_main, Tags, tag
       
    28 from logilab.common import tempattr
    27 from logilab.common import tempattr
    29 
    28 
    30 from yams.constraints import UniqueConstraint
    29 from yams.constraints import UniqueConstraint
    31 
    30 
    32 from cubicweb import (ConfigurationError, ValidationError,
    31 from cubicweb import (ConfigurationError, ValidationError,
    68 class MigrationTC(CubicWebTC):
    67 class MigrationTC(CubicWebTC):
    69 
    68 
    70     appid = 'data-migractions'
    69     appid = 'data-migractions'
    71 
    70 
    72     configcls = MigrationConfig
    71     configcls = MigrationConfig
    73 
       
    74     tags = CubicWebTC.tags | Tags(('server', 'migration', 'migractions'))
       
    75 
    72 
    76     def _init_repo(self):
    73     def _init_repo(self):
    77         super(MigrationTC, self)._init_repo()
    74         super(MigrationTC, self)._init_repo()
    78         # we have to read schema from the database to get eid for schema entities
    75         # we have to read schema from the database to get eid for schema entities
    79         self.repo.set_schema(self.repo.deserialize_schema(), resetvreg=False)
    76         self.repo.set_schema(self.repo.deserialize_schema(), resetvreg=False)
   508             mh.cmd_sync_schema_props_perms('travaille', commit=True)
   505             mh.cmd_sync_schema_props_perms('travaille', commit=True)
   509             rdef = self.schema['travaille'].rdefs[('Personne', 'Societe')]
   506             rdef = self.schema['travaille'].rdefs[('Personne', 'Societe')]
   510             newconstraints = rdef.constraints
   507             newconstraints = rdef.constraints
   511             self.assertEqual(len(newconstraints), 0)
   508             self.assertEqual(len(newconstraints), 0)
   512 
   509 
   513     @tag('longrun')
       
   514     def test_sync_schema_props_perms(self):
   510     def test_sync_schema_props_perms(self):
   515         with self.mh() as (cnx, mh):
   511         with self.mh() as (cnx, mh):
   516             nbrqlexpr_start = cnx.execute('Any COUNT(X) WHERE X is RQLExpression')[0][0]
   512             nbrqlexpr_start = cnx.execute('Any COUNT(X) WHERE X is RQLExpression')[0][0]
   517             migrschema['titre'].rdefs[('Personne', 'String')].order = 7
   513             migrschema['titre'].rdefs[('Personne', 'String')].order = 7
   518             migrschema['adel'].rdefs[('Personne', 'String')].order = 6
   514             migrschema['adel'].rdefs[('Personne', 'String')].order = 6
   651             try:
   647             try:
   652                 mh.cmd_set_size_constraint('CWEType', 'description', 256)
   648                 mh.cmd_set_size_constraint('CWEType', 'description', 256)
   653             finally:
   649             finally:
   654                 mh.cmd_set_size_constraint('CWEType', 'description', None)
   650                 mh.cmd_set_size_constraint('CWEType', 'description', None)
   655 
   651 
   656     @tag('longrun')
       
   657     def test_add_drop_cube_and_deps(self):
   652     def test_add_drop_cube_and_deps(self):
   658         with self.mh() as (cnx, mh):
   653         with self.mh() as (cnx, mh):
   659             schema = self.repo.schema
   654             schema = self.repo.schema
   660             self.assertEqual(sorted((str(s), str(o)) for s, o in schema['see_also'].rdefs),
   655             self.assertEqual(sorted((str(s), str(o)) for s, o in schema['see_also'].rdefs),
   661                              sorted([('EmailThread', 'EmailThread'), ('Folder', 'Folder'),
   656                              sorted([('EmailThread', 'EmailThread'), ('Folder', 'Folder'),
   709                 self.maxeid = cnx.execute('Any MAX(X)')[0][0]
   704                 self.maxeid = cnx.execute('Any MAX(X)')[0][0]
   710                 # why this commit is necessary is unclear to me (though without it
   705                 # why this commit is necessary is unclear to me (though without it
   711                 # next test may fail complaining of missing tables
   706                 # next test may fail complaining of missing tables
   712                 cnx.commit()
   707                 cnx.commit()
   713 
   708 
   714 
       
   715     @tag('longrun')
       
   716     def test_add_drop_cube_no_deps(self):
   709     def test_add_drop_cube_no_deps(self):
   717         with self.mh() as (cnx, mh):
   710         with self.mh() as (cnx, mh):
   718             cubes = set(self.config.cubes())
   711             cubes = set(self.config.cubes())
   719             schema = self.repo.schema
   712             schema = self.repo.schema
   720             try:
   713             try:
   736         with self.mh() as (cnx, mh):
   729         with self.mh() as (cnx, mh):
   737             with self.assertRaises(ConfigurationError) as cm:
   730             with self.assertRaises(ConfigurationError) as cm:
   738                 mh.cmd_drop_cube('file')
   731                 mh.cmd_drop_cube('file')
   739             self.assertEqual(str(cm.exception), "can't remove cube file, used as a dependency")
   732             self.assertEqual(str(cm.exception), "can't remove cube file, used as a dependency")
   740 
   733 
   741     @tag('longrun')
       
   742     def test_introduce_base_class(self):
   734     def test_introduce_base_class(self):
   743         with self.mh() as (cnx, mh):
   735         with self.mh() as (cnx, mh):
   744             mh.cmd_add_entity_type('Para')
   736             mh.cmd_add_entity_type('Para')
   745             self.assertEqual(sorted(et.type for et in self.schema['Para'].specialized_by()),
   737             self.assertEqual(sorted(et.type for et in self.schema['Para'].specialized_by()),
   746                              ['Note'])
   738                              ['Note'])
  1011             self.assertEqual([[300]],
  1003             self.assertEqual([[300]],
  1012                              cnx.execute('Any CS WHERE C score100 CS, C is Company').rows)
  1004                              cnx.execute('Any CS WHERE C score100 CS, C is Company').rows)
  1013 
  1005 
  1014 
  1006 
  1015 if __name__ == '__main__':
  1007 if __name__ == '__main__':
  1016     unittest_main()
  1008     import unittest
       
  1009     unittest.main()