Fixed the management of environment variables in workflows

demos/hazard/AreaSourceClassicalPSHA/job.toml

@@ -1,5 +1,6 @@
 [workflow]
 description = "AreaSource demo workflow"
+env.OQ_SAMPLE_SITES = 0.1
 
 [Job]
 ini = "job.ini"

openquake/engine/engine.py

@@ -449,34 +449,39 @@ def run_jobs(jobctxs, concurrent_jobs=None, nodes=1, sbatch=False,
 
 class _Workflow:
     # workflow objects are instantiated by the function `read_many`
+    # prefix is empty unless we are in a multi-workflow
     def __init__(self, workflow_toml, defaults, ddic, prefix=''):
         self.workflow_toml = workflow_toml
         self.workflow_dir = os.path.dirname(workflow_toml)
         self.defaults = defaults
         self.description = defaults.pop('description')
         self.checkout = self.defaults.pop('checkout', {})  # repo->branch
         self.may_fail = self.defaults.pop('may_fail', [])
+        self.env = defaults.pop('env', {})
+        self.override = defaults.pop('override', {})
 
         # set the passed environment variables if not already set
-        env = defaults.get('workflow', {}).get('env', {})
-        for k, v in env.items():
+        for k, v in self.env.items():
             if k not in os.environ:
                 os.environ[k] = str(v)
 
-        # override feature for multi-workflows
-        repl = defaults.get('workflow', {}).get('override', {})
-        if repl:
+        # override feature
+        if self.override:
             for _, dic in ddic.items():
                 for name in dic:
-                    if name in repl:
-                        dic[name] = repl[name]
+                    if name in self.override:
+                        dic[name] = self.override[name]
 
         # check the repositories exist
         for value in self.checkout:
             repodir = os.path.join(self.workflow_dir, value)
             if not os.path.exists(repodir):
                 raise FileNotFoundError(repodir)
 
+        self.inis, self.names = self._inis_names(ddic, prefix)
+        check_unique(self.names, workflow_toml)
+
+    def _inis_names(self, ddic, prefix):
         inis = []
         names = []
         self.success = []
@@ -499,10 +504,7 @@ def __init__(self, workflow_toml, defaults, ddic, prefix=''):
             self.fix_paths([dic])
             inis.append(dic)
             names.append(prefix + k)
-
-        check_unique(names, workflow_toml)
-        self.inis = numpy.array(inis)
-        self.names = numpy.array(names)
+        return numpy.array(inis), numpy.array(names)
 
     def fix_paths(self, dicts):
         """

openquake/qa_tests_data/event_based_risk/case_4a/jobs.toml

@@ -1,7 +1,8 @@
 [multi.workflow]
 description = "multi-workflow"
+env.OQ_SAMPLE_ASSETS = 0.1
 may_fail = ["riskJob"]
-replace = {out_file="SES.hdf5", hazard_calculation_id="SES.hdf5"}
+override = {out_file="SES.hdf5", hazard_calculation_id="SES.hdf5"}
 
 [hazard.Job]
 ini = "job_hazard.ini"