本文整理汇总了Python中zope.component.provideHandler函数的典型用法代码示例。如果您正苦于以下问题:Python provideHandler函数的具体用法?Python provideHandler怎么用?Python provideHandler使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了provideHandler函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: main
def main(argv=None):
"""
run this as a cron job and execute all
time based transitions
"""
db = create_engine('postgres://localhost/bungeni', echo=False)
component.provideUtility( db, IDatabaseEngine, 'bungeni-db' )
model.metadata.bind = db
session = Session()
component.provideAdapter(
bungeni.core.workflows.states.WorkflowState,
(bungeni.core.interfaces.IBungeniContent,))
component.provideAdapter(
bungeni.core.workflows.question.QuestionWorkflowAdapter,
(domain.Question,))
component.provideAdapter(
bungeni.core.workflows.states.StateWorkflowInfo,
(domain.Question,))
component.provideHandler(
bungeni.core.workflows.question.workflowTransitionEventDispatcher)
# add autitor for time based transitions
#component.provideAdapter(
# (bungeni.core.interfaces.IAuditable, bungeni.core.interfaces.IQuestion, ),
# (domain.Question, ))
#component.provideAdapter( audit.objectModified,
#(domain.Question, bungeni.core.interfaces.IAuditable, ))
deferAdmissibleQuestions()
session.flush()
session.commit()
开发者ID:BenoitTalbot,项目名称:bungeni-portal,代码行数:33,代码来源:transitioncron.py
示例2: __init__
def __init__(self, connection_id, site_id, user_id=None,
logger=None, handler=None):
h = self.headers = []
h.append('User-Agent: AMQP Consuming Server')
h.append('Accept: text/html,text/plain')
h.append('Host: %s' % socket.gethostname())
self.logger = LogHelper(logger)
self.log_info(("AMQP Consuming Server for connection '%s' started "
"(site '%s' user: '%s')")\
% (connection_id, site_id, user_id))
if handler is None:
# for unit testing
handler = handle
self.zhandler = handler
self.connection_id = connection_id
self.site_id = site_id
self.user_id = user_id
self.consumers = []
provideHandler(self.on_before_broker_connect,
[IBeforeBrokerConnectEvent])
开发者ID:jfroche,项目名称:collective.zamqp,代码行数:25,代码来源:server.py
示例3: setUp
def setUp(cls):
# Register event subscribers
from Products.ExtFile.tests.testCopySupportEvents import ITestItem
from Products.ExtFile.tests.testCopySupportEvents import ITestFolder
from Products.ExtFile.tests.testCopySupportEvents import eventlog
component.provideHandler(eventlog.trace, (ITestItem, IObjectEvent))
component.provideHandler(eventlog.trace, (ITestFolder, IObjectEvent))
开发者ID:clearwired,项目名称:Blend-CMS,代码行数:7,代码来源:testing.py
示例4: test_filtered_handler
def test_filtered_handler(self):
from zope.schema import Text
field = Text(
__name__='testing',
description=u'DESCRIPTION',
default=u'DEFAULT',
required=True,
)
prop = self._makeOne(field=field)
class IFoo(interface.Interface):
"""
"""
@interface.implementer(IFoo)
class Foo(object):
testing = prop
foo = Foo()
logs = []
@component.adapter(IFoo, schema.Text, schema.interfaces.IFieldUpdatedEvent)
def add_field_events(obj, field, event):
logs.append((obj, field, event))
component.provideHandler(add_field_events)
self.assertEqual(len(logs), 0)
foo.testing = u'Bla'
self.assertEqual(len(logs), 1)
event_inst, event_field, _event = logs[0]
self.assertEqual(event_inst, foo)
self.assertEqual(event_field, field)
开发者ID:zopefoundation,项目名称:zope.schemaevent,代码行数:33,代码来源:test_fieldevent.py
示例5: test_update_inherit
def test_update_inherit(self):
""" tests update inherit method
We are owner of the portal, with inheritance we are also owner of
the repo and dossier. If we disable role aquisition on context,
we lose the role 'owner' on the context
"""
# Mock event handlers
provideHandler(factory=self.mock_event.mock_handler, adapts=[ILocalRolesAcquisitionBlocked])
provideHandler(factory=self.mock_event.mock_handler, adapts=[ILocalRolesAcquisitionActivated])
# We disable locale role aquisition on dossier
self.view_dossier.update_inherit(False, reindex=False)
last_event = self.mock_event.last_event()
self.assertTrue(ILocalRolesAcquisitionBlocked.providedBy(last_event))
self.assertEquals(last_event.object, self.dossier)
# we disable it again,it shouldn't fire a event because nothing changed
self.view_dossier.update_inherit(False, reindex=False)
self.assertEquals(last_event, self.mock_event.last_event())
# # We enable locale role aquisition on dossier
self.view_dossier.update_inherit(True, reindex=False)
# and check the fired event
last_event = self.mock_event.last_event()
self.assertTrue(ILocalRolesAcquisitionActivated.providedBy(last_event))
self.assertEquals(last_event.object, self.dossier)
# we disable it again,it shouldn't fire a event because nothing changed
self.view_dossier.update_inherit(True, reindex=False)
self.assertEquals(last_event, self.mock_event.last_event())
开发者ID:pemzurigo,项目名称:opengever.core,代码行数:32,代码来源:test_integration.py
示例6: test_subscriber
def test_subscriber(self):
from zope.schema import Text
field = Text(
__name__='testing',
description=u'DESCRIPTION',
default=u'DEFAULT',
required=True,
)
prop = self._makeOne(field=field)
class Foo(object):
testing = prop
foo = Foo()
logs = []
@component.adapter(schema.interfaces.IFieldUpdatedEvent)
def add_field_event(event):
logs.append(event)
component.provideHandler(add_field_event)
self.assertEqual(len(logs), 0)
foo.testing = u'Bla'
self.assertEqual(len(logs), 1)
event = logs[0]
self.assertEqual(event.field, field)
self.assertEqual(event.inst, foo)
开发者ID:zopefoundation,项目名称:zope.schemaevent,代码行数:28,代码来源:test_fieldevent.py
示例7: setUp
def setUp(self):
self.site = DummySite('site')
self.site._setObject('portal_types', DummyTool())
self.site._setObject('portal_workflow', WorkflowTool())
self._constructDummyWorkflows()
provideHandler(react.object_transitioned)
开发者ID:collective,项目名称:collective.subtractiveworkflow,代码行数:7,代码来源:test_permissions.py
示例8: load_notification_config
def load_notification_config(file_name, domain_class):
"""Loads the notification configuration for each document
"""
notifications_utility = component.getUtility(INotificationsUtility)
path = capi.get_path_for("notifications")
file_path = os.path.join(path, file_name)
notification = capi.schema.validate_file_rng("notifications", file_path)
item_type = file_name.split(".")[0]
notifications_utility.register_item_type(domain_class, item_type)
for notify in notification.iterchildren("notify"):
roles = capi.schema.qualified_roles(notify.get("roles"))
for role in roles:
assert component.queryUtility(IRole, role, None), \
"Notifications configuration error : " \
"Invalid role - %s. file: %s, state : %s" % (
role, file_name)
if notify.get("onstate"):
states = notify.get("onstate").split()
for state in states:
notifications_utility.set_transition_based_notification(
domain_class, state, roles)
elif notify.get("afterstate"):
states = notify.get("afterstate").split()
time = notify.get("time")
for state in states:
notifications_utility.set_time_based_notification(
domain_class, state, roles, time)
else:
raise ValueError("Please specify either onstate or afterstate")
# Register subscriber for domain class
component.provideHandler(queue_notification,
adapts=(domain_class, IWorkflowTransitionEvent))
开发者ID:kohsah,项目名称:bungeni-portal,代码行数:32,代码来源:notifications.py
示例9: __init__
def __init__(self, parent):
MicroViewPlugIn.MicroViewPlugIn.__init__(self, parent)
self._helpLink = "isosurface-isosurface-tool"
# Create a geometry writer
(self._writer, self._ft) = vtkLoadWriters.LoadGeometryWriters()
# create a gui container for this plugin
MicroViewPlugIn.MicroViewPlugIn.CreatePluginDialogFrame(self, parent)
self._color = (1.0, 1.0, 1.0)
self._filename = None
self._app_states = {}
self._current_image_index = None
self._transform = None
# create remaining gui
self.CreateGUI()
# listen to certain zope events
component.provideHandler(self.onModelModifiedEvent)
component.provideHandler(self.onROIEnabledEvent)
component.provideHandler(self.OnImageChangeEvent)
component.provideHandler(self.OnNoImagesLoadedEvent)
component.provideHandler(self.OnCurrentImageClosingEvent)
# Invoke an event to react to currently loaded image
self.InvokeImageLoadedEvent()
开发者ID:andyTsing,项目名称:MicroView,代码行数:32,代码来源:IsoSurfaceDisplay.py
示例10: setUp
def setUp(test):
site = setup.placefulSetUp(True)
site.__name__ = u'portal'
setup.setUpTestAsModule(test, name='zojax.principal.users.TESTS')
setUpContents()
pau = PluggableAuthentication('xyz_')
component.provideUtility(pau, IAuthentication)
component.provideAdapter(SiteManagerAdapter)
component.provideAdapter(FoundPrincipalFactory)
component.provideAdapter(foundsubscriber.getInternalPrincipal)
component.provideHandler(foundsubscriber.foundPrincipalCreated)
component.provideAdapter(adapters.PrincipalLogin)
component.provideAdapter(adapters.PasswordChanger)
component.provideAdapter(adapters.PrincipalMailAddress)
component.provideUtility(adapters.PrincipalByLogin(),
name='zojax.principal.users.principals')
component.provideUtility(adapters.PrincipalByEMail(),
name='zojax.principal.users.principals')
ptool = PasswordTool()
ptool.passwordManager = 'MD5'
component.provideUtility(ptool)
开发者ID:Zojax,项目名称:zojax.principal.users,代码行数:27,代码来源:tests.py
示例11: testObject_pasteUncommitOnException
def testObject_pasteUncommitOnException(self):
"""Ensure that pasted objects aren't commited if an IObjectMovedEvent raises an exception.
See https://dev.plone.org/ticket/9365
"""
# register event handler
provideHandler(failingEventHandler, [IContentish, IObjectMovedEvent])
try:
browser = self.getBrowser()
browser.open(self.folder['source-folder']['doc'].absolute_url())
browser.getLink('Cut').click()
browser.open(self.folder['destination-folder'].absolute_url())
try:
browser.getLink('Paste').click()
except HTTPError:
# a HTTP 500 Server error is currently expected,
# unless we find a better way to abort the
# transaction.
pass
# test if document is not moved
self.assertTrue('doc' in self.folder['source-folder'])
self.assertFalse('doc' in self.folder['destination-folder'])
finally:
# unregister event handler
getGlobalSiteManager().unregisterHandler(
failingEventHandler, [IContentish, IObjectMovedEvent])
开发者ID:erral,项目名称:Products.CMFPlone,代码行数:27,代码来源:testCutPasteSecurity.py
示例12: test_deserializer_notifies_object_modified
def test_deserializer_notifies_object_modified(self):
def handler(obj, event):
obj._handler_called = True
provideHandler(handler, (IDexterityItem, IObjectModifiedEvent,))
self.deserialize(body='{"test_textline_field": "My Item"}')
self.assertTrue(getattr(self.portal.doc1, '_handler_called', False),
'IObjectEditedEvent not notified')
开发者ID:lccruz,项目名称:plone.restapi,代码行数:7,代码来源:test_dxcontent_deserializer.py
示例13: testFolderNotifyAddedEvent
def testFolderNotifyAddedEvent(self):
from zope.component import adapter
from zope.component import provideHandler
from zope.lifecycleevent import ObjectAddedEvent
e = GroupByDateAction()
e.base_folder = "/target"
e.container = "Folder"
class Handler(object):
def __init__(self):
self.invocations = []
self.counter = 0
@adapter(ObjectAddedEvent)
def handler(self, event):
obj = event.object
self.counter += 1
if obj not in self.invocations:
self.invocations.append(obj)
self.handler = Handler()
provideHandler(self.handler.handler)
ex = getMultiAdapter((self.folder, e, DummyEvent(self.folder.d1)), IExecutable)
self.assertTrue(ex())
invocations = self.handler.invocations
self.assertEqual(len(invocations), 3)
self.assertEqual(self.handler.counter, 3)
self.assertEqual(invocations[0].getId(), "2009")
self.assertEqual(invocations[1].getId(), "04")
self.assertEqual(invocations[2].getId(), "22")
开发者ID:collective,项目名称:sc.contentrules.groupbydate,代码行数:32,代码来源:test_action_groupbydate.py
示例14: test_deserializer_notifies_object_edited
def test_deserializer_notifies_object_edited(self):
def handler(obj, event):
obj._handler_called = True
provideHandler(handler, (IBaseObject, IObjectEditedEvent,))
self.doc1.unmarkCreationFlag()
self.deserialize(body='{"testStringField": "Updated"}')
self.assertTrue(getattr(self.doc1, '_handler_called', False),
'IObjectEditedEvent not notified')
开发者ID:casdr,项目名称:plone.restapi,代码行数:8,代码来源:test_atcontent_deserializer.py
示例15: setUp
def setUp(self):
setHooks()
context = config.ConfigurationMachine()
xmlconfig.registerCommonDirectives(context)
for feature in self.features:
context.provideFeature(feature)
self.context = self._load_zcml(context)
provideHandler(events.append, (None,))
开发者ID:Andyvs,项目名称:TrackMonthlyExpenses,代码行数:8,代码来源:testlayer.py
示例16: __init__
def __init__(self, decl_class, name, type, original):
self.decl_class = decl_class
self.name = intern(name)
self._name = intern('_' + name)
self.type = type
self.original = original
component.provideHandler(self._association_changed)
开发者ID:Nyox,项目名称:gaphor,代码行数:8,代码来源:properties.py
示例17: setUp
def setUp(cls):
import Products
zcml._initialized = 0
zcml.load_site()
setHooks()
component.provideHandler(eventlog.trace, (ITestItem, IObjectEvent))
component.provideHandler(eventlog.trace, (IRelationship, IObjectEvent))
开发者ID:SteelHouseLabs,项目名称:zenoss-prodbin,代码行数:8,代码来源:testEvents.py
示例18: test_events
def test_events(self):
events = []
@adapter(IBeforeTransitionEvent)
def _handleBefore(event):
events.append(event)
provideHandler(_handleBefore)
@adapter(IAfterTransitionEvent)
def _handleAfter(event):
events.append(event)
provideHandler(_handleAfter)
wftool = self.site.portal_workflow
wf = self._getDummyWorkflow()
dummy = self.site._setObject( 'dummy', DummyContent() )
wftool.notifyCreated(dummy)
wf.doActionFor(dummy, 'publish', comment='foo', test='bar')
self.assertEquals(4, len(events))
evt = events[0]
self.failUnless(IBeforeTransitionEvent.providedBy(evt))
self.assertEquals(dummy, evt.object)
self.assertEquals('private', evt.old_state.id)
self.assertEquals('private', evt.new_state.id)
self.assertEquals(None, evt.transition)
self.assertEquals({}, evt.status)
self.assertEquals(None, evt.kwargs)
evt = events[1]
self.failUnless(IAfterTransitionEvent.providedBy(evt))
self.assertEquals(dummy, evt.object)
self.assertEquals('private', evt.old_state.id)
self.assertEquals('private', evt.new_state.id)
self.assertEquals(None, evt.transition)
self.assertEquals({}, evt.status)
self.assertEquals(None, evt.kwargs)
evt = events[2]
self.failUnless(IBeforeTransitionEvent.providedBy(evt))
self.assertEquals(dummy, evt.object)
self.assertEquals('private', evt.old_state.id)
self.assertEquals('published', evt.new_state.id)
self.assertEquals('publish', evt.transition.id)
self.assertEquals({'state': 'private', 'comments': ''}, evt.status)
self.assertEquals({'test' : 'bar', 'comment' : 'foo'}, evt.kwargs)
evt = events[3]
self.failUnless(IAfterTransitionEvent.providedBy(evt))
self.assertEquals(dummy, evt.object)
self.assertEquals('private', evt.old_state.id)
self.assertEquals('published', evt.new_state.id)
self.assertEquals('publish', evt.transition.id)
self.assertEquals({'state': 'private', 'comments': ''}, evt.status)
self.assertEquals({'test' : 'bar', 'comment' : 'foo'}, evt.kwargs)
开发者ID:goschtl,项目名称:zope,代码行数:58,代码来源:test_DCWorkflow.py
示例19: listen_for_object_events
def listen_for_object_events(self):
"""listen for all events so that tests can verify they were sent
events are stored in the test case "events" attribute"""
self.events = events = []
def generic_listener(obj, event):
events.append((obj, event))
provideHandler(generic_listener, adapts=[Interface, IObjectEvent])
开发者ID:socialplanning,项目名称:opencore,代码行数:9,代码来源:openplanstestcase.py
示例20: testEVTObjectInitializedEvent
def testEVTObjectInitializedEvent(self):
self.sourceObject = None
@adapter(IObjectInitializedEvent)
def setSourceObject(new_event):
self.sourceObject = new_event.object
component.provideHandler(setSourceObject)
self.factory.createEvent(ICALSTR, self.folder)
self.failUnless(self.folder.listFolderContents()[0] == self.sourceObject)
开发者ID:Goldmund-Wyldebeast-Wunderliebe,项目名称:mailtoplone.base,代码行数:9,代码来源:test_eventfactory.py
注:本文中的zope.component.provideHandler函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论