[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
...
>>>
--- 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}))