associations/ 0040755 0001751 0000144 00000000000 07743057454 012620 5 ustar roche users associations/interfaces.py 0100644 0001751 0000144 00000005664 07743047061 015316 0 ustar roche users from zope.interface import Interface, Attribute
from zope.schema import TextLine, Object, Int
from zope.schema.interfaces import IField
from zope.app.component.interfacefield import InterfaceField
from zope.app.interfaces.container import IContained
class IAssociationService(Interface):
"Provides read-write IAssociationService info"
def dissociateObject(object):
""" Remove object from all associations and dereference object
"""
def associateCopiedObject(object):
""" rebuild associations from references on copied object
"""
class IAssociationEnd(Interface):
"Stores associations as UIDs"
name = TextLine(
title=u"Name",
description=u"The name of the association end",
default=u"",
required=True)
interface = InterfaceField(
title=u"Interface",
description=(u"The required interface for objects to be stored "
u"in this association end"),
required=True)
upperbound = Int(
title=u"Upper bound",
description=u"The maximum number of references allowed to this end",
default=0,
required=True)
def add(fromUID, toUID):
"""Add an association between 'fromUID' and 'toUID' """
def delete(fromUID, toUID):
"""Delete an association between 'fromUID' and 'toUID' """
def get(UID):
""" Return a the UIDs for all associations to 'UID' """
class IAssociation(Interface):
"Describes the association between two classes"
name = TextLine(
title=u"Name",
description=u"The name of the association",
default=u"",
required=True)
description = TextLine(
title=u"Description",
description=u"A description of the association",
default=u"",
required=True)
fromEnd = Object(
IAssociationEnd,
title=u"From end",
description=u"The 'from' end of the association",
required=True)
toEnd = Object(
IAssociationEnd,
title=u"To end",
description=u"The 'to' end of the association",
required=True)
def add(fromObject, toObject):
""" Add an assocation between two objects """
def delete(fromObject, toObject):
""" Delete an association between two objects """
def get(obj):
""" get objects associated to 'obj' """
class IReference(IField):
u"""Field containing a reference to another object."""
association_name = Attribute("association_name",
_(u"The name of the Association that this "
u"reference participates in."))
association_end_name = Attribute("association_end_name",
_(u"The name of the AssociationEnd this field corresponds to."))
class IReferenceSet(IReference):
u"""Field containing a set of references to other objects."""
class IReferenceQuery(Interface):
u"""Query that provides field with allowed values"""
def query():
""" Do query """
associations/__init__.py 0100644 0001751 0000144 00000000050 07736326701 014715 0 ustar roche users AssociationServiceName = 'Associations'
associations/associationservice.py 0100644 0001751 0000144 00000005316 07743047061 017062 0 ustar roche users from persistence import Persistent
from persistence.dict import PersistentDict
from zope.interface import implements
from zope.app.interfaces.services.service import ISimpleService
from zope.app.interfaces.event import ISubscriber
from zope.app.interfaces.annotation import IAttributeAnnotatable
from zope.app.container.sample import SampleContainer
from zopeproducts.associations.interfaces import IAssociationService
class AssociationService(Persistent, SampleContainer):
__doc__ = IAssociationService.__doc__
implements(IAssociationService, ISimpleService, ISubscriber,
IAttributeAnnotatable)
_subscribed = False
def notify(self, event):
if IObjectRemovedEvent.isImplementedBy(event):
self.dissociateObject(event.object)
elif IObjectCopiedEvent.isImplementedBy(event):
self.associateCopiedObject(event.object)
def dissociateObject(self, object):
for schema in providedBy(object):
for name in schema:
field = schema[name]
if IReference.isImplementedBy(field):
reference = field.query(object)
association = self[field.association_name]
association.delete(object, reference)
# TODO: Delete the reference on the object itself
elif IReferenceSet.isImplementedBy(field)):
references = field.query(object)
association = self[field.association_name]
for reference in references:
association.delete(object, reference)
# TODO: Delete the reference on the object
# itself
# Remove references to this object
for association in self:
try:
references = association.get(object)
for reference in references
association.delete(object, reference)
except AssociationError:
pass
def associateCopiedObject(self, object):
for schema in providedBy(object):
for name in schema:
field = schema[name]
if IReference.isImplementedBy(field):
reference = field.query(object)
association = self[field.association_name]
association.add(object, reference)
elif IReferenceSet.isImplementedBy(field)):
references = field.query(object)
association = self[field.association_name]
for reference in references:
association.add(object, reference)
def _newContainerData(self):
return PersistentDict()
associations/association.py 0100644 0001751 0000144 00000005230 07743047532 015477 0 ustar roche users from persistence import Persistent
from zope.interface import implements
from zope.app.zapi import getService
from zope.app.services.servicenames import HubIds
from zope.app.container.contained import Contained
from zopeproducts.associations.interfaces import IAssociation
class Association(Persistent, Contained):
__doc__ = IAssociation.__doc__
implements(IAssociation)
def __init__(self, name, description, fromEnd, toEnd):
self.name = name
self.description = description
self.fromEnd = fromEnd
self.toEnd = toEnd
self.fromEnd.otherEnd = toEnd
self.toEnd.otherEnd = fromEnd
def add(self, fromObject, toObject):
hub = getService(self, HubIds)
fromUID = hub.getHubId(fromObject)
toUID = hub.getHubId(toObject)
# check bounds
self._check_upperbound(self.fromEnd, fromUID, toUID)
self._check_upperbound(self.toEnd, toUID, fromUID)
# add associations
self.fromEnd.add(fromUID, toUID)
self.toEnd.add(toUID, fromUID)
def delete(self, fromObject, toObject):
hub = getService(self, HubIds)
fromUID = hub.getHubId(fromObject)
toUID = hub.getHubId(toObject)
if self.fromEnd.interface.isImplementedBy(fromObject):
self.fromEnd.delete(fromUID, toUID)
self.toEnd.delete(toUID, fromUID)
elif self.toEnd.interface.isImplementedBy(fromObject):
self.fromEnd.delete(toUID, fromUID)
self.toEnd.delete(fromUID, toUID)
def get(self, obj):
hub = getService(self, HubIds)
uid = hub.getHubId(obj)
if not (self.fromEnd.interface.isImplementedBy(obj) or
self.toEnd.interface.isImplementedBy(obj)):
msg = (u"%s does not implement the interfaces"
u"required by the association %s") % (
obj.__class__.__name__, self.name)
raise AssociationError, msg
# figure which end to query
if self.fromEnd.interface.isImplementedBy(obj):
return self.fromEnd.get(uid)
elif self.toEnd.interface.isImplementedBy(obj):
return self.toEnd.get(uid)
def _check_upperbound(self, end, fromUID, toUID):
keys = end.otherEnd._map.get(toUID, {}).keys()
if end.upperbound and len(keys) == end.upperbound and \
fromUID not in keys:
msg = 'Upper bound exceeded for %s in %s' % (
end.id, self.id)
raise AssociationError, msg
def __getitem__(self, item):
if item == self.fromEnd.name:
return self.fromEnd
elif item == self.toEnd.name:
return self.toEnd
associations/browser/ 0040755 0001751 0000144 00000000000 07743047061 014274 5 ustar roche users associations/browser/referencewidget.py 0100644 0001751 0000144 00000010540 07742333522 020004 0 ustar roche users from zope.app.zapi import getService
from zope.app.services.servicenames import HubIds
from zope.app.browser.form.widget import ListWidget as BaseListWidget
from zope.app.browser.form.widget import RadioWidget as BaseRadioWidget
from zope.app.browser.form.widget import MultiItemsWidget as \
BaseMultiItemsWidget
from zope.app.browser.form.widget import MultiListWidget as \
BaseMultiListWidget
from zope.app.browser.form.widget import MultiCheckBoxWidget as \
BaseMultiCheckBoxWidget
class ReferenceWidgetMixin:
text_attr = '__name__'
def textForValue(self, value):
return getattr(value, self.text_attr)
def renderItems(self, value):
name = self.name
# get items
items = self.context.allowed_values
# check if we want to select first item
if (not value and getattr(self.context, 'firstItem', False)
and len(items) > 0):
value = items[0]
cssClass = self.getValue('cssClass')
rendered_items = []
count = 0
for item_value in items:
item_text = self.textForValue(item_value)
item_value = self._unconvert(item_value)
if item_value == value:
rendered_item = self.renderSelectedItem(count,
item_text,
item_value,
name,
cssClass)
else:
rendered_item = self.renderItem(count,
item_text,
item_value,
name,
cssClass)
rendered_items.append(rendered_item)
count += 1
return rendered_items
def _convert(self, value):
hubid = int(value)
hub = getService(self, HubIds)
return hub.getObject(hubid)
def _unconvert(self, value):
hub = getService(self, HubIds)
hubid = hub.getHubId(value)
return str(hubid)
class ListWidget(ReferenceWidgetMixin, BaseListWidget):
pass
class RadioWidget(ReferenceWidgetMixin, BaseRadioWidget):
pass
class ReferenceSetWidgetMixin:
text_attr = '__name__'
def textForValue(self, value):
return getattr(value, self.text_attr)
def renderItems(self, value):
# need to deal with single item selects
value = removeAllProxies(value)
if not isinstance(value, ListTypes):
value = [value]
name = self.name
items = self.context.allowed_values
cssClass = self.getValue('cssClass')
rendered_items = []
count = 0
for item_value in items:
item_text = self.textForValue(item_value)
item_value = self._unconvert(item_value)
if item_value in value:
rendered_item = self.renderSelectedItem(count,
item_text,
item_value,
name,
cssClass)
else:
rendered_item = self.renderItem(count,
item_text,
item_value,
name,
cssClass)
rendered_items.append(rendered_item)
count += 1
return rendered_items
def _convert(self, value):
# TODO: incomplete
if not value:
return []
if isinstance(value, ListTypes):
return value
return [self._hubIdToObject(value)]
def _hubIdToObject(self, value):
hubid = int(value)
hub = getService(self, HubIds)
return hub.getObject(hubid)
def _unconvert(self, value):
# TODO: incomplete
hub = getService(self, HubIds)
hubid = hub.getHubId(value)
return str(hubid)
class MultiListWidget(ReferenceSetWidgetMixin, BaseMultiListWidget):
pass
class MultiCheckBoxWidget(ReferenceSetWidgetMixin, BaseMultiCheckBoxWidget):
pass
associations/browser/configure.zcml 0100644 0001751 0000144 00000005004 07743047061 017140 0 ustar roche users
associations/browser/__init__.py 0100644 0001751 0000144 00000000000 07735345262 016375 0 ustar roche users associations/browser/association.py 0100644 0001751 0000144 00000001422 07740322312 017146 0 ustar roche users from zope.app.browser.container.adding import Adding
from zope.app.browser.form.editview import EditView
from zope.app.browser.form.add import AddView
from zope.app.form.widget import CustomWidget
from zope.app.browser.form.widget import ObjectWidget
from zopeproducts.associations.interfaces import IAssociation
from zopeproducts.associations.associationend import AssociationEnd
class AssociationAdding(Adding):
menu_id = 'association_menu'
class AssociationView:
__used_for__ = IAssociation
ow = CustomWidget(ObjectWidget, AssociationEnd)
class AssociationEditView(EditView):
__used_for__ = IAssociation
fromEnd_widget = ow
toEnd_widget = ow
class AssociationAddView(AddView):
__used_for__ = IAssociation
fromEnd_widget = ow
toEnd_widget = ow
associations/browser/tests/ 0040755 0001751 0000144 00000000000 07743045642 015441 5 ustar roche users associations/browser/tests/test_referencewidget.py 0100644 0001751 0000144 00000007757 07742060612 022222 0 ustar roche users import unittest
from zope.interface import Interface, implements
from zope.component.view import provideView
from zope.app.browser.form.widget import TextWidget
from zope.schema.interfaces import ITextLine, ValidationError
from zope.publisher.interfaces.browser import IBrowserPresentation
from zope.publisher.browser import TestRequest
from zope.app.zapi import getService
from zope.app.services.servicenames import HubIds
from zopeproducts.associations.referencefield import Reference
from zopeproducts.associations.referencefield import ReferenceQuery
from zopeproducts.associations.browser.referencewidget import \
ListWidget
from zopeproducts.associations.association import Association
from zopeproducts.associations.associationend import AssociationEnd
from zope.schema import TextLine
from zope.app.browser.form.tests.test_browserwidget import BrowserWidgetTest
from zopeproducts.associations.tests.test_referencefield import Student
from zopeproducts.associations.tests.setup import PlacelessSetup
from zopeproducts.associations import AssociationServiceName
s1 = Student()
s1.__name__ = 'Pete'
s2 = Student()
s2.__name__ = 'Susan'
class StudentQuery(ReferenceQuery):
def query(self):
return [s1, s2]
class ITeacher(Interface):
Student = Reference(
association_name='TeacherStudent',
association_end_name='Student',
title=u"Student",
description=u"The student we teach",
required=True)
class IStudent(Interface):
Teacher = Reference(
association_name='TeacherStudent',
association_end_name='Teacher',
title=u"Teacher",
description=u"The teacher that teaches us",
required=True)
class Teacher(object):
implements(ITeacher)
class Student(object):
implements(IStudent)
class ReferenceWidgetTest(PlacelessSetup, BrowserWidgetTest):
def _FieldFactory(self, **kw):
kw.update({'__name__': u'foo',
'association_name': 'TeacherStudent',
'association_end_name': 'Teacher',
'queryFactory': StudentQuery})
return Reference(**kw)
_WidgetFactory = ListWidget
def setUpContent(self, desc=u''):
class ITestContent(Interface):
foo = self._FieldFactory(
title = u"Foo Title",
description = desc)
class TestObject:
implements(ITestContent)
self.content = TestObject()
self.field = ITestContent['foo']
request = TestRequest(HTTP_ACCEPT_LANGUAGE='pl')
request.form[u'field.foo'] = '1'
self._widget = self._WidgetFactory(self.field, request)
def setUp(self):
PlacelessSetup.setUp(self)
self.setUpContent()
provideView(ITextLine, 'edit', IBrowserPresentation, [TextWidget])
associations = getService(self, AssociationServiceName)
# add associations
TeacherEnd = AssociationEnd()
TeacherEnd.name = 'Teacher'
TeacherEnd.interface = ITeacher
TeacherEnd.upperbound = 0
StudentEnd = AssociationEnd()
StudentEnd.name = 'Student'
StudentEnd.interface = IStudent
StudentEnd.upperbound = 0
TeacherStudent = Association('TeacherStudent',
u'Teacher teaches Student', TeacherEnd, StudentEnd)
associations['TeacherStudent'] = TeacherStudent
def testRender(self):
hub = getService(self, HubIds)
check_list = ('select', 'id="field.foo"', 'name="field.foo"',
'size="5"', 'option', 'value="1"', '>Pete<',
'value="2"', '>Susan<', 'selected="selected"')
value = self.field.allowed_values[0]
self._widget.setRenderedValue(value)
r = self._widget()
self.verifyResult(self._widget(), check_list)
def tearDown(self):
PlacelessSetup.tearDown(self)
def test_suite():
return unittest.makeSuite(ReferenceWidgetTest)
if __name__=='__main__':
unittest.main(defaultTest='test_suite')
# vim: set filetype=python ts=4 sw=4 et si
associations/browser/tests/__init__.py 0100644 0001751 0000144 00000000000 07740607147 017536 0 ustar roche users associations/configure.zcml 0100644 0001751 0000144 00000003155 07740321644 015461 0 ustar roche users
associations/referencefield.py 0100644 0001751 0000144 00000012347 07740604010 016117 0 ustar roche users from zodb.btrees.OOBTree import OOSet
from zope.interface import implements, providedBy
from zope.interface.exceptions import DoesNotImplement
from zope.schema import Field
from zope.app.zapi import getService, getUtility
from zopeproducts.associations.interfaces import IReference
from zopeproducts.associations.interfaces import IReferenceSet
from zopeproducts.associations.interfaces import IReferenceQuery
from zopeproducts.associations import AssociationServiceName
class ComputedAttribute(object):
def __init__(self, method):
self.__method = method
def __get__(self, inst, klass):
method = getattr(inst, self.__method)
return method()
class ReferenceQuery(object):
implements(IReferenceQuery)
def __init__(self, context):
self.context = context
def query(self):
raise TypeError("query has not been implemented")
class Reference(Field):
__doc__ = IReference.__doc__
implements(IReference)
association_name = u''
association_end_name = u''
queryFactory = None
allowed_values = ComputedAttribute('queryReferences')
def __init__(self, association_name=u'', association_end_name=u'',
queryFactory=None, **kw):
super(Reference, self).__init__(**kw)
self.association_name = association_name
self.association_end_name = association_end_name
self.queryFactory = queryFactory
def set(self, object, referenced):
associations = getService(self, AssociationServiceName)
association = associations[self.association_name]
# delete association between object and current reference
current_reference = getattr(object, self.__name__, None)
if current_reference:
_dissociate(self, association, object, current_reference)
# add association between object and referenced
_associate(self, association, object, referenced)
super(Reference, self).set(object, referenced)
end = association[self.association_end_name]
_set_back_reference(object, referenced, association, end)
def _validate(self, value):
super(Reference, self)._validate(value)
associations = getService(self, AssociationServiceName)
association = associations[self.association_name]
end = association[self.association_end_name]
if not end.interface.isImplementedBy(value):
raise DoesNotImplement(end.interface)
def queryReferences(self):
q = self.queryFactory(self)
return q.query()
class ReferenceSet(Field):
__doc__ = IReference.__doc__
implements(IReferenceSet)
_type = OOSet
def __init__(self, association_name=u'', association_end_name=u'', **kw):
super(ReferenceSet, self).__init__(**kw)
self.association_name = association_name
self.association_end_name = association_end_name
def set(self, object, referenced):
associations = getService(self, AssociationServiceName)
association = associations[self.association_name]
old_references = getattr(object, self.__name__, [])
for old_reference in old_references:
_dissociate(self, association, object, old_reference)
for reference in referenced:
_associate(self, association, object, reference)
super(ReferenceSet, self).set(object, referenced)
end = association[self.association_end_name]
for reference in referenced:
_set_back_reference(object, reference, association, end)
def _validate(self, value):
super(Reference, self)._validate(value)
associations = getService(self, AssociationServiceName)
association = associations[self.association_name]
end = association[self.association_end_name]
for v in value:
if not end.interface.isImplementedBy(v):
raise DoesNotImplement(end.interface)
def _dissociate(field, association, object, referenced):
""" Delete association between object and referenced """
if field.association_end_name == association.fromEnd.name:
association.delete(referenced, object)
else:
association.delete(object, referenced)
def _associate(field, association, object, referenced):
""" Add association between object and reference """
if field.association_end_name == association.fromEnd.name:
association.add(referenced, object)
else:
association.add(object, referenced)
def _set_back_reference(object, referenced, association, end):
""" set the back reference on the referenced object """
for schema in providedBy(referenced):
for name in schema:
field = schema[name]
if IReference.isImplementedBy(field) and \
field.association_name == association.name and \
field.association_end_name == end.otherEnd.name:
setattr(referenced, field.__name__, object)
elif IReferenceSet.isImplementedBy(field) and \
field.association_name == association.name and \
field.association_end_name != end.otherEnd.name:
ooset = getattr(referenced, field.__name__)
ooset.insert(object)
setattr(referenced, field.__name__, ooset)
associations/tests/ 0040755 0001751 0000144 00000000000 07743045567 013764 5 ustar roche users associations/tests/__init__.py 0100644 0001751 0000144 00000000075 07735613366 016073 0 ustar roche users #
# This file is necessary to make this directory a package.
associations/tests/test_referencefield.py 0100644 0001751 0000144 00000017740 07742045227 020335 0 ustar roche users from unittest import main, makeSuite
from zodb.btrees.OOBTree import OOSet
from zope.interface import Interface, implements
from zope.interface.exceptions import DoesNotImplement
from zope.schema.errornames import RequiredMissing
from zope.schema.tests.test_field import FieldTestBase
from zope.app.zapi import getService
from zope.app.services.servicenames import HubIds
from zopeproducts.associations import AssociationServiceName
from zopeproducts.associations.referencefield import Reference
from zopeproducts.associations.referencefield import ReferenceSet
from zopeproducts.associations.referencefield import ReferenceQuery
from zopeproducts.associations.association import Association
from zopeproducts.associations.associationend import AssociationEnd
from zopeproducts.associations.tests.setup import PlacelessSetup
class ITeacher(Interface):
Student = Reference(
association_name='TeacherStudent',
association_end_name='Student',
title=u"Student",
description=u"The student we teach",
required=True)
Courses = ReferenceSet(
association_name='TeacherCourses',
association_end_name='Course',
title=u"Courses",
description=u"The courses we teach",
required=True)
class IStudent(Interface):
Teacher = Reference(
association_name='TeacherStudent',
association_end_name='Teacher',
title=u"Teacher",
description=u"The teacher that teaches us",
required=True)
class ICourse(Interface):
Teacher = Reference(
association_name='TeacherCourses',
association_end_name='Teacher',
title=u"Teacher",
description=u"The teacher that teaches this course",
required=True)
class Teacher(object):
implements(ITeacher)
class Student(object):
implements(IStudent)
class Course(object):
implements(ICourse)
class LazyStudent(object): pass
class ReferenceTest(PlacelessSetup, FieldTestBase):
"""Test the Container Field."""
_Field_Factory = Reference
def setUp(self):
PlacelessSetup.setUp(self)
associations = getService(self, AssociationServiceName)
# add associations
TeacherEnd = AssociationEnd()
TeacherEnd.name = 'Teacher'
TeacherEnd.interface = ITeacher
TeacherEnd.upperbound = 0
StudentEnd = AssociationEnd()
StudentEnd.name = 'Student'
StudentEnd.interface = IStudent
StudentEnd.upperbound = 0
TeacherStudent = Association('TeacherStudent',
u'Teacher teaches Student', TeacherEnd, StudentEnd)
associations['TeacherStudent'] = TeacherStudent
TeacherEnd = AssociationEnd()
TeacherEnd.name = 'Teacher'
TeacherEnd.interface = ITeacher
TeacherEnd.upperbound = 0
CoursesEnd = AssociationEnd()
CoursesEnd.name = 'Course'
CoursesEnd.interface = ICourse
CoursesEnd.upperbound = 0
TeacherCourses = Association('TeacherCourses',
u'Teacher teaches courses', TeacherEnd, CoursesEnd)
associations['TeacherCourses'] = TeacherCourses
def testValidate(self):
field = self._Field_Factory(title=u'test field', description=u'',
readonly=False, required=False,
association_name='TeacherStudent',
association_end_name='Teacher')
field.validate(Teacher())
self.assertRaises(DoesNotImplement, field.validate, Student())
def testValidateRequired(self):
field = self._Field_Factory(title=u'test field', description=u'',
readonly=False, required=True,
association_name='TeacherStudent',
association_end_name='Student')
field.validate(Student())
self.assertRaisesErrorNames(RequiredMissing, field.validate, None)
def testSetReference(self):
field = self._Field_Factory(__name__='Student',
title=u'test field', description=u'',
readonly=False, required=False,
association_name='TeacherStudent',
association_end_name='Student')
t = Teacher()
s = Student()
field.set(t, s)
assert t.Student == s
assert s.Teacher == t
associations = getService(self, AssociationServiceName)
association = associations['TeacherStudent']
hub = getService(self, HubIds)
t_id = hub.getHubId(t)
s_id = hub.getHubId(s)
assert association.fromEnd._map.has_key(t_id)
assert [k for k in association.fromEnd._map[t_id].keys()] == [s_id]
assert association.toEnd._map.has_key(s_id)
assert [k for k in association.toEnd._map[s_id].keys()] == [t_id]
# The LazyStudent shouldn't have a back reference to the teacher
y = LazyStudent()
field.set(t, y)
assert t.Student == y
assert getattr(y, 'Teacher', None) is None
y_id = hub.getHubId(y)
assert association.fromEnd._map.has_key(t_id)
assert [k for k in association.fromEnd._map[t_id].keys()] == [y_id]
assert association.toEnd._map.has_key(y_id)
assert [k for k in association.toEnd._map[y_id].keys()] == [t_id]
def testSetReferenceSet(self):
field = ReferenceSet(__name__='Courses',
title=u'test field', description=u'',
readonly=False, required=False,
association_name='TeacherCourses',
association_end_name='Course')
a = Teacher(); b = Course(); c = Course()
courses = OOSet([b, c])
field.set(a, courses)
assert a.Courses == courses
assert b.Teacher == a
assert c.Teacher == a
associations = getService(self, AssociationServiceName)
association = associations['TeacherCourses']
hub = getService(self, HubIds)
a_id = hub.getHubId(a)
b_id = hub.getHubId(b)
c_id = hub.getHubId(c)
assert association.fromEnd._map.has_key(a_id)
assert [k for k in association.fromEnd._map[
a_id].keys()].sort() == [b_id, c_id].sort()
assert association.toEnd._map.has_key(b_id)
assert association.toEnd._map.has_key(c_id)
assert [k for k in association.toEnd._map[b_id].keys()] == [a_id]
assert [k for k in association.toEnd._map[c_id].keys()] == [a_id]
d = Course()
courses = OOSet([c, d])
field.set(a, courses)
associations = getService(self, AssociationServiceName)
association = associations['TeacherCourses']
hub = getService(self, HubIds)
d_id = hub.getHubId(d)
assert association.fromEnd._map.has_key(a_id)
assert [k for k in association.fromEnd._map[
a_id].keys()].sort() == [c_id, d_id].sort()
assert association.toEnd._map.has_key(c_id)
assert association.toEnd._map.has_key(d_id)
assert [k for k in association.toEnd._map[c_id].keys()] == [a_id]
assert [k for k in association.toEnd._map[d_id].keys()] == [a_id]
def testQueryReferences(self):
s1 = Student(); s2 = Student()
class StudentQuery(ReferenceQuery):
def query(self):
return [s1, s2]
field = self._Field_Factory(__name__='Student',
title=u'test field', description=u'',
readonly=False, required=False,
association_name='TeacherStudent',
association_end_name='Student',
queryFactory=StudentQuery)
assert field.queryReferences() == [s1, s2]
assert field.allowed_values == [s1, s2]
def test_suite():
return makeSuite(ReferenceTest)
if __name__ == '__main__':
main(defaultTest='test_suite')
associations/tests/setup.py 0100644 0001751 0000144 00000002704 07742044665 015473 0 ustar roche users from UserDict import UserDict
from zope.interface import implements
from zope.component import getServiceManager
from zope.app.tests.placelesssetup import PlacelessSetup as \
BasePlacelessSetup
from zope.app.interfaces.services.hub import IObjectHub
from zope.app.services.servicenames import HubIds
from zope.app.index.tests.test_objectretrievingprocessor import FakeObjectHub
from zopeproducts.associations import AssociationServiceName
from zopeproducts.associations.interfaces import IAssociationService
class FakeAssociationService(UserDict):
implements(IAssociationService)
class RFakeObjectHub(FakeObjectHub):
def getHubId(self, object):
for k,v in self.data.items():
if v == object:
return k
return self.register(object)
class PlacelessSetup(BasePlacelessSetup):
def setUp(self):
BasePlacelessSetup.setUp(self)
# add association service
associations = FakeAssociationService()
service_manager = getServiceManager(self)
service_manager.defineService(AssociationServiceName,
IAssociationService)
service_manager.provideService(AssociationServiceName,
associations)
hub = RFakeObjectHub()
service_manager = getServiceManager(self)
service_manager.defineService(HubIds, IObjectHub)
service_manager.provideService(HubIds, hub)
def tearDown(self):
BasePlacelessSetup.tearDown(self)
associations/README.txt 0100644 0001751 0000144 00000001521 07743046106 014302 0 ustar roche users XXX: Very incomplete.
Associations
------------
Associations makes bidirectional associations between objects easy.
Associations are loosely based on the MOF model for associations. Each
association has two association ends. Each association end has an upper
bound which determines how many objects of that class may be associated
to. An association end only excepts objects of a specific class and
maintains a mapping to objects on the other end.
In this implementation of associations all ends are navigable, which
means it is possible for one end to reach the other end through its
'otherEnd' attribute.
Associations are managed by an AssociationService.
A class do not have to do anything special to participate in an
association. A class can however define references that makes
association ends accessible as attributes on that class.
associations/associationend.py 0100644 0001751 0000144 00000002150 07740321524 016155 0 ustar roche users from persistence import Persistent
from zodb.btrees.IIBTree import IITreeSet, IISet
from zodb.btrees.IOBTree import IOBTree
from zope.interface import implements
from zope.app.catalog.catalog import ResultSet
from zopeproducts.associations.interfaces import IAssociationEnd
class AssociationEnd(Persistent):
implements(IAssociationEnd)
__doc__ = IAssociationEnd.__doc__
def __init__(self):
self._map = IOBTree()
def add(self, fromUID, toUID):
map = self._map
if not map.has_key(fromUID):
map[fromUID] = IITreeSet()
map[fromUID].insert(toUID)
def delete(self, fromUID, toUID):
map = self._map
if map.has_key(fromUID) and toUID in map[fromUID].keys():
map[fromUID].remove(toUID)
if len(map[fromUID]) == 0:
del map[fromUID]
def get(self, uid):
""" Return a LazyMap of associated uids
"""
if uid is None:
r = IITreeSet()
else:
r = self._map.get(uid, IITreeSet())
hub = getService(self, HubIds)
return ResultSet(IISet(r), hub)