test(decorator): synchronized

tests/test_synchronized.py

+#
+# Aura Engine (https://gitlab.servus.at/aura/engine)
+#
+# Copyright (C) 2017-now() - The Aura Engine Team.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+import threading
+import time
+import unittest
+
+from aura_engine.base.lang import synchronized
+
+
+class MessageHolder:
+    message = None
+
+    def __init__(self):
+        self.message = "no message"
+
+    def set_message(self, message: str, sleep_time: float):
+        print(f"Updating message to '{message}' in {sleep_time} seconds")
+        time.sleep(sleep_time)
+        self.message = message
+        # print("Message updated")
+
+    @synchronized
+    def set_synced_message(self, message: str, sleep_time: float):
+        print(f"Synced: Updating message to '{message}' in {sleep_time} seconds")
+        time.sleep(sleep_time)
+        self.message = message
+
+    def get_message(self):
+        return self.message
+
+
+class TestSynchronized(unittest.TestCase):
+    """
+    Testing the Logger.
+    """
+
+    mh = None
+
+    def setUp(self):
+        self.mh = MessageHolder()
+
+    def test_not_synchronized(self):
+
+        # Functions to update the message via threads
+        def fast_cat_1():
+            self.mh.set_message("fast cat 1", 0.05)
+
+        def sleepy_dog_1():
+            self.mh.set_message("sleepy dog 1", 0.5)
+
+        def fast_cat_2():
+            self.mh.set_message("fast cat 2", 0.1)
+
+        def sleepy_dog_2():
+            self.mh.set_message("sleepy dog 2", 1)
+
+        # CASE#0: Get initial message
+        msg = self.mh.get_message()
+        print(msg)
+        self.assertEqual("no message", msg)
+
+        # Start threads
+        thread1 = threading.Thread(target=fast_cat_1)
+        thread2 = threading.Thread(target=sleepy_dog_1)
+        thread3 = threading.Thread(target=sleepy_dog_2)
+        thread4 = threading.Thread(target=fast_cat_2)
+        thread1.start()
+        thread2.start()
+        thread3.start()
+        thread4.start()
+
+        # CASE#1: First thread quickly updates the message
+        time.sleep(0.08)
+        msg = self.mh.get_message()
+        print(msg)
+        self.assertEqual("fast cat 1", msg)
+
+        # CASE#2: Last thread has overtaken the two slow ones
+        time.sleep(0.12)
+        msg = self.mh.get_message()
+        print(msg)
+        self.assertEqual("fast cat 2", msg)
+
+        # # CASE#3: Slow one arrived
+        time.sleep(0.5)
+        msg = self.mh.get_message()
+        print(msg)
+        self.assertEqual("sleepy dog 1", msg)
+
+        # # CASE#3: The other slow one arrived
+        time.sleep(0.5)
+        msg = self.mh.get_message()
+        print(msg)
+        self.assertEqual("sleepy dog 2", msg)
+
+        thread1.join()
+        thread2.join()
+        thread3.join()
+        thread4.join()
+
+    def test_synchronized(self):
+
+        # Functions to update the message via threads
+        def fast_cat_1():
+            self.mh.set_synced_message("fast cat 1", 0.05)
+
+        def sleepy_dog_1():
+            self.mh.set_synced_message("sleepy dog 1", 0.5)
+
+        def fast_cat_2():
+            self.mh.set_synced_message("fast cat 2", 0.1)
+
+        def sleepy_dog_2():
+            self.mh.set_synced_message("sleepy dog 2", 1)
+
+        # CASE#0: Get initial message
+        msg = self.mh.get_message()
+        print(msg)
+        self.assertEqual("no message", msg)
+
+        # Start threads
+        thread1 = threading.Thread(target=fast_cat_1)
+        thread2 = threading.Thread(target=sleepy_dog_1)
+        thread3 = threading.Thread(target=sleepy_dog_2)
+        thread4 = threading.Thread(target=fast_cat_2)
+        thread1.start()
+        time.sleep(0.01)
+        thread2.start()
+        time.sleep(0.01)
+        thread3.start()
+        time.sleep(0.01)
+        thread4.start()
+
+        # CASE#1: First thread quickly updates the message
+        time.sleep(0.1)
+        msg = self.mh.get_message()
+        print(msg)
+        self.assertEqual("fast cat 1", msg)
+
+        # # CASE#2: Any fast cat has to wait for this dog
+        time.sleep(0.5)
+        msg = self.mh.get_message()
+        print(msg)
+        self.assertEqual("sleepy dog 1", msg)
+
+        # # CASE#3: And for the other dog too
+        time.sleep(1)
+        msg = self.mh.get_message()
+        print(msg)
+        self.assertEqual("sleepy dog 2", msg)
+
+        # # CASE#3: Finally it's the fast cats turn
+        time.sleep(0.2)
+        msg = self.mh.get_message()
+        print(msg)
+        self.assertEqual("fast cat 2", msg)
+
+        thread1.join()
+        thread2.join()
+        thread3.join()
+        thread4.join()
+
+
+if __name__ == "__main__":
+    unittest.main()