[Lib] Add dependency and output path retrieval library functions

src/conductor/lib/__init__.py

@@ -0,0 +1,7 @@
+"""
+This module is Conductor's user library. It contains various utilities that
+can be useful in Python scripts that run as Conductor tasks.
+"""
+
+# Path-related utilities.
+from .path import *

src/conductor/lib/path.py

@@ -0,0 +1,38 @@
+import os
+import pathlib
+from typing import List
+
+from conductor.config import (
+    DEPS_ENV_VARIABLE_NAME,
+    DEPS_ENV_PATH_SEPARATOR,
+    OUTPUT_ENV_VARIABLE_NAME,
+)
+
+
+def get_deps_paths() -> List[pathlib.Path]:
+    """
+    Returns a list of the output paths of this task's dependencies.
+    """
+    if DEPS_ENV_VARIABLE_NAME not in os.environ:
+        raise RuntimeError(
+            "The {} environment variable was not set. Make sure your code is "
+            "being executed by Conductor.".format(DEPS_ENV_VARIABLE_NAME)
+        )
+    return list(
+        map(
+            pathlib.Path,
+            os.environ[DEPS_ENV_VARIABLE_NAME].split(DEPS_ENV_PATH_SEPARATOR),
+        )
+    )
+
+
+def get_output_path() -> pathlib.Path:
+    """
+    Returns the path where this task's outputs should be stored.
+    """
+    if OUTPUT_ENV_VARIABLE_NAME not in os.environ:
+        raise RuntimeError(
+            "The {} environment variable was not set. Make sure your code is "
+            "being executed by Conductor.".format(OUTPUT_ENV_VARIABLE_NAME)
+        )
+    return pathlib.Path(os.environ[OUTPUT_ENV_VARIABLE_NAME])

tests/conductor_runner.py

@@ -100,4 +100,5 @@ def count_task_outputs(in_dir: pathlib.Path):
         _TESTS_DIR, "fixture-projects", "output-recording"
     ).resolve(),
     "partial-success": pathlib.Path(_TESTS_DIR, "fixture-projects", "partial-success"),
+    "lib-test": pathlib.Path(_TESTS_DIR, "fixture-projects", "lib-test"),
 }

tests/fixture-projects/lib-test/path/COND

@@ -0,0 +1,23 @@
+run_command(
+  name="one",
+  run="exit 0",
+)
+
+run_command(
+  name="two",
+  run="exit 0",
+)
+
+run_command(
+  name="deps",
+  run="python3 deps.py",
+  deps=[
+    ":one",
+    ":two",
+  ],
+)
+
+run_command(
+  name="output_path",
+  run="python3 output_path.py",
+)

tests/fixture-projects/lib-test/path/deps.py

@@ -0,0 +1,14 @@
+import conductor.lib as cond
+from conductor.config import TASK_OUTPUT_DIR_SUFFIX
+
+
+def main():
+    expected_deps = {"one" + TASK_OUTPUT_DIR_SUFFIX, "two" + TASK_OUTPUT_DIR_SUFFIX}
+    deps = cond.get_deps_paths()
+    assert len(deps) > 0
+    for dep_path in deps:
+        assert dep_path.name in expected_deps
+
+
+if __name__ == "__main__":
+    main()

tests/fixture-projects/lib-test/path/output_path.py

@@ -0,0 +1,12 @@
+import conductor.lib as cond
+from conductor.config import TASK_OUTPUT_DIR_SUFFIX
+
+
+def main():
+    output_dir = cond.get_output_path()
+    assert output_dir.exists()
+    assert output_dir.name == "output_path" + TASK_OUTPUT_DIR_SUFFIX
+
+
+if __name__ == "__main__":
+    main()

tests/lib_path_test.py

@@ -0,0 +1,29 @@
+import pathlib
+import pytest
+
+import conductor.lib as condlib
+from .conductor_runner import ConductorRunner, FIXTURE_TEMPLATES
+
+
+def test_get_output_path_non_cond():
+    with pytest.raises(RuntimeError):
+        _ = condlib.get_output_path()
+
+
+def test_get_deps_paths_non_cond():
+    with pytest.raises(RuntimeError):
+        _ = condlib.get_deps_paths()
+
+
+def test_get_deps_paths(tmp_path: pathlib.Path):
+    cond = ConductorRunner.from_template(tmp_path, FIXTURE_TEMPLATES["lib-test"])
+    result = cond.run("//path:deps")
+    # The deps.py script in lib-test/path/ does the correctness assertions.
+    assert result.returncode == 0
+
+
+def test_get_output_path(tmp_path: pathlib.Path):
+    cond = ConductorRunner.from_template(tmp_path, FIXTURE_TEMPLATES["lib-test"])
+    result = cond.run("//path:output_path")
+    # The output_path.py script in lib-test/path/ does the correctness assertions.
+    assert result.returncode == 0