diff -r 058bb3dc685f -r 0b59724cb3f2 cubicweb/test/unittest_schema.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/cubicweb/test/unittest_schema.py Sat Jan 16 13:48:51 2016 +0100 @@ -0,0 +1,563 @@ +# copyright 2003-2014 LOGILAB S.A. (Paris, FRANCE), all rights reserved. +# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr +# +# This file is part of CubicWeb. +# +# CubicWeb is free software: you can redistribute it and/or modify it under the +# terms of the GNU Lesser General Public License as published by the Free +# Software Foundation, either version 2.1 of the License, or (at your option) +# any later version. +# +# CubicWeb is distributed in the hope that it will be useful, but WITHOUT +# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more +# details. +# +# You should have received a copy of the GNU Lesser General Public License along +# with CubicWeb. If not, see . +"""unit tests for module cubicweb.schema""" + +import sys +from os.path import join, isabs, basename, dirname + +from logilab.common.testlib import TestCase, unittest_main + +from rql import RQLSyntaxError + +from yams import ValidationError, BadSchemaDefinition +from yams.constraints import SizeConstraint, StaticVocabularyConstraint +from yams.buildobjs import (RelationDefinition, EntityType, RelationType, + Int, String, SubjectRelation, ComputedRelation) +from yams.reader import fill_schema + +from cubicweb.schema import ( + CubicWebSchema, CubicWebEntitySchema, CubicWebSchemaLoader, + RQLConstraint, RQLUniqueConstraint, RQLVocabularyConstraint, + RQLExpression, ERQLExpression, RRQLExpression, + normalize_expression, order_eschemas, guess_rrqlexpr_mainvars, + build_schema_from_namespace) +from cubicweb.devtools import TestServerConfiguration as TestConfiguration +from cubicweb.devtools.testlib import CubicWebTC + +DATADIR = join(dirname(__file__), 'data') + +# build a dummy schema ######################################################## + + +PERSONNE_PERMISSIONS = { + 'read': ('managers', 'users', 'guests'), + 'update': ('managers', 'owners'), + 'add': ('managers', ERQLExpression('X travaille S, S owned_by U')), + 'delete': ('managers', 'owners',), + } + +CONCERNE_PERMISSIONS = { + 'read': ('managers', 'users', 'guests'), + 'add': ('managers', RRQLExpression('U has_update_permission S')), + 'delete': ('managers', RRQLExpression('O owned_by U')), + } + +schema = CubicWebSchema('Test Schema') +enote = schema.add_entity_type(EntityType('Note')) +eaffaire = schema.add_entity_type(EntityType('Affaire')) +eperson = schema.add_entity_type(EntityType('Personne', __permissions__=PERSONNE_PERMISSIONS)) +esociete = schema.add_entity_type(EntityType('Societe')) + +RELS = ( + # attribute relations + ('Note date String'), + ('Note type String'), + ('Affaire sujet String'), + ('Affaire ref String'), + ('Personne nom String'), + ('Personne prenom String'), + ('Personne sexe String'), + ('Personne tel Int'), + ('Personne fax Int'), + ('Personne datenaiss Date'), + ('Personne promo String'), + # real relations + ('Personne travaille Societe'), + ('Personne evaluee Note'), + ('Societe evaluee Note'), + ('Personne concerne Affaire'), + ('Personne concerne Societe'), + ('Affaire concerne Societe'), + ) +done = {} +for rel in RELS: + _from, _type, _to = rel.split() + if not _type.lower() in done: + schema.add_relation_type(RelationType(_type)) + done[_type.lower()] = True + if _type == 'concerne': + schema.add_relation_def(RelationDefinition(_from, _type, _to, + __permissions__=CONCERNE_PERMISSIONS)) + else: + schema.add_relation_def(RelationDefinition(_from, _type, _to)) + +class CubicWebSchemaTC(TestCase): + + def test_rql_constraints_inheritance(self): + # isinstance(cstr, RQLVocabularyConstraint) + # -> expected to return RQLVocabularyConstraint and RQLConstraint + # instances but not RQLUniqueConstraint + # + # isinstance(cstr, RQLConstraint) + # -> expected to return RQLConstraint instances but not + # RQLVocabularyConstraint and RQLUniqueConstraint + self.assertFalse(issubclass(RQLUniqueConstraint, RQLVocabularyConstraint)) + self.assertFalse(issubclass(RQLUniqueConstraint, RQLConstraint)) + + def test_entity_perms(self): + self.assertEqual(eperson.get_groups('read'), set(('managers', 'users', 'guests'))) + self.assertEqual(eperson.get_groups('update'), set(('managers', 'owners',))) + self.assertEqual(eperson.get_groups('delete'), set(('managers', 'owners'))) + self.assertEqual(eperson.get_groups('add'), set(('managers',))) + self.assertEqual([str(e) for e in eperson.get_rqlexprs('add')], + ['Any X WHERE X travaille S, S owned_by U, X eid %(x)s, U eid %(u)s']) + eperson.set_action_permissions('read', ('managers',)) + self.assertEqual(eperson.get_groups('read'), set(('managers',))) + + def test_relation_perms(self): + rconcerne = schema.rschema('concerne').rdef('Personne', 'Societe') + self.assertEqual(rconcerne.get_groups('read'), set(('managers', 'users', 'guests'))) + self.assertEqual(rconcerne.get_groups('delete'), set(('managers',))) + self.assertEqual(rconcerne.get_groups('add'), set(('managers', ))) + rconcerne.set_action_permissions('read', ('managers',)) + self.assertEqual(rconcerne.get_groups('read'), set(('managers',))) + self.assertEqual([str(e) for e in rconcerne.get_rqlexprs('add')], + ['Any S,U WHERE U has_update_permission S, S eid %(s)s, U eid %(u)s']) + + def test_erqlexpression(self): + self.assertRaises(RQLSyntaxError, ERQLExpression, '1') + expr = ERQLExpression('X travaille S, S owned_by U') + self.assertEqual(str(expr), 'Any X WHERE X travaille S, S owned_by U, X eid %(x)s, U eid %(u)s') + expr = ERQLExpression('X foo S, S bar U, X baz XE, S quux SE HAVING XE > SE') + self.assertEqual(str(expr), 'Any X WHERE X foo S, S bar U, X baz XE, S quux SE, X eid %(x)s, U eid %(u)s HAVING XE > SE') + + def test_rrqlexpression(self): + self.assertRaises(Exception, RRQLExpression, '1') + self.assertRaises(RQLSyntaxError, RRQLExpression, 'O X Y') + expr = RRQLExpression('U has_update_permission O') + self.assertEqual(str(expr), 'Any O,U WHERE U has_update_permission O, O eid %(o)s, U eid %(u)s') + +loader = CubicWebSchemaLoader() +config = TestConfiguration('data', apphome=DATADIR) +config.bootstrap_cubes() + +class SchemaReaderClassTest(TestCase): + + def test_order_eschemas(self): + schema = loader.load(config) + self.assertEqual(order_eschemas([schema['Note'], schema['SubNote']]), + [schema['Note'], schema['SubNote']]) + self.assertEqual(order_eschemas([schema['SubNote'], schema['Note']]), + [schema['Note'], schema['SubNote']]) + + def test_knownValues_load_schema(self): + schema = loader.load(config) + self.assertIsInstance(schema, CubicWebSchema) + self.assertEqual(schema.name, 'data') + entities = sorted([str(e) for e in schema.entities()]) + expected_entities = ['Ami', 'BaseTransition', 'BigInt', 'Bookmark', 'Boolean', 'Bytes', 'Card', + 'Date', 'Datetime', 'Decimal', + 'CWCache', 'CWComputedRType', 'CWConstraint', + 'CWConstraintType', 'CWDataImport', 'CWEType', + 'CWAttribute', 'CWGroup', 'EmailAddress', + 'CWRelation', 'CWPermission', 'CWProperty', 'CWRType', + 'CWSource', 'CWSourceHostConfig', 'CWSourceSchemaConfig', + 'CWUniqueTogetherConstraint', 'CWUser', + 'ExternalUri', 'FakeFile', 'Float', 'Int', 'Interval', 'Note', + 'Password', 'Personne', 'Produit', + 'RQLExpression', 'Reference', + 'Service', 'Societe', 'State', 'StateFull', 'String', 'SubNote', 'SubWorkflowExitPoint', + 'Tag', 'TZDatetime', 'TZTime', 'Time', 'Transition', 'TrInfo', + 'Usine', + 'Workflow', 'WorkflowTransition'] + self.assertListEqual(sorted(expected_entities), entities) + relations = sorted([str(r) for r in schema.relations()]) + expected_relations = ['actionnaire', 'add_permission', 'address', 'alias', 'allowed_transition', 'associe', + 'bookmarked_by', 'by_transition', + + 'cardinality', 'comment', 'comment_format', + 'composite', 'condition', 'config', 'connait', + 'constrained_by', 'constraint_of', + 'content', 'content_format', 'contrat_exclusif', + 'created_by', 'creation_date', 'cstrtype', 'custom_workflow', + 'cwuri', 'cw_for_source', 'cw_import_of', 'cw_host_config_of', 'cw_schema', 'cw_source', + + 'data', 'data_encoding', 'data_format', 'data_name', 'default_workflow', 'defaultval', 'delete_permission', + 'description', 'description_format', 'destination_state', 'dirige', + + 'ean', 'ecrit_par', 'eid', 'end_timestamp', 'evaluee', 'expression', 'exprtype', 'extra_props', + + 'fabrique_par', 'final', 'firstname', 'for_user', 'formula', 'fournit', + 'from_entity', 'from_state', 'fulltext_container', 'fulltextindexed', + + 'has_group_permission', 'has_text', + 'identity', 'in_group', 'in_state', 'in_synchronization', 'indexed', + 'initial_state', 'inlined', 'internationalizable', 'is', 'is_instance_of', + + 'label', 'last_login_time', 'latest_retrieval', 'lieu', 'log', 'login', + + 'mainvars', 'match_host', 'modification_date', + + 'name', 'nom', + + 'options', 'ordernum', 'owned_by', + + 'parser', 'path', 'pkey', 'prefered_form', 'prenom', 'primary_email', + + 'read_permission', 'relation_type', 'relations', 'require_group', 'rule', + + 'specializes', 'start_timestamp', 'state_of', 'status', 'subworkflow', 'subworkflow_exit', 'subworkflow_state', 'surname', 'symmetric', 'synopsis', + + 'tags', 'timestamp', 'title', 'to_entity', 'to_state', 'transition_of', 'travaille', 'type', + + 'upassword', 'update_permission', 'url', 'uri', 'use_email', + + 'value', + + 'wf_info_for', 'wikiid', 'workflow_of', 'tr_count'] + + self.assertListEqual(sorted(expected_relations), relations) + + eschema = schema.eschema('CWUser') + rels = sorted(str(r) for r in eschema.subject_relations()) + self.assertListEqual(rels, ['created_by', 'creation_date', 'custom_workflow', + 'cw_source', 'cwuri', 'eid', + 'evaluee', 'firstname', 'has_group_permission', + 'has_text', 'identity', + 'in_group', 'in_state', 'is', + 'is_instance_of', 'last_login_time', + 'login', 'modification_date', 'owned_by', + 'primary_email', 'surname', 'upassword', + 'use_email']) + rels = sorted(r.type for r in eschema.object_relations()) + self.assertListEqual(rels, ['bookmarked_by', 'created_by', 'for_user', + 'identity', 'owned_by', 'wf_info_for']) + rschema = schema.rschema('relation_type') + properties = rschema.rdef('CWAttribute', 'CWRType') + self.assertEqual(properties.cardinality, '1*') + constraints = properties.constraints + self.assertEqual(len(constraints), 1, constraints) + constraint = constraints[0] + self.assertTrue(isinstance(constraint, RQLConstraint)) + self.assertEqual(constraint.expression, 'O final TRUE') + + def test_fulltext_container(self): + schema = loader.load(config) + self.assertIn('has_text', schema['CWUser'].subject_relations()) + self.assertNotIn('has_text', schema['EmailAddress'].subject_relations()) + + def test_permission_settings(self): + schema = loader.load(config) + aschema = schema['TrInfo'].rdef('comment') + self.assertEqual(aschema.get_groups('read'), + set(('managers', 'users', 'guests'))) + self.assertEqual(aschema.get_rqlexprs('read'), + ()) + self.assertEqual(aschema.get_groups('update'), + set(('managers',))) + self.assertEqual([x.expression for x in aschema.get_rqlexprs('update')], + ['U has_update_permission X']) + + def test_nonregr_allowed_type_names(self): + schema = CubicWebSchema('Test Schema') + schema.add_entity_type(EntityType('NaN')) + + def test_relation_perm_overriding(self): + loader = CubicWebSchemaLoader() + config = TestConfiguration('data', apphome=join(dirname(__file__), 'data_schemareader')) + config.bootstrap_cubes() + schema = loader.load(config) + rdef = next(iter(schema['in_group'].rdefs.values())) + self.assertEqual(rdef.permissions, + {'read': ('managers',), + 'add': ('managers',), + 'delete': ('managers',)}) + rdef = next(iter(schema['cw_for_source'].rdefs.values())) + self.assertEqual(rdef.permissions, + {'read': ('managers', 'users'), + 'add': ('managers',), + 'delete': ('managers',)}) + + def test_computed_attribute(self): + """Check schema finalization for computed attributes.""" + class Person(EntityType): + salary = Int() + + class works_for(RelationDefinition): + subject = 'Person' + object = 'Company' + cardinality = '?*' + + class Company(EntityType): + total_salary = Int(formula='Any SUM(SA) GROUPBY X WHERE ' + 'P works_for X, P salary SA') + good_schema = build_schema_from_namespace(vars().items()) + rdef = good_schema['Company'].rdef('total_salary') + # ensure 'X is Company' is added to the rqlst to avoid ambiguities, see #4901163 + self.assertEqual(str(rdef.formula_select), + 'Any SUM(SA) GROUPBY X WHERE P works_for X, P salary SA, X is Company') + # check relation definition permissions + self.assertEqual(rdef.permissions, + {'add': (), 'update': (), + 'read': ('managers', 'users', 'guests')}) + + class Company(EntityType): + total_salary = String(formula='Any SUM(SA) GROUPBY X WHERE ' + 'P works_for X, P salary SA') + + with self.assertRaises(BadSchemaDefinition) as exc: + bad_schema = build_schema_from_namespace(vars().items()) + + self.assertEqual(str(exc.exception), + 'computed attribute total_salary on Company: ' + 'computed attribute type (Int) mismatch with ' + 'specified type (String)') + + +class SchemaReaderComputedRelationAndAttributesTest(TestCase): + + def test_infer_computed_relation(self): + class Person(EntityType): + name = String() + + class Company(EntityType): + name = String() + + class Service(EntityType): + name = String() + + class works_for(RelationDefinition): + subject = 'Person' + object = 'Company' + + class produce(RelationDefinition): + subject = ('Person', 'Company') + object = 'Service' + + class achete(RelationDefinition): + subject = 'Person' + object = 'Service' + + class produces_and_buys(ComputedRelation): + rule = 'S produce O, S achete O' + + class produces_and_buys2(ComputedRelation): + rule = 'S works_for SO, SO produce O' + + class reproduce(ComputedRelation): + rule = 'S produce O' + + schema = build_schema_from_namespace(vars().items()) + + # check object/subject type + self.assertEqual([('Person','Service')], + list(schema['produces_and_buys'].rdefs.keys())) + self.assertEqual([('Person','Service')], + list(schema['produces_and_buys2'].rdefs.keys())) + self.assertCountEqual([('Company', 'Service'), ('Person', 'Service')], + list(schema['reproduce'].rdefs.keys())) + # check relation definitions are marked infered + rdef = schema['produces_and_buys'].rdefs[('Person','Service')] + self.assertTrue(rdef.infered) + # and have no add/delete permissions + self.assertEqual(rdef.permissions, + {'add': (), + 'delete': (), + 'read': ('managers', 'users', 'guests')}) + + class autoname(ComputedRelation): + rule = 'S produce X, X name O' + + with self.assertRaises(BadSchemaDefinition) as cm: + build_schema_from_namespace(vars().items()) + self.assertEqual(str(cm.exception), 'computed relations cannot be final') + + +class BadSchemaTC(TestCase): + def setUp(self): + self.loader = CubicWebSchemaLoader() + self.loader.defined = {} + self.loader.loaded_files = [] + self.loader.post_build_callbacks = [] + + def _test(self, schemafile, msg): + self.loader.handle_file(join(DATADIR, schemafile)) + sch = self.loader.schemacls('toto') + with self.assertRaises(BadSchemaDefinition) as cm: + fill_schema(sch, self.loader.defined, False) + self.assertEqual(str(cm.exception), msg) + + def test_lowered_etype(self): + self._test('lowered_etype.py', + "'my_etype' is not a valid name for an entity type. It should " + "start with an upper cased letter and be followed by at least " + "a lower cased letter") + + def test_uppered_rtype(self): + self._test('uppered_rtype.py', + "'ARelation' is not a valid name for a relation type. It should be lower cased") + + def test_rrqlexpr_on_etype(self): + self._test('rrqlexpr_on_eetype.py', + "can't use RRQLExpression on ToTo, use an ERQLExpression") + + def test_erqlexpr_on_rtype(self): + self._test('erqlexpr_on_ertype.py', + "can't use ERQLExpression on relation ToTo toto TuTu, use a RRQLExpression") + + def test_rqlexpr_on_rtype_read(self): + self._test('rqlexpr_on_ertype_read.py', + "can't use rql expression for read permission of relation ToTo toto TuTu") + + def test_rrqlexpr_on_attr(self): + self._test('rrqlexpr_on_attr.py', + "can't use RRQLExpression on attribute ToTo.attr[String], use an ERQLExpression") + + def test_rqlexpr_on_computedrel(self): + self._test('rqlexpr_on_computedrel.py', + "can't use rql expression for read permission of relation Subject computed Object") + + +class NormalizeExpressionTC(TestCase): + + def test(self): + self.assertEqual(normalize_expression('X bla Y,Y blur Z , Z zigoulou X '), + 'X bla Y, Y blur Z, Z zigoulou X') + self.assertEqual(normalize_expression('X bla Y, Y name "x,y"'), + 'X bla Y, Y name "x,y"') + + +class RQLExpressionTC(TestCase): + def test_comparison(self): + self.assertEqual(ERQLExpression('X is CWUser', 'X', 0), + ERQLExpression('X is CWUser', 'X', 0)) + self.assertNotEqual(ERQLExpression('X is CWUser', 'X', 0), + ERQLExpression('X is CWGroup', 'X', 0)) + + +class GuessRrqlExprMainVarsTC(TestCase): + def test_exists(self): + mainvars = guess_rrqlexpr_mainvars(normalize_expression('NOT EXISTS(O team_competition C, C level < 3, C concerns S)')) + self.assertEqual(mainvars, set(['S', 'O'])) + + +class RQLConstraintTC(CubicWebTC): + def test_user_constraint(self): + cstr = RQLConstraint('U identity O') + with self.admin_access.repo_cnx() as cnx: + anoneid = cnx.execute('Any X WHERE X login "anon"')[0][0] + self.assertRaises(ValidationError, + cstr.repo_check, cnx, 1, 'rel', anoneid) + self.assertEqual(cstr.repo_check(cnx, 1, cnx.user.eid), + None) # no validation error, constraint checked + + +class WorkflowShemaTC(CubicWebTC): + def test_trinfo_default_format(self): + with self.admin_access.web_request() as req: + tr = req.user.cw_adapt_to('IWorkflowable').fire_transition('deactivate') + self.assertEqual(tr.comment_format, 'text/plain') + + +class CompositeSchemaTC(CubicWebTC): + composites = { + 'BaseTransition': [('condition', 'BaseTransition', 'RQLExpression', 'subject')], + 'CWAttribute': [('add_permission', 'CWAttribute', 'RQLExpression', 'subject'), + ('constrained_by', 'CWAttribute', 'CWConstraint', 'subject'), + ('read_permission', 'CWAttribute', 'RQLExpression', 'subject'), + ('update_permission', 'CWAttribute', 'RQLExpression', 'subject')], + 'CWEType': [('add_permission', 'CWEType', 'RQLExpression', 'subject'), + ('constraint_of', 'CWUniqueTogetherConstraint', 'CWEType', 'object'), + ('cw_schema', 'CWSourceSchemaConfig', 'CWEType', 'object'), + ('delete_permission', 'CWEType', 'RQLExpression', 'subject'), + ('from_entity', 'CWAttribute', 'CWEType', 'object'), + ('from_entity', 'CWRelation', 'CWEType', 'object'), + ('read_permission', 'CWEType', 'RQLExpression', 'subject'), + ('to_entity', 'CWAttribute', 'CWEType', 'object'), + ('to_entity', 'CWRelation', 'CWEType', 'object'), + ('update_permission', 'CWEType', 'RQLExpression', 'subject')], + 'CWRType': [('cw_schema', 'CWSourceSchemaConfig', 'CWRType', 'object'), + ('relation_type', 'CWAttribute', 'CWRType', 'object'), + ('relation_type', 'CWRelation', 'CWRType', 'object')], + 'CWRelation': [('add_permission', 'CWRelation', 'RQLExpression', 'subject'), + ('constrained_by', 'CWRelation', 'CWConstraint', 'subject'), + ('cw_schema', 'CWSourceSchemaConfig', 'CWRelation', 'object'), + ('delete_permission', 'CWRelation', 'RQLExpression', 'subject'), + ('read_permission', 'CWRelation', 'RQLExpression', 'subject')], + 'CWComputedRType': [('read_permission', 'CWComputedRType', 'RQLExpression', 'subject')], + 'CWSource': [('cw_for_source', 'CWSourceSchemaConfig', 'CWSource', 'object'), + ('cw_host_config_of', 'CWSourceHostConfig', 'CWSource', 'object'), + ('cw_import_of', 'CWDataImport', 'CWSource', 'object'), + ('cw_source', 'Ami', 'CWSource', 'object'), + ('cw_source', 'BaseTransition', 'CWSource', 'object'), + ('cw_source', 'Bookmark', 'CWSource', 'object'), + ('cw_source', 'CWAttribute', 'CWSource', 'object'), + ('cw_source', 'CWCache', 'CWSource', 'object'), + ('cw_source', 'CWComputedRType', 'CWSource', 'object'), + ('cw_source', 'CWConstraint', 'CWSource', 'object'), + ('cw_source', 'CWConstraintType', 'CWSource', 'object'), + ('cw_source', 'CWDataImport', 'CWSource', 'object'), + ('cw_source', 'CWEType', 'CWSource', 'object'), + ('cw_source', 'CWGroup', 'CWSource', 'object'), + ('cw_source', 'CWPermission', 'CWSource', 'object'), + ('cw_source', 'CWProperty', 'CWSource', 'object'), + ('cw_source', 'CWRType', 'CWSource', 'object'), + ('cw_source', 'CWRelation', 'CWSource', 'object'), + ('cw_source', 'CWSource', 'CWSource', 'object'), + ('cw_source', 'CWSourceHostConfig', 'CWSource', 'object'), + ('cw_source', 'CWSourceSchemaConfig', 'CWSource', 'object'), + ('cw_source', 'CWUniqueTogetherConstraint', 'CWSource', 'object'), + ('cw_source', 'CWUser', 'CWSource', 'object'), + ('cw_source', 'Card', 'CWSource', 'object'), + ('cw_source', 'EmailAddress', 'CWSource', 'object'), + ('cw_source', 'ExternalUri', 'CWSource', 'object'), + ('cw_source', 'FakeFile', 'CWSource', 'object'), + ('cw_source', 'Note', 'CWSource', 'object'), + ('cw_source', 'Personne', 'CWSource', 'object'), + ('cw_source', 'Produit', 'CWSource', 'object'), + ('cw_source', 'RQLExpression', 'CWSource', 'object'), + ('cw_source', 'Reference', 'CWSource', 'object'), + ('cw_source', 'Service', 'CWSource', 'object'), + ('cw_source', 'Societe', 'CWSource', 'object'), + ('cw_source', 'State', 'CWSource', 'object'), + ('cw_source', 'StateFull', 'CWSource', 'object'), + ('cw_source', 'SubNote', 'CWSource', 'object'), + ('cw_source', 'SubWorkflowExitPoint', 'CWSource', 'object'), + ('cw_source', 'Tag', 'CWSource', 'object'), + ('cw_source', 'TrInfo', 'CWSource', 'object'), + ('cw_source', 'Transition', 'CWSource', 'object'), + ('cw_source', 'Usine', 'CWSource', 'object'), + ('cw_source', 'Workflow', 'CWSource', 'object'), + ('cw_source', 'WorkflowTransition', 'CWSource', 'object')], + 'CWUser': [('for_user', 'CWProperty', 'CWUser', 'object'), + ('use_email', 'CWUser', 'EmailAddress', 'subject'), + ('wf_info_for', 'TrInfo', 'CWUser', 'object')], + 'StateFull': [('wf_info_for', 'TrInfo', 'StateFull', 'object')], + 'Transition': [('condition', 'Transition', 'RQLExpression', 'subject')], + 'Workflow': [('state_of', 'State', 'Workflow', 'object'), + ('transition_of', 'BaseTransition', 'Workflow', 'object'), + ('transition_of', 'Transition', 'Workflow', 'object'), + ('transition_of', 'WorkflowTransition', 'Workflow', 'object')], + 'WorkflowTransition': [('condition', 'WorkflowTransition', 'RQLExpression', 'subject'), + ('subworkflow_exit', 'WorkflowTransition', 'SubWorkflowExitPoint', 'subject')] + } + + def test_composite_entities(self): + schema = self.vreg.schema + self.assertEqual(sorted(self.composites), + [eschema.type for eschema in sorted(schema.entities()) + if eschema.is_composite]) + for etype in self.composites: + self.set_description('composite rdefs for %s' % etype) + yield self.assertEqual, self.composites[etype], \ + sorted([(r.rtype.type, r.subject.type, r.object.type, role) + for r, role in schema[etype].composite_rdef_roles]) + + +if __name__ == '__main__': + unittest_main()