def testRenameEvents(self):
        self.assertRaises(KeyError, self.reflecto.manage_renameObjects,
                          ('monty',), ('python',))
        self.assertEqual(len(getEvents()), 0)

        self.assertRaises(KeyError, self.reflecto.manage_renameObjects,
                          ('foo', 'monty'), ('bar', 'python'))
        self.assertEqual(len(getEvents()), 1)
        events = getEvents(IObjectWillBeMovedEvent)
        self.assertEqual(len(events), 1)
        self.assertEqual(events[0].oldName, 'foo')
        self.assertEqual(events[0].newName, 'bar')

        self.assertEqual(self.reflecto.manage_renameObjects(('foo',), ('bar',)),
                         ['foo'])
        events = getEvents()
        self.assertEqual(len(events), 4)
        self.assertTrue(IObjectWillBeMovedEvent.providedBy(events[1]))

        self.assertTrue(IObjectMovedEvent.providedBy(events[2]))
        self.assertEqual(events[2].oldName, 'foo')
        self.assertEqual(events[2].newName, 'bar')

        self.assertTrue(IContainerModifiedEvent.providedBy(events[3]))
        self.assertTrue(events[3].object is self.reflecto)

    def test_createAndAdd(self):

        @implementer(IAdding)
        class Adding(object):

            def __init__(self, test):
                self.test = test

            def add(self, ob):
                self.ob = ob
                self.test.assertEqual(
                    ob.__dict__,
                    {'args': ("bar", "baz"),
                     'kw': {'email': '[email protected]'},
                     '_foo': 'foo',
                    })
                return ob
            def nextURL(self):
                return "."

        adding = Adding(self)
        self._invoke_add()
        (descriminator, callable, args, kw) = self._context.last_action
        factory = AddViewFactory(*args)
        request = TestRequest()
        view = getMultiAdapter((adding, request), name='addthis')

        view.createAndAdd(SampleData.__dict__)

        self.assertEqual(adding.ob.extra1, "extra1")
        self.assertEqual(adding.ob.extra2, "extra2")
        self.assertEqual(adding.ob.name, "foo")
        self.assertEqual(adding.ob.address, "aa")
        self.assertEqual(len(getEvents(IObjectCreatedEvent)), 1)
        self.assertEqual(len(getEvents(IObjectModifiedEvent)), 1)

 def testChangingHEaderMatchFiresEvent(self):
     plugin = self._makeOne()
     self.assertEqual(eventtesting.getEvents(), [])
     plugin.manage_changeProperties(match_roles=[r'REMOTE_ADDR;^10\.0\.(100|101)\.;Authenticated,Member'])
     self.assertEqual(len(eventtesting.getEvents()), 1)
     event = eventtesting.getEvents()[0]
     self.assertEqual(event.__class__.__name__, 'ConfigurationChangedEvent')
     self.assertEqual(event.object, plugin)
     self.assertEqual(event.object.match_roles, (r'REMOTE_ADDR;^10\.0\.(100|101)\.;Authenticated,Member',))

 def testChangingIPRolesFiresEvent(self):
     plugin = self._makeOne()
     self.assertEqual(eventtesting.getEvents(), [])
     plugin.manage_changeProperties(ip_roles=['127.0.0.0/24:Authenticated'])
     self.assertEqual(len(eventtesting.getEvents()), 1)
     event = eventtesting.getEvents()[0]
     self.assertEqual(event.__class__.__name__, 'ConfigurationChangedEvent')
     self.assertEqual(event.object, plugin)
     self.assertEqual(event.object.ip_roles, ('127.0.0.0/24:Authenticated',))

    def test_remove_group(self):
        browser = self._connect_as_manager()
        self._create_group(browser)
        browser.getControl(name="remove-folder-members").click()
        self.assertNotIn("remove-folder-members", browser.contents)
        self.assertIn("Folder members", browser.contents)
        events = eventtesting.getEvents(IGroupRemoved)
#        self.assertEqual(len(events), 1)
        self.assertTrue(len(events) > 1)

 def test_payment_event(self):
     """Check that the MollieIdealPaymentEvent was fired."""
     request = TestRequest(
         form=dict(transaction_id=self.transaction_id))
     payment_view = getMultiAdapter((self.foo, request),
                                    name='report_multiple_payment_status')
     payment_view()
     payment_events = [event for event in eventtesting.getEvents()
                       if IMollieIdealPaymentEvent.providedBy(event)]
     self.assertTrue(len(payment_events) > 0)

 def test_poll(self):
     cacher = self.layer.cacher
     config = self._get_poller_config('3', 'c') # poll time 0, should loop once
     config.update({'exit_': threading.Event()})
     
     # poll once
     t1 = threading.Thread(target=cacher.poll, kwargs=(config))
     t1.start()
     t1.join(.1)
     self.assertFalse(t1.is_alive())
     self.assertEquals(len(getEvents(
                         ICompletedCachableSourcePoll)), 1)
     
     # poll again
     t2 = threading.Thread(target=cacher.poll, kwargs=(config))
     t2.start()
     t2.join(.1)
     self.assertFalse(t2.is_alive())
     self.assertEquals(len(getEvents(
                         ICompletedCachableSourcePoll)), 2)

    def testDeleteEvents(self):
        self.assertRaises(KeyError, self.reflecto.manage_delObjects, ('monty',))
        self.assertEqual(len(getEvents()), 0)

        self.assertRaises(KeyError, self.reflecto.manage_delObjects,
                          ('foo', 'monty',))
        self.assertEqual(len(getEvents()), 1)
        events = getEvents(IObjectWillBeRemovedEvent)
        self.assertEqual(len(events), 1)
        self.assertEqual(events[0].oldName, 'foo')

        self.assertEqual(self.reflecto.manage_delObjects(('foo',)), None)
        events = getEvents()
        self.assertEqual(len(events), 4)
        self.assertTrue(IObjectWillBeRemovedEvent.providedBy(events[1]))

        self.assertTrue(IObjectRemovedEvent.providedBy(events[2]))
        self.assertEqual(events[2].oldName, 'foo')

        self.assertTrue(IContainerModifiedEvent.providedBy(events[3]))
        self.assertTrue(events[3].object is self.reflecto)

    def test_register_available_tiles_record(self):
        registry = getUtility(IRegistry)
        record = 'collective.cover.controlpanel.ICoverSettings.available_tiles'

        eventtesting.setUp()

        # calling the handler here should have no effect as we are running the
        # latest profile version
        eventtesting.clearEvents()
        register_available_tiles_record(self.portal)
        events = eventtesting.getEvents(IRecordAddedEvent)
        self.assertEqual(len(events), 0)

        # now we delete the record and rerun the handler to verify the record
        # was added
        del registry.records[record]
        eventtesting.clearEvents()
        register_available_tiles_record(self.portal)
        events = eventtesting.getEvents(IRecordAddedEvent)
        self.assertNotEqual(len(events), 0)
        self.assertIn(record, registry.records)
        eventtesting.clearEvents()

 def test_payment_event_content(self):
     """Check that the MollieIdealPaymentEvent has the content we need."""
     request = TestRequest(
         form=dict(transaction_id=self.transaction_id))
     payment_view = getMultiAdapter((self.foo, request),
                                    name='report_multiple_payment_status')
     payment_view()
     payment_events = [event for event in eventtesting.getEvents()
                       if IMollieIdealPaymentEvent.providedBy(event)]
     event = payment_events[0]
     self.assertEqual(event.context, self.foo)
     self.assertEqual(event.request, request)
     self.assertEqual(event.transaction_id, self.transaction_id)

    def test_register_styles_record(self):
        registry = getUtility(IRegistry)
        record = 'collective.cover.controlpanel.ICoverSettings.styles'

        eventtesting.setUp()

        # just delete the existing record and rerun the handler to verify it
        # was added again
        del registry.records[record]
        eventtesting.clearEvents()
        register_styles_record(self.portal)
        events = eventtesting.getEvents(IRecordAddedEvent)
        self.assertNotEqual(len(events), 0)
        self.assertIn(record, registry.records)
        eventtesting.clearEvents()

    def test_accept_token_view(self):
        token_id, token_url = self.util.generate_new_token(usage_limit=1)
        self.assertTrue(self.util.valid(token_id))

        view = getMultiAdapter((self.portal, self.request), name=u"accept-token")
        url = view.publishTraverse(self.request, token_id)()
        self.assertEqual(url, self.portal.absolute_url())
        self.assertFalse(self.util.valid(token_id))
        events = eventtesting.getEvents(ITokenAccepted)
        self.assertEqual(len(events), 1)
        self.assertEqual(events[0].token_id, token_id)

        token_id, token_url = self.util.generate_new_token(redirect_path="a/b/c")
        url = view.publishTraverse(self.request, token_id)()
        self.assertEqual(url, self.portal.absolute_url() + "/a/b/c")

 def test_update_via_adapter(self):
     f = Foo()
     request = TestRequest()
     v = BarV(f, request)
     # check adapter
     self.assertEqual(f.foo, u'Foo foo')
     a = IBar(f)
     self.assertEqual(a.bar, u'Foo foo')
     # update
     request.form[Update] = ''
     request.form['field.bar'] = u'r bar'
     message = v.update()
     self.assertTrue(message.startswith('Updated '), message)
     self.assertEqual(a.bar, u'r bar')
     # wrong update
     self.assertFalse(getEvents())

    def testCopyEvents(self):
        cp = self.reflecto.manage_copyObjects(('foo',))
        self.reflecto.manage_pasteObjects(cp)

        events = getEvents()
        self.assertEqual(len(events), 3)

        self.assertTrue(IObjectCopiedEvent.providedBy(events[0]))
        self.assertEqual(events[0].original.getId(), 'foo')
        self.assertEqual(events[0].object.getId(), 'copy_of_foo')

        self.assertTrue(IObjectClonedEvent.providedBy(events[1]))
        self.assertEqual(events[1].object.getId(), 'copy_of_foo')

        self.assertTrue(IContainerModifiedEvent.providedBy(events[2]))
        self.assertTrue(events[2].object is self.reflecto)

    def test_automatic_purging(self):
        # z3c.caching must purge objects automatically on modified,
        # moved, renamed or removed; testing the whole enchilada is out
        # of the scope of this package
        from z3c.caching.interfaces import IPurgeEvent
        from zope.component import eventtesting
        from zope.event import notify
        from zope.lifecycleevent import ObjectModifiedEvent
        eventtesting.setUp()

        # modifying the poll must fire a Purge event
        notify(ObjectModifiedEvent(self.poll))
        events = eventtesting.getEvents()
        # XXX: Purge event is fired more than once
        #      https://community.plone.org/t/6411?u=hvelarde
        self.assertEventFiredOnce(IPurgeEvent, events)

 def assertImportZip(self, filename, imported, replace=False, update=False,
                     ignore_top_level=False):
     """Import a ZIP file.
     """
     clearEvents()
     request = TestRequest()
     importer = ZipImporter(self.root, request, {
             'replace_content': replace,
             'update_content': update,
             'ignore_top_level_content': ignore_top_level})
     with self.layer.open_fixture(filename) as source:
         importer.importStream(source)
     self.assertItemsEqual(
         map(lambda event:  '/'.join(event.object.getPhysicalPath()),
             getEvents(IContentImported)),
         imported)
     return importer

def test_order_events():
    oc = OrderedBTreeContainer()

    oc['foo'] = 'bar'
    oc['baz'] = 'quux'
    oc['zork'] = 'grue'

    assert oc.keys() == ['foo', 'baz', 'zork']

    clearEvents()
    oc.updateOrder(['baz', 'foo', 'zork'])
    assert oc.keys() == ['baz', 'foo', 'zork']

    events = getEvents()
    assert [event.__class__.__name__ for event in events] == (
        ['ContainerModifiedEvent'])

    assert IObjectModifiedEvent.providedBy(events[0])

    def test_disable_disabled_member(self, DT):
        """Test disabling an already disabled member. This should not happen
        often, but it still could."""
        DT.return_value = DateTime('2012/01/01')

        # first disable a member
        self.test_disable_enabled_member()

        # now let's say a month goes by and we get another notification from
        # IPN that this member is disabled
        DT.return_value = DateTime('2012/02/01')
        self.ipn.disable_member(
            email='[email protected]',
            product_id='1',
            trans_type='CANCEL',
        )

        # test event emitted
        events = list(set(eventtesting.getEvents(IMemberDisabledEvent)))
        self.assertEquals(len(events), 2)
        self.assertEquals(events[0].username, '[email protected]')
        self.assertEquals(events[1].username, '[email protected]')

        # test member history
        self.assert_member_history(
            username='[email protected]',
            history=[
                '2012/01/01 00:00:00|disable_member|1|CANCEL|'
                'removed from groups: ipn_1, ',
                '2012/02/01 00:00:00|disable_member|1|CANCEL|',
            ]
        )

        # test log output
        for record in self.log.records:
            self.assertNotIn(
                "Adding member '[email protected]' to Disabled group.",
                record.getMessage(),
            )

            self.assertNotIn(
                "Revoking member '[email protected]' the Member role.",
                record.getMessage(),
            )

    def testCutEvents(self):
        cp = self.reflecto.manage_cutObjects(('foo',))
        self.reflecto['subdir'].manage_pasteObjects(cp)

        events = getEvents()
        self.assertEqual(len(events), 4)

        self.assertTrue(IObjectWillBeMovedEvent.providedBy(events[0]))
        self.assertEqual(events[0].oldName, 'foo')
        self.assertEqual(events[0].oldParent, self.reflecto)
        self.assertEqual(events[0].newName, 'foo')
        self.assertEqual(events[0].newParent, self.reflecto['subdir'])

        self.assertTrue(IObjectMovedEvent.providedBy(events[1]))
        self.assertEqual(events[1].object.getId(), 'foo')

        self.assertTrue(IContainerModifiedEvent.providedBy(events[2]))
        self.assertEqual(events[2].object, self.reflecto)
        self.assertTrue(IContainerModifiedEvent.providedBy(events[3]))
        self.assertEqual(events[3].object, self.reflecto['subdir'])

    def test_delete_tile_persistent_data(self):
        eventtesting.clearEvents()
        # First, let's store some data on the tile
        data_mgr = ITileDataManager(self.tile)
        data_mgr.set({'test': 'data'})

        # We see that the data persists
        self.assertIn('test', data_mgr.get())
        self.assertEqual(data_mgr.get()['test'], 'data')

        # Call the delete method
        self.tile.delete()

        # Now we should not see the stored data anymore
        self.assertNotIn('test', data_mgr.get())

        events = eventtesting.getEvents()

        # Finally, test that ObjectModifiedEvent was fired for the cover
        self.assertEqual(events[0].object, self.cover)