Add workflow integration test

phys2denoise/tests/conftest.py

@@ -3,7 +3,7 @@
 import pytest
 from _pytest.logging import LogCaptureFixture
 from loguru import logger
-from physutils import Physio
+from physutils import Physio, io
 
 import phys2denoise.tasks as tasks
 

phys2denoise/tests/test_tasks.py

@@ -97,7 +97,6 @@ def test_export_metrics(fake_physio_with_metrics):
 
     # Check if the output directory was created
     assert os.path.exists(outdir)
-    assert task.result().output.out is None
 
     # Check if the files were created
     assert os.path.exists(

phys2denoise/tests/test_tasks_integration.py

@@ -1,4 +1,5 @@
 """Tests for phys2denoise.tasks and their integration."""
+import logging
 import os
 
 import nest_asyncio
@@ -10,23 +11,33 @@
 
 import phys2denoise.tasks as tasks
 
+from .utils import create_fake_phys
+
 nest_asyncio.apply()
 
+LGR = logging.getLogger(__name__)
+LGR.setLevel(logging.DEBUG)
+
 
 def test_integration(fake_phys):
     """Test the integration of phys2denoise tasks."""
     file_dir = os.path.dirname(os.path.abspath(__file__))
     export_dir = os.path.join(file_dir, "test_output_data/integration")
+    create_fake_phys()
+
+    physio_file = os.path.abspath("phys2denoise/tests/data/fake_phys.phys")
 
     wf = Workflow(
         name="metrics_wf",
         input_spec=["phys", "fs"],
-        phys="phys2denoise/tests/data/ECG.csv",
-        fs=62.5,
+        phys=physio_file,
     )
     wf.add(
         transform_to_physio(
-            name="transform_to_physio", phys=wf.lzin.phys, fs=wf.lzin.fs, mode="physio"
+            name="transform_to_physio",
+            input_file=wf.lzin.phys,
+            fs=wf.lzin.fs,
+            mode="physio",
         )
     )
     wf.add(
@@ -60,6 +71,6 @@ def test_integration(fake_phys):
 
     with Submitter(plugin="cf") as sub:
         sub(wf)
-    wf().result()
+    wf()
 
     assert os.path.exists(export_dir)

phys2denoise/tests/utils.py

@@ -0,0 +1,36 @@
+import logging
+import os
+
+import numpy as np
+import peakdet
+from physutils import Physio, io
+
+LGR = logging.getLogger(__name__)
+LGR.setLevel(logging.DEBUG)
+
+
+def create_fake_phys():
+    f = 0.3
+    fs = 62.5  # sampling rate
+    t = 300
+    samples = np.arange(t * fs) / fs
+    noise = np.random.normal(0, 0.5, len(samples))
+    fake_phys = 10 * np.sin(2 * np.pi * f * samples) + noise
+
+    phys = peakdet.Physio(fake_phys, fs=62.5)
+    phys = peakdet.operations.filter_physio(phys, cutoffs=3, method="lowpass")
+    phys = peakdet.operations.peakfind_physio(phys)
+
+    # TODO: Change to a simpler call once physutils are
+    # integrated to peakdet/prep4phys
+    physio_obj = Physio(phys.data, phys.fs)
+    physio_obj._metadata["peaks"] = phys.peaks
+    physio_obj._metadata["troughs"] = phys.troughs
+
+    LGR.debug(f"Current working directory: {os.getcwd()}")
+    save_path = os.path.abspath("phys2denoise/tests/data/fake_phys.phys")
+    LGR.info(f"save_path: {save_path}")
+
+    io.save_physio(save_path, physio_obj)
+
+    return physio_obj