source file: /home/buildslave/tahoe/edgy/build/src/allmydata/util/pollmixin.py
file stats: 32 lines, 29 executed: 90.6% covered
coverage versus previous test: 0 lines added, 0 lines removed
1.
2. import time
3. from twisted.internet import task
4.
5. class TimeoutError(Exception):
6. pass
7.
8. class PollComplete(Exception):
9. pass
10.
11. class PollMixin:
12. _poll_should_ignore_these_errors = []
13.
14. def poll(self, check_f, pollinterval=0.01, timeout=1000):
15. # Return a Deferred, then call check_f periodically until it returns
16. # True, at which point the Deferred will fire.. If check_f raises an
17. # exception, the Deferred will errback. If the check_f does not
18. # indicate success within timeout= seconds, the Deferred will
19. # errback. If timeout=None, no timeout will be enforced, and the loop
20. # will poll forever (or really until Trial times out).
21. cutoff = None
22. if timeout is not None:
23. cutoff = time.time() + timeout
24. lc = task.LoopingCall(self._poll, check_f, cutoff)
25. d = lc.start(pollinterval)
26. def _convert_done(f):
27. f.trap(PollComplete)
28. return None
29. d.addErrback(_convert_done)
30. return d
31.
32. def _poll(self, check_f, cutoff):
33. if cutoff is not None and time.time() > cutoff:
34. raise TimeoutError("PollMixin never saw %s return True" % check_f)
35. if check_f():
36. raise PollComplete()
37. # since PollMixin is mostly used for unit tests, quit if we see any
38. # logged errors. This should give us a nice fast failure, instead of
39. # waiting for a timeout. Tests which use flushLoggedErrors() will
40. # need to warn us by putting the error types they'll be ignoring in
41. # self._poll_should_ignore_these_errors
42. if hasattr(self, "_observer") and hasattr(self._observer, "getErrors"):
43. errs = []
44. for e in self._observer.getErrors():
45. if not e.check(*self._poll_should_ignore_these_errors):
46. errs.append(e)
47. if errs:
48. print errs
49. self.fail("Errors snooped, terminating early")
50.