Add polling worker for watcher to probe

[pti/o2.git] / tests / unit / test_watcher.py

diff --git a/tests/unit/test_watcher.py b/tests/unit/test_watcher.py
index ec28519..4cdfb76 100644 (file)
--- a/tests/unit/test_watcher.py
+++ b/tests/unit/test_watcher.py
@@ -12,6 +12,7 @@
 #  See the License for the specific language governing permissions and\r
 #  limitations under the License.\r
 \r
+import time\r
 from datetime import datetime\r
 import json\r
 from typing import List\r
@@ -20,9 +21,11 @@ import pytest
 from o2ims.domain import ocloud\r
 from o2ims import config\r
 import uuid\r
-from o2ims.service.watcher.base import OcloudWather\r
+from o2ims.service.watcher.base import BaseWatcher, OcloudWather\r
 from o2ims.domain import stx_object as ocloudModel\r
 from o2ims.adapter.ocloud_repository import OcloudRepository\r
+from o2ims.service.watcher import worker\r
+from o2ims.service.watcher.executor import start_watchers\r
 \r
 class FakeOcloudClient(BaseClient):\r
     def __init__(self):\r
@@ -69,3 +72,41 @@ def test_probe_new_ocloud():
     ocloudwatcher.probe()\r
     assert len(fakeRepo.oclouds) == 1\r
     assert fakeRepo.oclouds[0].name == "stx1"\r
+\r
+def test_default_worker():\r
+\r
+    class FakeOCloudWatcher(BaseWatcher):\r
+        def __init__(self, client: BaseClient,\r
+                     repo: OcloudRepository) -> None:\r
+            super().__init__(client)\r
+            self.fakeOcloudWatcherCounter = 0\r
+            self._client = client\r
+            self._repo = repo\r
+\r
+        def _targetname(self):\r
+            return "fakeocloudwatcher"\r
+        \r
+        def _probe(self):\r
+            self.fakeOcloudWatcherCounter += 1\r
+            # hacking to stop the blocking sched task\r
+            if self.fakeOcloudWatcherCounter > 2:\r
+                worker.defaultworker.stop()\r
+\r
+\r
+    fakeRepo = FakeOcloudRepo()\r
+    fakeClient = FakeOcloudClient()\r
+    fakewatcher = FakeOCloudWatcher(fakeClient, fakeRepo)\r
+\r
+    worker.defaultworker.set_interval(1)\r
+    worker.defaultworker.add_watcher(fakewatcher)\r
+    assert fakewatcher.fakeOcloudWatcherCounter == 0\r
+\r
+    count1 = fakewatcher.fakeOcloudWatcherCounter\r
+    worker.defaultworker.start()\r
+    time.sleep(20)\r
+    assert fakewatcher.fakeOcloudWatcherCounter > count1\r
+\r
+    # assumed hacking: probe has stopped the sched task\r
+    count3 = fakewatcher.fakeOcloudWatcherCounter\r
+    time.sleep(3)\r
+    assert fakewatcher.fakeOcloudWatcherCounter == count3\r