# HG changeset patch # User Adrien Di Mascio # Date 1285079600 -7200 # Node ID 054fa36060d5cd06dccbda06e59c636300db0b4a # Parent 42079f752a9c5232c6751eee07d5ed9cfc70ac43 [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 ... >>> diff -r 42079f752a9c -r 054fa36060d5 server/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 diff -r 42079f752a9c -r 054fa36060d5 server/test/data/migratedapp/schema.py --- 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')]) diff -r 42079f752a9c -r 054fa36060d5 server/test/unittest_migractions.py --- 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}))