[migractions] add a dropped_constraints() migration command
authorAdrien Di Mascio <Adrien.DiMascio@logilab.fr>
Tue, 21 Sep 2010 16:33:20 +0200
changeset 6292 054fa36060d5
parent 6279 42079f752a9c
child 6293 df44d7163582
[migractions] add a dropped_constraints() migration command Example usage: >>> with dropped_constraints('MyType', 'myattr', ... UniqueConstraint, droprequired=True): ... add_attribute('MyType', 'myattr') ... # + instructions to fill MyType.myattr column ... >>>
server/migractions.py
server/test/data/migratedapp/schema.py
server/test/unittest_migractions.py
--- a/server/migractions.py	Thu Sep 16 18:56:35 2010 +0200
+++ b/server/migractions.py	Tue Sep 21 16:33:20 2010 +0200
@@ -41,6 +41,7 @@
 from glob import glob
 from copy import copy
 from warnings import warn
+from contextlib import contextmanager
 
 from logilab.common.deprecation import deprecated
 from logilab.common.decorators import cached, clear_cache
@@ -48,6 +49,7 @@
 
 from yams.constraints import SizeConstraint
 from yams.schema2sql import eschema2sql, rschema2sql
+from yams.schema import RelationDefinitionSchema
 
 from cubicweb import AuthenticationError, ExecutionError
 from cubicweb.selectors import is_instance
@@ -1116,11 +1118,20 @@
         """synchronize the persistent schema against the current definition
         schema.
 
+        `ertype` can be :
+        - None, in that case everything will be synced ;
+        - a string, it should be an entity type or
+          a relation type. In that case, only the corresponding
+          entities / relations will be synced ;
+        - an rdef object to synchronize only this specific relation definition
+
         It will synch common stuff between the definition schema and the
         actual persistent schema, it won't add/remove any entity or relation.
         """
         assert syncperms or syncprops, 'nothing to do'
         if ertype is not None:
+            if isinstance(ertype, RelationDefinitionSchema):
+                ertype = ertype.as_triple()
             if isinstance(ertype, (tuple, list)):
                 assert len(ertype) == 3, 'not a relation definition'
                 self._synchronize_rdef_schema(ertype[0], ertype[1], ertype[2],
@@ -1377,6 +1388,40 @@
         """add a new entity of the given type"""
         return self.cmd_create_entity(etype, *args, **kwargs).eid
 
+    @contextmanager
+    def cmd_dropped_constraints(self, etype, attrname, cstrtype,
+                                droprequired=False):
+        """context manager to drop constraints temporarily on fs_schema
+
+        `cstrtype` should be a constraint class (or a tuple of classes)
+        and will be passed to isinstance directly
+
+        For instance::
+
+            >>> with dropped_constraints('MyType', 'myattr',
+            ...                          UniqueConstraint, droprequired=True):
+            ...     add_attribute('MyType', 'myattr')
+            ...     # + instructions to fill MyType.myattr column
+            ...
+            >>>
+
+        """
+        rdef = self.fs_schema.eschema(etype).rdef(attrname)
+        original_constraints = rdef.constraints
+        # remove constraints
+        rdef.constraints = [cstr for cstr in original_constraints
+                            if not (cstrtype and isinstance(cstr, cstrtype))]
+        if droprequired:
+            original_cardinality = rdef.cardinality
+            rdef.cardinality = '?' + rdef.cardinality[1]
+        yield
+        # restore original constraints
+        rdef.constraints = original_constraints
+        if droprequired:
+            rdef.cardinality = original_cardinality
+        # update repository schema
+        self.cmd_sync_schema_props_perms(rdef, syncperms=False)
+
     def sqlexec(self, sql, args=None, ask_confirm=True):
         """execute the given sql if confirmed
 
--- a/server/test/data/migratedapp/schema.py	Thu Sep 16 18:56:35 2010 +0200
+++ b/server/test/data/migratedapp/schema.py	Tue Sep 21 16:33:20 2010 +0200
@@ -66,6 +66,7 @@
     whatever = Int(default=2)  # keep it before `date` for unittest_migraction.test_add_attribute_int
     date = Datetime()
     type = String(maxsize=1)
+    unique_id = String(maxsize=1, required=True, unique=True)
     mydate = Date(default='TODAY')
     shortpara = String(maxsize=64)
     ecrit_par = SubjectRelation('Personne', constraints=[RQLConstraint('S concerne A, O concerne A')])
--- a/server/test/unittest_migractions.py	Thu Sep 16 18:56:35 2010 +0200
+++ b/server/test/unittest_migractions.py	Tue Sep 21 16:33:20 2010 +0200
@@ -18,13 +18,17 @@
 """unit tests for module cubicweb.server.migractions
 """
 
+from __future__ import with_statement
+
 from copy import deepcopy
 from datetime import date
 from os.path import join
 
 from logilab.common.testlib import TestCase, unittest_main
 
-from cubicweb import ConfigurationError
+from yams.constraints import UniqueConstraint
+
+from cubicweb import ConfigurationError, ValidationError
 from cubicweb.devtools.testlib import CubicWebTC
 from cubicweb.schema import CubicWebSchemaLoader
 from cubicweb.server.sqlutils import SQL_PREFIX
@@ -130,6 +134,28 @@
         self.assertEquals(d2, testdate)
         self.mh.rollback()
 
+    def test_drop_chosen_constraints_ctxmanager(self):
+        with self.mh.cmd_dropped_constraints('Note', 'unique_id', UniqueConstraint):
+            self.mh.cmd_add_attribute('Note', 'unique_id')
+            # make sure the maxsize constraint is not dropped
+            self.assertRaises(ValidationError,
+                              self.mh.rqlexec,
+                              'INSERT Note N: N unique_id "xyz"')
+            # make sure the unique constraint is dropped
+            self.mh.rqlexec('INSERT Note N: N unique_id "x"')
+            self.mh.rqlexec('INSERT Note N: N unique_id "x"')
+            self.mh.rqlexec('DELETE Note N')
+        self.mh.rollback()
+
+    def test_drop_required_ctxmanager(self):
+        with self.mh.cmd_dropped_constraints('Note', 'unique_id', cstrtype=None,
+                                             droprequired=True):
+            self.mh.cmd_add_attribute('Note', 'unique_id')
+            self.mh.rqlexec('INSERT Note N')
+        # make sure the required=True was restored
+        self.assertRaises(ValidationError, self.mh.rqlexec, 'INSERT Note N')
+        self.mh.rollback()
+
     def test_rename_attribute(self):
         self.failIf('civility' in self.schema)
         eid1 = self.mh.rqlexec('INSERT Personne X: X nom "lui", X sexe "M"')[0][0]
@@ -524,7 +550,7 @@
         self.assertEquals(self.schema['Text'].specializes().type, 'Para')
         # test columns have been actually added
         text = self.execute('INSERT Text X: X para "hip", X summary "hop", X newattr "momo"').get_entity(0, 0)
-        note = self.execute('INSERT Note X: X para "hip", X shortpara "hop", X newattr "momo"').get_entity(0, 0)
+        note = self.execute('INSERT Note X: X para "hip", X shortpara "hop", X newattr "momo", X unique_id "x"').get_entity(0, 0)
         aff = self.execute('INSERT Affaire X').get_entity(0, 0)
         self.failUnless(self.execute('SET X newnotinlined Y WHERE X eid %(x)s, Y eid %(y)s',
                                      {'x': text.eid, 'y': aff.eid}))