Add a test that runs with an exec_config and checks an environment va…

…riable is set in the job

src/jobflow_remote/testing/__init__.py

@@ -20,3 +20,10 @@ def write_file(n):
     with open("results.txt", "w") as f:
         f.write(str(n))
     return
+
+
+@job
+def check_env_var() -> str:
+    import os
+
+    return os.environ.get("TESTING_ENV_VAR", "unset")

tests/integration/conftest.py

@@ -229,6 +229,7 @@ def write_tmp_settings(
                 connect_kwargs={"allow_agent": False, "look_for_keys": False},
             ),
         },
+        exec_config={"test": {"export": {"TESTING_ENV_VAR": random_project_name}}},
         runner=dict(
             delay_checkout=1,
             delay_check_run_status=1,

tests/integration/test_slurm.py

@@ -161,3 +161,34 @@ def test_expected_failure(worker, job_controller):
 
     assert job_controller.count_jobs(state=JobState.FAILED) == 2
     assert job_controller.count_flows(state=FlowState.FAILED) == 1
+
+
+@pytest.mark.parametrize(
+    "worker",
+    ["test_local_worker", "test_remote_worker"],
+)
+def test_exec_config(worker, job_controller, random_project_name):
+    """Tests that an environment variable set in the exec config
+    is available to the job.
+
+    """
+    from jobflow import Flow
+
+    from jobflow_remote import submit_flow
+    from jobflow_remote.jobs.runner import Runner
+    from jobflow_remote.jobs.state import FlowState, JobState
+    from jobflow_remote.testing import check_env_var
+
+    job = check_env_var()
+    submit_flow(job, worker=worker, exec_config="test")
+
+    assert job_controller.count_jobs({}) == 1
+    assert len(job_controller.get_jobs({})) == 1
+    assert job_controller.count_flows({}) == 1
+
+    runner = Runner()
+    runner.run(ticks=5)
+
+    job = job_controller.get_jobs({})[0]
+    output = job_controller.jobstore.get_output(uuid=job["uuid"])
+    assert output == random_project_name