Support parameterized job_id in dependencies

Author: lrighi-xtm

python/apsis/cond/dependency.py

@@ -2,7 +2,7 @@
 
 from apsis.lib.json import check_schema
 from apsis.lib.py import format_ctor, iterize
-from apsis.runs import Instance, get_bind_args
+from apsis.runs import Instance, get_bind_args, template_expand
 from apsis.states import State, reachable
 from .base import RunStoreCondition, _bind
 
@@ -90,11 +90,12 @@ def from_jso(cls, jso):
             )
 
     def bind(self, run, jobs):
-        job = jobs[self.job_id]
         bind_args = get_bind_args(run)
+        job_id = template_expand(self.job_id, bind_args)
+        job = jobs[job_id]
         args = _bind(job, self.args, run.inst.args, bind_args)
         return type(self)(
-            self.job_id,
+            job_id,
             args,
             states=self.states,
             exist=self.exist,

test/int/cond/test_cond.py

@@ -148,6 +148,65 @@ def test_to_error(client):
     assert res["state"] == "running"
 
 
+def test_parametrized_dependency(client):
+    """
+    Tests that a dependency with a templated job_id resolves correctly.
+
+    "region dep" depends on "region/{{ region }}" with args date={{ date }}.
+    We schedule a "region dep" run with region=us, which should wait for
+    "region/us" to succeed.
+    """
+    # Schedule a run that depends on region/us.
+    res = client.schedule("region dep", {"region": "us", "date": "2024-01-01"})
+    run_id = res["run_id"]
+
+    # It should be waiting for its dependency.
+    res = client.get_run(run_id)
+    assert res["state"] == "waiting"
+
+    # Satisfy the dependency by scheduling and completing region/us.
+    res = client.schedule("region/us", {"date": "2024-01-01"})
+    dep_run_id = res["run_id"]
+
+    # The dependency run completes immediately (no-op, duration 0).
+    time.sleep(0.5)
+
+    # The dependent run should now have succeeded.
+    res = client.get_run(dep_run_id)
+    assert res["state"] == "success"
+    res = client.get_run(run_id)
+    assert res["state"] == "success"
+
+
+def test_parametrized_dependency_different_regions(client):
+    """
+    Tests that different parameter values resolve to different dependency jobs.
+    """
+    # Schedule runs for two different regions.
+    res_us = client.schedule("region dep", {"region": "us", "date": "2024-01-01"})
+    run_us = res_us["run_id"]
+    res_eu = client.schedule("region dep", {"region": "eu", "date": "2024-01-01"})
+    run_eu = res_eu["run_id"]
+
+    # Both should be waiting.
+    assert client.get_run(run_us)["state"] == "waiting"
+    assert client.get_run(run_eu)["state"] == "waiting"
+
+    # Satisfy only the US dependency.
+    client.schedule("region/us", {"date": "2024-01-01"})
+    time.sleep(0.5)
+
+    # Only the US run should proceed; EU remains waiting.
+    assert client.get_run(run_us)["state"] == "success"
+    assert client.get_run(run_eu)["state"] == "waiting"
+
+    # Now satisfy the EU dependency.
+    client.schedule("region/eu", {"date": "2024-01-01"})
+    time.sleep(0.5)
+
+    assert client.get_run(run_eu)["state"] == "success"
+
+
 def test_thread_cond(inst):
     client = inst.client
 

test/unit/test_cond.py

@@ -1,3 +1,5 @@
+import pytest
+
 from apsis.cond import Condition
 from apsis.cond.dependency import Dependency
 from apsis.cond.skip_duplicate import SkipDuplicate
@@ -10,6 +12,8 @@
 JOBS = {
     "testjob0": Job("testjob0", {"foo"}, (), None),
     "testjob1": Job("testjob1", {"foo", "bar"}, (), None),
+    "dep/us": Job("dep/us", {"date"}, (), None),
+    "dep/eu": Job("dep/eu", {"date"}, (), None),
 }
 
 
@@ -57,6 +61,76 @@ def test_bind3():
     assert bound.states == dep.states
 
 
+def test_bind_parametrized_job_id():
+    # Dependency job_id is a template expanded from the run's args.
+    dep = Dependency("dep/{{ region }}", {})
+    run = Run(Instance("testjob0", {"foo": "x", "region": "us"}))
+    # Use a jobs dict that includes the expanded job_id.
+    jobs = {**JOBS, "dep/us": Job("dep/us", {"foo"}, (), None)}
+    bound = dep.bind(run, jobs)
+
+    assert bound.job_id == "dep/us"
+    assert bound.args == {"foo": "x"}
+
+
+def test_bind_parametrized_job_id_with_args():
+    # Templated job_id combined with templated args.
+    dep = Dependency("dep/{{ region }}", {"date": "{{ date }}"})
+    run = Run(Instance("testjob0", {"foo": "x", "region": "eu", "date": "2024-01-01"}))
+    bound = dep.bind(run, JOBS)
+
+    assert bound.job_id == "dep/eu"
+    assert bound.args == {"date": "2024-01-01"}
+
+
+def test_bind_parametrized_job_id_inherited_args():
+    # Templated job_id with args inherited from the run.
+    dep = Dependency("dep/{{ region }}", {})
+    run = Run(Instance("testjob0", {"foo": "x", "region": "us", "date": "2024-03-15"}))
+    jobs = {**JOBS, "dep/us": Job("dep/us", {"date"}, (), None)}
+    bound = dep.bind(run, jobs)
+
+    assert bound.job_id == "dep/us"
+    assert bound.args == {"date": "2024-03-15"}
+
+
+def test_bind_parametrized_job_id_not_found():
+    # Templated job_id that expands to a nonexistent job.
+    dep = Dependency("dep/{{ region }}", {})
+    run = Run(Instance("testjob0", {"foo": "x", "region": "xx"}))
+    with pytest.raises(LookupError):
+        dep.bind(run, JOBS)
+
+
+def test_bind_parametrized_job_id_missing_param():
+    # Templated job_id referencing a param not in the run's args.
+    dep = Dependency("dep/{{ region }}", {})
+    run = Run(Instance("testjob0", {"foo": "x"}))
+    with pytest.raises(NameError):
+        dep.bind(run, JOBS)
+
+
+def test_bind_literal_job_id_unchanged():
+    # A literal job_id (no template) still works as before.
+    dep = Dependency("testjob0", {"foo": "banana"})
+    run = Run(Instance("testjob1", {"foo": "apple", "bar": "celery"}))
+    bound = dep.bind(run, JOBS)
+
+    assert bound.job_id == "testjob0"
+    assert bound.args == {"foo": "banana"}
+
+
+def test_bind_parametrized_job_id_jso_roundtrip():
+    # Verify that a dependency with a templated job_id survives JSO roundtrip
+    # (the unbound form preserves the template).
+    dep = Dependency("dep/{{ region }}", {"date": "2024-01-01"})
+    jso = dep.to_jso()
+    restored = Condition.from_jso(jso)
+
+    assert restored.job_id == "dep/{{ region }}"
+    assert restored.args == {"date": "2024-01-01"}
+
+
 def test_skip_duplicate_jso():
     cond = SkipDuplicate()
     cond = Condition.from_jso(cond.to_jso())