
from axiom.item import Item
from axiom.attributes import text
from axiom.batch import processor
from axiom.iaxiom import IScheduler

from axiom.test.historic.stubloader import StubbedTest
from axiom.test.historic.stub_processor1to2 import DummyProcessor



class ProcessorUpgradeTest(StubbedTest):
    def setUp(self):
        # Ick, we need to catch the run event of DummyProcessor, and I can't
        # think of another way to do it.
        self.dummyRun = DummyProcessor.run.im_func
        self.dummyRunCalls = []
        def dummyRun(calledOn):
            self.dummyRunCalls.append(calledOn)
        DummyProcessor.run = dummyRun

        return StubbedTest.setUp(self)


    def tearDown(self):
        # Okay this is a pretty irrelevant method on a pretty irrelevant class,
        # but we'll fix it anyway.
        DummyProcessor.run = self.dummyRun

        return StubbedTest.tearDown(self)


    def test_pollingRemoval(self):
        """
        Test that processors lose their idleInterval but none of the rest of
        their stuff, and that they get scheduled by the upgrader so they can
        figure out what state they should be in.
        """
        proc = self.store.findUnique(DummyProcessor)
        self.assertEquals(proc.busyInterval, 100)
        self.failIfEqual(proc.scheduled, None)

        self.assertEquals(self.dummyRunCalls, [proc])


from twisted.internet.defer import Deferred

from axiom.test.historic.stubloader import StubbedTest
from axiom.test.historic.stub_processor1to2 import DummyProcessor

class ProcessorUpgradeTest(StubbedTest):
    def setUp(self):
        # Ick, we need to catch the run event of DummyProcessor, and I can't
        # think of another way to do it.
        self.dummyRun = DummyProcessor.run.im_func
        self.calledBack = Deferred()
        def dummyRun(calledOn):
            self.calledBack.callback(calledOn)
        DummyProcessor.run = dummyRun

        return StubbedTest.setUp(self)


    def tearDown(self):
        # Okay this is a pretty irrelevant method on a pretty irrelevant class,
        # but we'll fix it anyway.
        DummyProcessor.run = self.dummyRun

        return StubbedTest.tearDown(self)


    def test_pollingRemoval(self):
        """
        Test that processors lose their idleInterval but none of the rest of
        their stuff, and that they get scheduled by the upgrader so they can
        figure out what state they should be in.
        """
        proc = self.store.findUnique(DummyProcessor)
        self.assertEquals(proc.busyInterval, 100)
        self.failIfEqual(proc.scheduled, None)
        def assertion(result):
            self.assertEquals(result, proc)
        return self.calledBack.addCallback(assertion)


Fix an intermittent axiom test failure that occurs only in certain environments.
Fix an intermittent axiom test failure that occurs only in certain environments.

Fixes #2857

Author: glyph

Reviewer: exarkun, mithrandi

This change fixes `axiom.test.historic.test_processor1to2.ProcessorUpgradeTest.test_pollingRemoval` to wait until its processor has actually been run before making any assertions.

  
+ from twisted.internet.defer import Deferred
- from axiom.item import Item
- from axiom.attributes import text
- from axiom.batch import processor
- from axiom.iaxiom import IScheduler
  
  from axiom.test.historic.stubloader import StubbedTest
  from axiom.test.historic.stub_processor1to2 import DummyProcessor
- 
- 
  
  class ProcessorUpgradeTest(StubbedTest):
      def setUp(self):
          # Ick, we need to catch the run event of DummyProcessor, and I can't
          # think of another way to do it.
          self.dummyRun = DummyProcessor.run.im_func
-         self.dummyRunCalls = []
+         self.calledBack = Deferred()
          def dummyRun(calledOn):
-             self.dummyRunCalls.append(calledOn)
+             self.calledBack.callback(calledOn)
          DummyProcessor.run = dummyRun
  
          return StubbedTest.setUp(self)
  
  
      def tearDown(self):
          # Okay this is a pretty irrelevant method on a pretty irrelevant class,
          # but we'll fix it anyway.
          DummyProcessor.run = self.dummyRun
  
          return StubbedTest.tearDown(self)
  
  
      def test_pollingRemoval(self):
          """
          Test that processors lose their idleInterval but none of the rest of
          their stuff, and that they get scheduled by the upgrader so they can
          figure out what state they should be in.
          """
          proc = self.store.findUnique(DummyProcessor)
          self.assertEquals(proc.busyInterval, 100)
          self.failIfEqual(proc.scheduled, None)
+         def assertion(result):
+             self.assertEquals(result, proc)
+         return self.calledBack.addCallback(assertion)
  
-         self.assertEquals(self.dummyRunCalls, [proc])
  
Fix an intermittent axiom test failure that occurs only in certain environments.
## Code Before:

from axiom.item import Item
from axiom.attributes import text
from axiom.batch import processor
from axiom.iaxiom import IScheduler

from axiom.test.historic.stubloader import StubbedTest
from axiom.test.historic.stub_processor1to2 import DummyProcessor



class ProcessorUpgradeTest(StubbedTest):
    def setUp(self):
        # Ick, we need to catch the run event of DummyProcessor, and I can't
        # think of another way to do it.
        self.dummyRun = DummyProcessor.run.im_func
        self.dummyRunCalls = []
        def dummyRun(calledOn):
            self.dummyRunCalls.append(calledOn)
        DummyProcessor.run = dummyRun

        return StubbedTest.setUp(self)


    def tearDown(self):
        # Okay this is a pretty irrelevant method on a pretty irrelevant class,
        # but we'll fix it anyway.
        DummyProcessor.run = self.dummyRun

        return StubbedTest.tearDown(self)


    def test_pollingRemoval(self):
        """
        Test that processors lose their idleInterval but none of the rest of
        their stuff, and that they get scheduled by the upgrader so they can
        figure out what state they should be in.
        """
        proc = self.store.findUnique(DummyProcessor)
        self.assertEquals(proc.busyInterval, 100)
        self.failIfEqual(proc.scheduled, None)

        self.assertEquals(self.dummyRunCalls, [proc])

## Instruction:
Fix an intermittent axiom test failure that occurs only in certain environments.
## Code After:

from twisted.internet.defer import Deferred

from axiom.test.historic.stubloader import StubbedTest
from axiom.test.historic.stub_processor1to2 import DummyProcessor

class ProcessorUpgradeTest(StubbedTest):
    def setUp(self):
        # Ick, we need to catch the run event of DummyProcessor, and I can't
        # think of another way to do it.
        self.dummyRun = DummyProcessor.run.im_func
        self.calledBack = Deferred()
        def dummyRun(calledOn):
            self.calledBack.callback(calledOn)
        DummyProcessor.run = dummyRun

        return StubbedTest.setUp(self)


    def tearDown(self):
        # Okay this is a pretty irrelevant method on a pretty irrelevant class,
        # but we'll fix it anyway.
        DummyProcessor.run = self.dummyRun

        return StubbedTest.tearDown(self)


    def test_pollingRemoval(self):
        """
        Test that processors lose their idleInterval but none of the rest of
        their stuff, and that they get scheduled by the upgrader so they can
        figure out what state they should be in.
        """
        proc = self.store.findUnique(DummyProcessor)
        self.assertEquals(proc.busyInterval, 100)
        self.failIfEqual(proc.scheduled, None)
        def assertion(result):
            self.assertEquals(result, proc)
        return self.calledBack.addCallback(assertion)

