Unit tests for octoprint.util.CountedEvent

tests/util/test_counted_event.py

@@ -0,0 +1,144 @@
+# coding=utf-8
+from __future__ import absolute_import
+
+__author__ = "Gina Häußge <osd@foosel.net>"
+__license__ = 'GNU Affero General Public License http://www.gnu.org/licenses/agpl.html'
+__copyright__ = "Copyright (C) 2015 The OctoPrint Project - Released under terms of the AGPLv3 License"
+
+import unittest
+import time
+import threading
+
+from octoprint.util import CountedEvent
+
+class CountedEventTest(unittest.TestCase):
+
+	def test_set_once(self):
+		"""The counter should go from 0 to 1."""
+
+		event = CountedEvent()
+
+		self.assertEquals(0, event._counter)
+		self.assertFalse(event._event.is_set())
+
+		event.set()
+
+		self.assertEquals(1, event._counter)
+		self.assertTrue(event._event.is_set())
+
+	def test_set_more_than_max(self):
+		"""The counter should never rise above max."""
+
+		event = CountedEvent(max=1)
+
+		self.assertEquals(0, event._counter)
+		self.assertFalse(event._event.is_set())
+
+		event.set()
+
+		self.assertEquals(1, event._counter)
+		self.assertTrue(event._event.is_set())
+
+		event.set()
+
+		self.assertEquals(1, event._counter)
+		self.assertTrue(event._event.is_set())
+
+	def test_clear_once(self):
+		"""The counter should to from 1 to 0."""
+
+		event = CountedEvent(1)
+
+		self.assertEquals(1, event._counter)
+		self.assertTrue(event._event.is_set())
+
+		event.clear()
+
+		self.assertEquals(0, event._counter)
+		self.assertFalse(event._event.is_set())
+
+	def test_clear_all(self):
+		"""The counter should go from 10 to 0."""
+
+		event = CountedEvent(10)
+
+		self.assertEquals(10, event._counter)
+		self.assertTrue(event._event.is_set())
+
+		event.clear(completely=True)
+
+		self.assertEquals(0, event._counter)
+		self.assertFalse(event._event.is_set())
+
+	def test_clear_more_than_available(self):
+		"""The counter should never sink below 0."""
+
+		event = CountedEvent(1)
+
+		self.assertEquals(1, event._counter)
+		self.assertTrue(event._event.is_set())
+
+		event.clear()
+
+		self.assertEquals(0, event._counter)
+		self.assertFalse(event._event.is_set())
+
+		event.clear()
+
+		self.assertEquals(0, event._counter)
+		self.assertFalse(event._event.is_set())
+
+	def test_blocked(self):
+		"""Blocked should only be true if the counter is 0."""
+
+		event = CountedEvent(0)
+
+		self.assertTrue(event.blocked())
+
+		event.set()
+		self.assertFalse(event.blocked())
+
+		event.clear()
+		self.assertTrue(event.blocked())
+
+	def test_wait_immediately(self):
+		"""Unblocked wait should immediately return."""
+
+		event = CountedEvent(1)
+
+		start = time.time()
+		event.wait(timeout=2)
+		duration = time.time() - start
+
+		self.assertLess(duration, 1)
+
+	def test_wait_blocking(self):
+		"""Set should immediately have blocked wait return."""
+
+		event = CountedEvent(0)
+
+		def set_event():
+			time.sleep(1)
+			event.set()
+
+		thread = threading.Thread(target=set_event)
+		thread.daemon = True
+		thread.start()
+
+		start = time.time()
+		event.wait(timeout=2)
+		duration = time.time() - start
+
+		self.assertLess(duration, 2)
+
+	def test_wait_timeout(self):
+		"""Blocked should only wait until timeout."""
+
+		event = CountedEvent(0)
+
+		start = time.time()
+		event.wait(timeout=2)
+		duration = time.time() - start
+
+		self.assertGreaterEqual(duration, 2)
+		self.assertLess(duration, 3)