Merge 772d07a into 0e3d33c

Author: adi611

.github/workflows/testpsijflux.yml

@@ -0,0 +1,38 @@
+name: PSI/J-Flux
+
+on:
+  push:
+    branches:
+      - master
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    permissions:
+      packages: read
+
+    container:
+      image: fluxrm/flux-sched:focal-v0.28.0
+      options: "--platform=linux/amd64 --user root -it --init"
+
+    steps:
+      - name: Make Space
+        run: |
+          rm -rf /usr/share/dotnet
+          rm -rf /opt/ghc
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Python
+        run: |
+          apt-get update && apt-get install -y python3-venv && apt-get install less
+          export PATH=$PWD/bin:$PATH
+          ln -s /usr/bin/python3 /usr/bin/python
+          python -m pip install --upgrade pip && pip install -e ".[test]" && python -c 'import pydra; print(pydra.__version__)'
+          pip install -e "git+https://github.com/ExaWorks/psij-python.git@main#egg=psij-python"
+      - name: Run pytest
+        run: |
+          export PATH=$PWD/bin:$PATH
+          flux start python -V
+          flux start pytest --psij=flux --color=yes -vs --cov pydra --cov-config .coveragerc --cov-report xml:cov.xml --doctest-modules pydra/

pydra/conftest.py

@@ -11,7 +11,7 @@ def pytest_addoption(parser):
         "--psij",
         action="store",
         help="run with psij subtype plugin",
-        choices=["local", "slurm"],
+        choices=["local", "slurm", "flux"],
     )
 
 

pydra/engine/workers.py

@@ -913,7 +913,7 @@ def __init__(self, subtype, **kwargs):
         self.psij = psij
 
         # Check if the provided subtype is valid
-        valid_subtypes = ["local", "slurm"]
+        valid_subtypes = ["local", "slurm", "flux"]
         if subtype not in valid_subtypes:
             raise ValueError(
                 f"Invalid 'subtype' provided. Available options: {', '.join(valid_subtypes)}"
@@ -992,7 +992,7 @@ async def exec_psij(self, runnable, rerun=False):
             with open(file_path, "wb") as file:
                 pickle.dump(runnable._run, file)
             func_path = absolute_path / "run_pickled.py"
-            spec = self.make_spec("python", [func_path, file_path])
+            spec = self.make_spec("python", [str(func_path), str(file_path)])
         else:  # it could be tuple that includes pickle files with tasks and inputs
             cache_dir = runnable[-1].cache_dir
             file_path_1 = cache_dir / "taskmain.pkl"
@@ -1006,9 +1006,9 @@ async def exec_psij(self, runnable, rerun=False):
             spec = self.make_spec(
                 "python",
                 [
-                    func_path,
-                    file_path_1,
-                    file_path_2,
+                    str(func_path),
+                    str(file_path_1),
+                    str(file_path_2),
                 ],
             )
 
@@ -1044,6 +1044,6 @@ def close(self):
     "sge": SGEWorker,
     **{
         "psij-" + subtype: lambda subtype=subtype: PsijWorker(subtype=subtype)
-        for subtype in ["local", "slurm"]
+        for subtype in ["local", "slurm", "flux"]
     },
 }