fix tests

scheduler/dag_tools/__init__.py

@@ -3,6 +3,7 @@
 
 from .build import (
     build_dag,
+    build_dag_cached,
     visualize_dag
 )
 from .node import (

scheduler/dag_tools/build.py

@@ -340,7 +340,7 @@ def validate_dag(dg, tasks_conf):
 def visualize_dag(dg=None, plot=True, write_dot=False, delete_plot=True):
     """For interactive use"""
     if not dg:
-        dg = build_dag()
+        dg = build_dag_cached()
     tmpf = tempfile.mkstemp(suffix='.dot')[1]
     try:
         if plot:

scheduler/dag_tools/traversal.py

@@ -8,7 +8,7 @@
 
 from scheduler.configuration_backend import TasksConfigBaseSequence
 
-from .build import build_dag
+from .build import build_dag_cached
 from .constants import DEPENDENCY_GROUP_DEFAULT_NAME
 from .node import (get_tasks_config, parse_job_id, get_job_id_template)
 from . import log
@@ -24,7 +24,7 @@ def topological_sort(lst):
     dct = defaultdict(list)
     for app_job in lst:
         dct[app_job[0]].append(app_job)
-    for node in nx.topological_sort(build_dag()):
+    for node in nx.topological_sort(build_dag_cached()):
         for app_job2 in dct[node]:
             yield app_job2
 
@@ -40,7 +40,7 @@ def get_parents(app_name, job_id, include_dependency_group=False,
         dependency group
 
     """
-    build_dag()  # run validations
+    build_dag_cached()  # run validations
     if job_id:
         parsed_job_id = parse_job_id(app_name, job_id)
         filter_deps = set(filter_deps)
@@ -285,7 +285,7 @@ def _iter_job_ids(dep_group, group_name, parent_app_name, ld):
 
 
 def get_children(node, job_id, include_dependency_group=True):
-    dg = build_dag()
+    dg = build_dag_cached()
     child_apps = ((k, vv) for k, v in dg.succ[node].items() for vv in v)
     child_apps = list(child_apps)
     for child, group_name in child_apps:

scheduler/tests/test_scheduler.py

@@ -52,7 +52,7 @@ def create_tasks_json(fname_suffix='', inject={}, rename=False):
 
     os.environ['TASKS_JSON'] = f
     try:
-        dt.build_dag.cache.clear()
+        dt.build_dag_cached.cache.clear()
     except AttributeError:
         pass
     return f
@@ -110,16 +110,16 @@ def _inject_into_dag(new_task_dct):
     # verify injection worked
     dg = dt.get_tasks_config()
     assert isinstance(dg, JSONConfig)
-    dag = dt.build_dag()
+    dag = dt.build_dag_cached()
     for k, v in new_task_dct.items():
         assert dg[k] == v, (
             "test code: _inject_into_dag didn't insert the new tasks?")
-        assert dag.node[k] == v, (
+        assert dag.node[k] == dict(v), (
             "test code: _inject_into_dag didn't reset the dag graph")
 
     yield
     os.remove(f)
-    dt.build_dag.cache.clear()
+    dt.build_dag_cached.cache.clear()
 
 
 @with_setup