Test correct increase in dep_fail count (#4003)

(These counts are not generally tested, but this PR is not introducing a
general tests for them; instead it is about testing dependency handling
behaviour more - based on regressions encountered during development of
adjacent PRs)

# Changed Behaviour

none

## Type of change
- Code maintenance/cleanup

Author: benclifford

parsl/tests/test_python_apps/test_depfail_propagation.py

@@ -1,5 +1,7 @@
+import parsl
 from parsl import python_app
 from parsl.dataflow.errors import DependencyError
+from parsl.dataflow.states import States
 
 
 @python_app
@@ -14,6 +16,7 @@ def depends(parent):
 
 def test_depfail_once():
     """Test the simplest dependency failure case"""
+    start_dep_fail_count = parsl.dfk().task_state_counts[States.dep_fail]
     f1 = fails()
     f2 = depends(f1)
 
@@ -25,9 +28,12 @@ def test_depfail_once():
     # in the DependencyError message
     assert ("task " + str(f1.task_record['id'])) in str(f2.exception())
 
+    assert parsl.dfk().task_state_counts[States.dep_fail] == start_dep_fail_count + 1
+
 
 def test_depfail_chain():
     """Test that dependency failures chain"""
+    start_dep_fail_count = parsl.dfk().task_state_counts[States.dep_fail]
     f1 = fails()
     f2 = depends(f1)
     f3 = depends(f2)
@@ -39,11 +45,13 @@ def test_depfail_chain():
     assert isinstance(f3.exception(), DependencyError)
     assert isinstance(f4.exception(), DependencyError)
 
+    assert parsl.dfk().task_state_counts[States.dep_fail] == start_dep_fail_count + 3
+
 
 def test_depfail_branches():
     """Test that dependency failures propagate in the
     presence of multiple downstream tasks."""
-
+    start_dep_fail_count = parsl.dfk().task_state_counts[States.dep_fail]
     f1 = fails()
     f2 = depends(f1)
     f3 = depends(f1)
@@ -52,3 +60,5 @@ def test_depfail_branches():
     assert not isinstance(f1.exception(), DependencyError)
     assert isinstance(f2.exception(), DependencyError)
     assert isinstance(f3.exception(), DependencyError)
+
+    assert parsl.dfk().task_state_counts[States.dep_fail] == start_dep_fail_count + 2