Merge pull request #9 from ITISFoundation/add_tools

Add tools subpackage

pyproject.toml

@@ -7,6 +7,7 @@ authors = [
 license={file="LICENSE.txt"}
 readme="README.md"
 dynamic=["version"]
+dependencies=["watchdog"]
 
 [build-system]
 requires=["setuptools", "setuptools_scm"]

src/__init__.py

@@ -1,4 +1,4 @@
-from importlib.metadata import version, PackageNotFoundError
+from importlib.metadata import PackageNotFoundError, version
 
 try:
     __version__ = version("osparc_filecomms")

src/tools.py

@@ -0,0 +1,70 @@
+import json
+import logging
+import pathlib as pl
+import time
+
+from watchdog.events import FileSystemEventHandler
+from watchdog.observers import Observer
+
+DEFAULT_FILE_POLLING_INTERVAL = 5
+
+logging.basicConfig(
+    level=logging.INFO, format="[%(filename)s:%(lineno)d] %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+
+def wait_for_path(file_path):
+    path = pl.Path(file_path).resolve()
+    if path.exists():
+        return str(path)
+
+    # Find the closest existing parent directory
+    watch_dir = path
+    while not watch_dir.exists():
+        watch_dir = watch_dir.parent
+
+    class Handler(FileSystemEventHandler):
+        def __init__(self):
+            self.created = False
+
+        def on_created(self, event):
+            nonlocal path
+            created_path = pl.Path(event.src_path).resolve()
+            if created_path == path:
+                self.created = True
+            elif path.is_relative_to(created_path):
+                # Update path if a parent directory was created
+                path = created_path / path.relative_to(created_path)
+
+    handler = Handler()
+    observer = Observer()
+    observer.schedule(handler, str(watch_dir), recursive=True)
+    observer.start()
+
+    try:
+        while not handler.created:
+            if path.exists():
+                return str(path)
+            observer.join(0.1)
+        return str(path)
+    finally:
+        observer.stop()
+        observer.join()
+
+
+def load_json(
+    file_path, wait=True, file_polling_interval=DEFAULT_FILE_POLLING_INTERVAL
+):
+    if wait:
+        wait_for_path(file_path)
+
+    while True:
+        try:
+            content = json.loads(file_path.read_text())
+            break
+        except json.decoder.JSONDecodeError:
+            logger.info(f"JSON read error, retrying read from {file_path}")
+            time.sleep(file_polling_interval)
+
+    return content

tests/functional/test_tools.py

@@ -0,0 +1,34 @@
+import concurrent.futures as futures
+import json
+import time
+
+from osparc_filecomms import tools
+
+
+def test_read_json(tmp_path):
+    test_json_path = tmp_path / "test.json"
+
+    test_json_data = {"test": "data"}
+    test_json_path.write_text(json.dumps(test_json_data))
+
+    assert tools.load_json(test_json_path, wait=False) == test_json_data
+
+
+def test_wait_for_path(tmp_path):
+    test_path = tmp_path / "test.json"
+
+    executor = futures.ProcessPoolExecutor(max_workers=1)
+
+    wait_future = executor.submit(tools.wait_for_path, test_path)
+
+    assert wait_future.running()
+
+    time.sleep(1)
+
+    test_path.write_text("test")
+
+    time.sleep(1)
+
+    assert wait_future.done()
+
+    executor.shutdown(wait=False, cancel_futures=True)

tests/unit/test_handshake.py

@@ -1,9 +1,8 @@
 import json
-import uuid
 import pathlib as pl
+import uuid
 
 import pytest
-
 from osparc_filecomms import handshakers
 
 test_input_dir = pl.Path("doesntexist_input")

tests/unit/test_tools.py

@@ -0,0 +1,35 @@
+import json
+import pathlib as pl
+import types
+
+import osparc_filecomms.tools as tools
+
+
+def test_import_tools():
+    assert isinstance(tools, types.ModuleType)
+
+
+def test_load_json(mocker):
+    test_json_path = pl.Path("test.json")
+
+    test_json_data = {"test": "data"}
+    mocker.patch.object(
+        pl.Path, "read_text", side_effect=[json.dumps(test_json_data)]
+    )
+
+    assert tools.load_json(test_json_path, wait=False) == test_json_data
+
+
+def test_wait_path(mocker):
+    test_json_path = pl.Path("test.json")
+
+    # Check if it fails is path doesnt exist
+    mocker.patch.object(pl.Path, "exists", side_effect=[False])
+    try:
+        tools.wait_for_path(test_json_path)
+    except StopIteration as error:
+        assert isinstance(error, StopIteration)
+
+    # Check if it succeeds if path does exist
+    mocker.patch.object(pl.Path, "exists", side_effect=[True])
+    tools.wait_for_path(test_json_path)