turtl / turtle / test / test_engine.py

from twisted.internet import defer
from twisted.trial import unittest
from turtle import engine

class TestEngine(unittest.TestCase):

    def _incr(self, result, by=1):
        self.counter += by

    def setUp(self):
        self.counter = 0

    def testThrottler(self):
        N = 13
        thr = engine.ThrottlingDeferred(3*N, N, 200)
        self.assert_(not thr._resetLoop.running)

        controlDeferred = defer.Deferred()
        def helper(self, arg):
            self.arg = arg
            return controlDeferred

        results = []
        uniqueObject = object()
        resultDeferred = thr.run(helper, self=self, arg=uniqueObject)
        self.assertEquals(results, [])
        self.assertEquals(self.arg, uniqueObject)
        self.assertEquals(results.pop(), None)
        self.assertEquals(self.counter, 1)

        self.counter = 0
        for i in range(1, 1 + N):
            self.assertEquals(self.counter, i)

        self.assertEquals(self.counter, N)
        self.assertEquals(thr.points, 0)

        for i in range(1, N):
            self.assertEquals(self.counter, N)

        # Scheduled a prioritized call for last and
        # later we'll check that this runs through
        # before everything else.
        thr.acquire(True).addCallback(self._incr, 10)

        # Even if I release a lot, nothing happens
        # until I have enough points to unlock the
        # situation in the throttler
        for i in thr.waiting:
            self.assertEquals(thr.points, 0)
        self.assertEquals(self.counter, N)
        self.assertEquals(thr.points, 0)

        # At this point I have: 14 waiting calls (N + 1)
        # and 0 points. After I reset I'll have 13 waiting
        # calls and 12 points.
        self.assertEquals(self.counter, N+10)
        self.assertEquals(thr.points, 12)

        # As I said 13 waiting and 12 points:
        for i in range(len(thr.waiting)):

        self.assertEquals(thr.points, 0)
        self.assertEquals(self.counter, N*2+9)

        # Now there's one waiting and 0 points

        # The last reset also called a release
        # 0 waiting and 12 points
        self.assertEquals(self.counter, 2*N+10)
        self.assertEquals(thr.points, N-1)
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.