def _runAllJobs(self):
     """
     Run all outstanding jobs.
     """
     # Run jobs
     jobs = yield JobItem.all(self.transactionUnderTest())
     while jobs:
         yield jobs[0].run()
         yield self.commit()
         jobs = yield JobItem.all(self.transactionUnderTest())
     yield self.commit()

    def action_refreshgroups(self, j):
        txn = self._store.newTransaction()
        yield txn.directoryService().flush()
        work = yield GroupCacherPollingWork.reschedule(txn, 0, force=True)
        jobID = work.jobID
        yield txn.commit()

        if "wait" in j and j["wait"]:
            yield JobItem.waitJobDone(self._store.newTransaction, reactor, 60.0, jobID)
            yield JobItem.waitWorkDone(self._store.newTransaction, reactor, 60.0, (
                GroupRefreshWork, GroupAttendeeReconciliationWork, GroupDelegateChangesWork, GroupShareeReconciliationWork,
            ))

        returnValue(self._ok("ok", "Group refresh scheduled"))

    def test_cascade_delete_cleanup(self):
        """
        Test that when work associated with L{txdav.caldav.datastore.scheduling.work.ScheduleWork}
        is removed with the L{ScheduleWork} item being removed, the associated L{JobItem} runs and
        removes itself and the L{ScheduleWork}.
        """

        ScheduleWorkMixin._queued = 0
        txn = self.transactionUnderTest()
        home = yield self.homeUnderTest(name="user01")
        yield ScheduleOrganizerWork.schedule(
            txn,
            "12345-67890",
            "create",
            home,
            None,
            None,
            self.calendar_new,
            "urn:uuid:user01",
            2,
            True,
        )
        yield self.commit()
        self.assertEqual(ScheduleWorkMixin._queued, 1)

        jobs = yield JobItem.all(self.transactionUnderTest())
        work = yield jobs[0].workItem()
        yield WorkItem.delete(work)
        yield self.commit()

        jobs = yield JobItem.all(self.transactionUnderTest())
        self.assertEqual(len(jobs), 1)
        baseWork = yield ScheduleWork.all(self.transactionUnderTest())
        self.assertEqual(len(baseWork), 1)
        self.assertEqual(baseWork[0].jobID, jobs[0].jobID)

        work = yield jobs[0].workItem()
        self.assertTrue(work is None)
        yield self.commit()

        yield JobItem.waitEmpty(self.storeUnderTest().newTransaction, reactor, 60)

        jobs = yield JobItem.all(self.transactionUnderTest())
        self.assertEqual(len(jobs), 0)
        work = yield ScheduleOrganizerWork.all(self.transactionUnderTest())
        self.assertEqual(len(work), 0)
        baseWork = yield ScheduleWork.all(self.transactionUnderTest())
        self.assertEqual(len(baseWork), 0)

    def test_addressbookObjectRevisions(self):
        """
        Verify that all extra addressbook object revisions are deleted by FindMinValidRevisionWork and RevisionCleanupWork
        """

        # get sync token
        addressbook = yield self.addressbookUnderTest(home="user01", name="addressbook")
        token = yield addressbook.syncToken()

        # make changes
        card1Object = yield self.addressbookObjectUnderTest(self.transactionUnderTest(), name="card1.vcf", addressbook_name="addressbook", home="user01")
        yield card1Object.remove()
        card2Object = yield self.addressbookObjectUnderTest(self.transactionUnderTest(), name="card2.vcf", addressbook_name="addressbook", home="user01")
        yield card2Object.remove()

        # Get object revisions
        rev = schema.ADDRESSBOOK_OBJECT_REVISIONS
        revisionRows = yield Select(
            [rev.REVISION],
            From=rev,
        ).on(self.transactionUnderTest())
        self.assertNotEqual(len(revisionRows), 0)

        # do FindMinValidRevisionWork
        yield self.transactionUnderTest().enqueue(FindMinValidRevisionWork, notBefore=datetime.datetime.utcnow())
        yield self.commit()
        yield JobItem.waitEmpty(self.storeUnderTest().newTransaction, reactor, 60)

        # Get the minimum valid revision and check it
        minValidRevision = yield self.transactionUnderTest().calendarserverValue("MIN-VALID-REVISION")
        self.assertEqual(int(minValidRevision), max([row[0] for row in revisionRows]) + 1)

        # do RevisionCleanupWork
        yield self.transactionUnderTest().enqueue(RevisionCleanupWork, notBefore=datetime.datetime.utcnow())
        yield self.commit()
        yield JobItem.waitEmpty(self.storeUnderTest().newTransaction, reactor, 60)

        # Get group1 object revision
        rev = schema.ADDRESSBOOK_OBJECT_REVISIONS
        revisionRows = yield Select(
            [rev.REVISION],
            From=rev,
        ).on(self.transactionUnderTest())
        self.assertEqual(len(revisionRows), 0)

        # old sync token fails
        addressbook = yield self.addressbookUnderTest(home="user01", name="addressbook")
        yield self.failUnlessFailure(addressbook.resourceNamesSinceToken(token), SyncTokenValidException)

    def test_old(self):
        """
        Verify that old inbox items are removed
        """
        self.patch(config.InboxCleanup, "ItemLifeBeyondEventEndDays", -1)

        # Predate some inbox items
        inbox = yield self.calendarUnderTest(home="user01", name="inbox")
        oldDate = datetime.datetime.utcnow() - datetime.timedelta(days=float(config.InboxCleanup.ItemLifetimeDays), seconds=10)

        itemsToPredate = ["cal2.ics", "cal3.ics"]
        co = schema.CALENDAR_OBJECT
        yield Update(
            {co.CREATED: oldDate},
            Where=co.RESOURCE_NAME.In(Parameter("itemsToPredate", len(itemsToPredate))).And(
                co.CALENDAR_RESOURCE_ID == inbox._resourceID)
        ).on(self.transactionUnderTest(), itemsToPredate=itemsToPredate)

        # do cleanup
        yield self.transactionUnderTest().enqueue(CleanupOneInboxWork, homeID=inbox.ownerHome()._resourceID, notBefore=datetime.datetime.utcnow())
        yield self.commit()
        yield JobItem.waitEmpty(self.storeUnderTest().newTransaction, reactor, 60)

        # check that old items are deleted
        inbox = yield self.calendarUnderTest(home="user01", name="inbox")
        items = yield inbox.objectResources()
        names = [item.name() for item in items]
        self.assertEqual(set(names), set(["cal1.ics"]))

        def stopIt():

            if enableJobProcessing:
                txn = store.newTransaction()
                jobs = yield JobItem.all(txn)
                yield txn.commit()
                yield pool.stopService()
            else:
                jobs = ()

            # active transactions should have been shut down.
            wasBusy = len(cp._busy)
            busyText = repr(cp._busy)
            result = yield cp.stopService()

            if deriveValue(testCase, _SPECIAL_TXN_CLEAN, lambda tc: False):
                if wasBusy:
                    testCase.fail("Outstanding Transactions: " + busyText)
                returnValue(result)

            if len(jobs):
                testCase.fail(
                    "Jobs left in job queue {}: {}".format(testCase, ",".join([job.workType for job in jobs]))
                )

            returnValue(result)

    def test_work(self):

        calendar = """BEGIN:VCALENDAR
PRODID:-//CALENDARSERVER.ORG//NONSGML Version 1//EN
VERSION:2.0
METHOD:REPLY
BEGIN:VEVENT
UID:12345-67890
DTSTAMP:20130208T120000Z
DTSTART:20180601T120000Z
DTEND:20180601T130000Z
ORGANIZER:urn:x-uid:user01
ATTENDEE:mailto:[email protected];PARTSTAT=ACCEPTED
END:VEVENT
END:VCALENDAR
"""
        txn = self.store.newTransaction()
        yield txn.enqueue(
            IMIPReplyWork,
            organizer="urn:x-uid:user01",
            attendee="mailto:[email protected]",
            icalendarText=calendar
        )
        yield txn.commit()
        yield JobItem.waitEmpty(self.store.newTransaction, reactor, 60)

    def test_create(self):
        """
        Test that jobs associated with L{txdav.caldav.datastore.scheduling.work.ScheduleOrganizerSendWork}
        can be created and correctly removed.
        """

        txn = self.transactionUnderTest()
        home = yield self.homeUnderTest(name="user01")
        yield ScheduleOrganizerSendWork.schedule(
            txn,
            "create",
            home,
            None,
            "urn:x-uid:user01",
            "urn:x-uid:user02",
            self.itip_new,
            True,
            1000,
        )

        jobs = yield JobItem.all(self.transactionUnderTest())
        self.assertEqual(len(jobs), 1)

        work = yield jobs[0].workItem()
        yield work.doWork()

        home2 = yield self.calendarUnderTest(home="user02", name="calendar")
        cobjs = yield home2.calendarObjects()
        self.assertEqual(len(cobjs), 1)
        # cal2 = yield cobjs[0].component()

        yield work.delete()
        yield jobs[0].delete()
        yield self.commit()

    def test_processReply(self):
        # Make sure an unknown token in an older email is deleted
        msg = email.message_from_string(self.dataFile('good_reply_past'))
        result = (yield self.receiver.processReply(msg))
        self.assertEquals(result, MailReceiver.UNKNOWN_TOKEN_OLD)

        # Make sure an unknown token is not processed
        msg = email.message_from_string(self.dataFile('good_reply_future'))
        result = (yield self.receiver.processReply(msg))
        self.assertEquals(result, MailReceiver.UNKNOWN_TOKEN)

        # Make sure a known token *is* processed
        txn = self.store.newTransaction()
        yield txn.imipCreateToken(
            "urn:x-uid:5A985493-EE2C-4665-94CF-4DFEA3A89500",
            "mailto:[email protected]",
            "1E71F9C8-AEDA-48EB-98D0-76E898F6BB5C",
            token="d7cdf68d-8b73-4df1-ad3b-f08002fb285f"
        )
        yield txn.commit()

        result = (yield self.receiver.processReply(msg))
        self.assertEquals(result, MailReceiver.INJECTION_SUBMITTED)

        yield JobItem.waitEmpty(self.store.newTransaction, reactor, 60)

    def test_inboxCleanupWorkQueueing(self):
        """
        Verify that InboxCleanupWork queues one CleanupOneInboxBoxWork per home
        """
        self.patch(config.InboxCleanup, "CleanupPeriodDays", -1)

        class FakeCleanupOneInboxWork(WorkItem):
            scheduledHomeIDs = []

            @classmethod
            def reschedule(cls, txn, seconds, homeID):
                cls.scheduledHomeIDs.append(homeID)
                pass

        self.patch(CleanupOneInboxWork, "reschedule", FakeCleanupOneInboxWork.reschedule)

        # do cleanup
        yield InboxCleanupWork.reschedule(self.transactionUnderTest(), 0)
        yield self.commit()
        yield JobItem.waitEmpty(self.storeUnderTest().newTransaction, reactor, 60)

        ch = schema.CALENDAR_HOME
        workRows = yield Select(
            [ch.OWNER_UID],
            From=ch,
            Where=ch.RESOURCE_ID.In(Parameter("scheduledHomeIDs", len(FakeCleanupOneInboxWork.scheduledHomeIDs))),
        ).on(self.transactionUnderTest(), scheduledHomeIDs=FakeCleanupOneInboxWork.scheduledHomeIDs)
        homeUIDs = [workRow[0] for workRow in workRows]
        self.assertEqual(set(homeUIDs), set(['user01', 'user02'])) # two homes

    def test_notificationObjectRevisions(self):
        """
        Verify that all extra notification object revisions are deleted by FindMinValidRevisionWork and RevisionCleanupWork
        """

        # get sync token
        home = yield self.homeUnderTest(name="user01")
        token = yield home.syncToken()

        # make notification changes as side effect of sharing
        yield self._createCalendarShare()

        # Get object revisions
        rev = schema.NOTIFICATION_OBJECT_REVISIONS
        revisionRows = yield Select(
            [rev.REVISION],
            From=rev,
        ).on(self.transactionUnderTest())
        self.assertNotEqual(len(revisionRows), 0)

        # do FindMinValidRevisionWork
        yield self.transactionUnderTest().enqueue(FindMinValidRevisionWork, notBefore=datetime.datetime.utcnow())
        yield self.commit()
        yield JobItem.waitEmpty(self.storeUnderTest().newTransaction, reactor, 60)

        # Get the minimum valid revision and check it
        minValidRevision = yield self.transactionUnderTest().calendarserverValue("MIN-VALID-REVISION")
        self.assertEqual(int(minValidRevision), max([row[0] for row in revisionRows]) + 1)

        # do RevisionCleanupWork
        yield self.transactionUnderTest().enqueue(RevisionCleanupWork, notBefore=datetime.datetime.utcnow())
        yield self.commit()
        yield JobItem.waitEmpty(self.storeUnderTest().newTransaction, reactor, 60)

        # Get group1 object revision
        rev = schema.NOTIFICATION_OBJECT_REVISIONS
        revisionRows = yield Select(
            [rev.REVISION],
            From=rev,
        ).on(self.transactionUnderTest())
        self.assertEqual(len(revisionRows), 0)

        # old sync token fails
        home = yield self.homeUnderTest(name="user01")
        yield self.failUnlessFailure(home.resourceNamesSinceToken(token, "1"), SyncTokenValidException)
        yield self.failUnlessFailure(home.resourceNamesSinceToken(token, "infinity"), SyncTokenValidException)

    def test_basicWork(self):
        """
        Verify that an L{TestWork} item can be enqueued and executed.
        """

        # do FindMinValidRevisionWork
        yield TestWork.schedule(self.storeUnderTest(), 0, 1, 2, 3)

        work = yield TestWork.all(self.transactionUnderTest())
        self.assertEqual(len(work), 1)
        self.assertEqual(work[0].delay, 3)
        job = yield JobItem.querysimple(self.transactionUnderTest(), jobID=work[0].jobID)
        self.assertEqual(len(job), 1)
        self.assertEqual(job[0].priority, 1)
        self.assertEqual(job[0].weight, 2)
        yield self.commit()

        yield JobItem.waitEmpty(self.storeUnderTest().newTransaction, reactor, 60)

    def data_jobcount(self):
        """
        Return a count of job types.

        @return: the JSON result.
        @rtype: L{int}
        """

        return succeed(JobItem.numberOfWorkTypes())

    def action_schedulingdone(self, j):
        """
        Wait for all schedule queue items to complete.
        """
        yield JobItem.waitWorkDone(self._store.newTransaction, reactor, 120.0, (
            ScheduleOrganizerWork, ScheduleOrganizerSendWork, ScheduleReplyWork, ScheduleRefreshWork, ScheduleAutoReplyWork,
        ))

        returnValue(self._ok("ok", "Scheduling done"))

 def _runOneJob(self):
     """
     Run the first outstanding jobs.
     """
     # Run jobs
     jobs = yield JobItem.all(self.transactionUnderTest())
     for job in jobs:
         yield job.run()
         break
     yield self.commit()

    def test_workFailure(self):
        self.sender.smtpSender.shouldSucceed = False

        txn = self.store.newTransaction()
        yield txn.enqueue(
            IMIPInvitationWork,
            fromAddr=ORGANIZER,
            toAddr=ATTENDEE,
            icalendarText=initialInviteText.replace("\n", "\r\n"),
        )
        yield txn.commit()
        yield JobItem.waitEmpty(self.store.newTransaction, reactor, 60)

    def checkTemporaryFailure(self, results):
        """
        Check to see whether whether a temporary failure should be raised as opposed to continuing on with a permanent failure.

        @param results: set of results gathered in L{extractSchedulingResponse}
        @type results: L{list}
        """
        if all([result[1] == iTIPRequestStatus.MESSAGE_PENDING_CODE for result in results]):
            job = yield JobItem.load(self.transaction, self.jobID)
            if job.failed >= config.Scheduling.Options.WorkQueues.MaxTemporaryFailures:
                # Set results to SERVICE_UNAVAILABLE
                for ctr, result in enumerate(results):
                    results[ctr] = (result[0], iTIPRequestStatus.SERVICE_UNAVAILABLE_CODE,)
                returnValue(None)
            else:
                raise JobTemporaryError(config.Scheduling.Options.WorkQueues.TemporaryFailureDelay)

    def test_processReplyMissingAttendee(self):
        msg = email.message_from_string(self.dataFile('reply_missing_attendee'))

        txn = self.store.newTransaction()
        yield txn.imipCreateToken(
            "urn:x-uid:5A985493-EE2C-4665-94CF-4DFEA3A89500",
            "mailto:[email protected]",
            "1E71F9C8-AEDA-48EB-98D0-76E898F6BB5C",
            token="d7cdf68d-8b73-4df1-ad3b-f08002fb285f"
        )
        yield txn.commit()

        result = (yield self.receiver.processReply(msg))
        self.assertEquals(result, MailReceiver.INJECTION_SUBMITTED)

        yield JobItem.waitEmpty(self.store.newTransaction, reactor, 60)

    def action_revisioncleanup(self, j):
        """
        Wait for all schedule queue items to complete.
        """
        from txdav.common.datastore.work.revision_cleanup import _triggerRevisionCleanup
        from txdav.common.datastore.work.revision_cleanup import RevisionCleanupWork

        txn = self._store.newTransaction()
        yield _triggerRevisionCleanup(txn, 60)
        yield txn.commit()

        yield JobItem.waitWorkDone(self._store.newTransaction, reactor, 120.0, (
            RevisionCleanupWork,
        ))

        returnValue(self._ok("ok", "RevisionCleanupWork done"))

    def data_jobs(self):
        """
        Return a summary of the job queue.

        @return: a string containing the JSON result.
        @rtype: L{str}
        """

        if self.factory.store:
            txn = self.factory.store.newTransaction()
            records = (yield JobItem.histogram(txn))
            yield txn.commit()
        else:
            records = {}

        returnValue(records)