Blackened tests

Author: jmafoster1

tests/data_collection_tests/test_observational_data_collector.py

@@ -9,15 +9,15 @@
 
 
 class TestObservationalDataCollector(unittest.TestCase):
-
     def setUp(self) -> None:
         temp_dir_path = create_temp_dir_if_non_existent()
         self.dag_dot_path = os.path.join(temp_dir_path, "dag.dot")
         self.observational_df_path = os.path.join(temp_dir_path, "observational_data.csv")
         # Y = 3*X1 + X2*X3 + 10
         self.observational_df = pd.DataFrame({"X1": [1, 2, 3, 4], "X2": [5, 6, 7, 8], "X3": [10, 20, 30, 40]})
         self.observational_df["Y"] = self.observational_df.apply(
-                lambda row: (3 * row.X1) + (row.X2 * row.X3) + 10, axis=1)
+            lambda row: (3 * row.X1) + (row.X2 * row.X3) + 10, axis=1
+        )
         self.observational_df.to_csv(self.observational_df_path)
         self.X1 = Input("X1", int, uniform(1, 4))
         self.X2 = Input("X2", int, rv_discrete(values=([7], [1])))
@@ -45,12 +45,13 @@ def test_data_constraints(self):
 
     def test_meta_population(self):
         def populate_m(data):
-            data['M'] = data['X1'] * 2
+            data["M"] = data["X1"] * 2
+
         meta = Meta("M", int, populate_m)
         scenario = Scenario({self.X1, meta})
         observational_data_collector = ObservationalDataCollector(scenario, self.observational_df_path)
         data = observational_data_collector.collect_data()
-        assert all((m == 2*x1 for x1, m in zip(data['X1'], data['M'])))
+        assert all((m == 2 * x1 for x1, m in zip(data["X1"], data["M"])))
 
     def tearDown(self) -> None:
         remove_temp_dir_if_existent()

tests/generation_tests/test_abstract_test_case.py

@@ -8,18 +8,21 @@
 from tests.test_helpers import create_temp_dir_if_non_existent, remove_temp_dir_if_existent
 from causal_testing.testing.causal_test_outcome import Positive
 
+
 class TestAbstractTestCase(unittest.TestCase):
     """
-        Class to test abstract test cases.
+    Class to test abstract test cases.
     """
+
     def setUp(self) -> None:
         temp_dir_path = create_temp_dir_if_non_existent()
         self.dag_dot_path = os.path.join(temp_dir_path, "dag.dot")
         self.observational_df_path = os.path.join(temp_dir_path, "observational_data.csv")
         # Y = 3*X1 + X2*X3 + 10
         self.observational_df = pd.DataFrame({"X1": [1, 2, 3, 4], "X2": [5, 6, 7, 8], "X3": [10, 20, 30, 40]})
-        self.observational_df["Y"] = self.observational_df.apply(lambda row: (3 * row.X1) + (row.X2 * row.X3) + 10,
-                                                                 axis=1)
+        self.observational_df["Y"] = self.observational_df.apply(
+            lambda row: (3 * row.X1) + (row.X2 * row.X3) + 10, axis=1
+        )
         self.observational_df.to_csv(self.observational_df_path)
         self.X1 = Input("X1", float, uniform(1, 4))
         self.X2 = Input("X2", int, rv_discrete(values=([7], [1])))
@@ -51,9 +54,9 @@ def test_str(self):
             expected_causal_effect={self.Y: Positive()},
             effect_modifiers=None,
         )
-        assert str(abstract) == \
-        "When we apply intervention {X1' > X1}, the effect on Output: Y::int should be Positive", \
-        f"Unexpected string {str(abstract)}"
+        assert (
+            str(abstract) == "When we apply intervention {X1' > X1}, the effect on Output: Y::int should be Positive"
+        ), f"Unexpected string {str(abstract)}"
 
     def test_datapath(self):
         scenario = Scenario({self.X1, self.X2, self.X3, self.X4})
@@ -109,7 +112,6 @@ def test_generate_concrete_test_cases_rct(self):
         assert len(concrete_tests) == 2, "Expected 2 concrete tests"
         assert len(runs) == 4, "Expected 4 runs"
 
-
     def test_infeasible_constraints(self):
         scenario = Scenario({self.X1, self.X2, self.X3, self.X4}, [self.X1.z3 > 2])
         scenario.setup_treatment_variables()
@@ -125,10 +127,9 @@ def test_infeasible_constraints(self):
 
         with self.assertWarns(Warning):
             concrete_tests, runs = abstract.generate_concrete_tests(4, rct=True, target_ks_score=0.1, hard_max=HARD_MAX)
-        self.assertTrue(all((x > 2 for x in runs['X1'])))
+        self.assertTrue(all((x > 2 for x in runs["X1"])))
         self.assertEqual(len(concrete_tests), HARD_MAX * NUM_STRATA)
 
-
     def test_feasible_constraints(self):
         scenario = Scenario({self.X1, self.X2, self.X3, self.X4})
         scenario.setup_treatment_variables()
@@ -142,7 +143,6 @@ def test_feasible_constraints(self):
         concrete_tests, _ = abstract.generate_concrete_tests(4, rct=True, target_ks_score=0.1, hard_max=1000)
         assert len(concrete_tests) < 1000
 
-
     def tearDown(self) -> None:
         remove_temp_dir_if_existent()
 

tests/json_front_tests/test_json_class.py

@@ -93,12 +93,10 @@ def test_generate_tests_from_json(self):
         self.json_class.test_plan = example_test
         effects = {"NoEffect": NoEffect()}
         mutates = {
-            "Increase": lambda x: self.json_class.modelling_scenario.treatment_variables[x].z3 >
-                                  self.json_class.modelling_scenario.variables[x].z3
-        }
-        estimators = {
-            "LinearRegressionEstimator": LinearRegressionEstimator
+            "Increase": lambda x: self.json_class.modelling_scenario.treatment_variables[x].z3
+            > self.json_class.modelling_scenario.variables[x].z3
         }
+        estimators = {"LinearRegressionEstimator": LinearRegressionEstimator}
 
         with self.assertLogs() as captured:
             self.json_class.generate_tests(effects, mutates, estimators, False)
@@ -108,9 +106,8 @@ def test_generate_tests_from_json(self):
 
     def tearDown(self) -> None:
         pass
-        #remove_temp_dir_if_existent()
+        # remove_temp_dir_if_existent()
 
 
 def populate_example(*args, **kwargs):
     pass
-

tests/specification_tests/test_causal_dag.py

@@ -50,9 +50,12 @@ def test_valid_causal_dag(self):
         """Test whether the Causal DAG is valid."""
         causal_dag = CausalDAG(self.dag_dot_path)
         print(causal_dag)
-        assert list(causal_dag.graph.nodes) == ["A", "B", "C", "D"] and list(
-            causal_dag.graph.edges
-        ) == [("A", "B"), ("B", "C"), ("D", "A"), ("D", "C")]
+        assert list(causal_dag.graph.nodes) == ["A", "B", "C", "D"] and list(causal_dag.graph.edges) == [
+            ("A", "B"),
+            ("B", "C"),
+            ("D", "A"),
+            ("D", "C"),
+        ]
 
     def test_invalid_causal_dag(self):
         """Test whether a cycle-containing directed graph is an invalid causal DAG."""
@@ -96,9 +99,7 @@ class TestDAGDirectEffectIdentification(unittest.TestCase):
     def setUp(self) -> None:
         temp_dir_path = create_temp_dir_if_non_existent()
         self.dag_dot_path = os.path.join(temp_dir_path, "dag.dot")
-        dag_dot = (
-            """digraph G { X1->X2;X2->V;X2->D1;X2->D2;D1->Y;D1->D2;Y->D3;Z->X2;Z->Y;}"""
-        )
+        dag_dot = """digraph G { X1->X2;X2->V;X2->D1;X2->D2;D1->Y;D1->D2;Y->D3;Z->X2;Z->Y;}"""
         f = open(self.dag_dot_path, "w")
         f.write(dag_dot)
         f.close()
@@ -122,9 +123,7 @@ class TestDAGIdentification(unittest.TestCase):
     def setUp(self) -> None:
         temp_dir_path = create_temp_dir_if_non_existent()
         self.dag_dot_path = os.path.join(temp_dir_path, "dag.dot")
-        dag_dot = (
-            """digraph G { X1->X2;X2->V;X2->D1;X2->D2;D1->Y;D1->D2;Y->D3;Z->X2;Z->Y;}"""
-        )
+        dag_dot = """digraph G { X1->X2;X2->V;X2->D1;X2->D2;D1->Y;D1->D2;Y->D3;Z->X2;Z->Y;}"""
         f = open(self.dag_dot_path, "w")
         f.write(dag_dot)
         f.close()
@@ -137,13 +136,10 @@ def test_get_indirect_graph(self):
         self.assertEqual(list(indirect_graph.graph.edges), original_edges)
         self.assertEqual(indirect_graph.graph.nodes, causal_dag.graph.nodes)
 
-
     def test_proper_backdoor_graph(self):
         """Test whether converting a Causal DAG to a proper back-door graph works correctly."""
         causal_dag = CausalDAG(self.dag_dot_path)
-        proper_backdoor_graph = causal_dag.get_proper_backdoor_graph(
-            ["X1", "X2"], ["Y"]
-        )
+        proper_backdoor_graph = causal_dag.get_proper_backdoor_graph(["X1", "X2"], ["Y"])
         self.assertEqual(
             list(proper_backdoor_graph.graph.edges),
             [
@@ -163,11 +159,7 @@ def test_constructive_backdoor_criterion_should_hold(self):
         causal_dag = CausalDAG(self.dag_dot_path)
         xs, ys, zs = ["X1", "X2"], ["Y"], ["Z"]
         proper_backdoor_graph = causal_dag.get_proper_backdoor_graph(xs, ys)
-        self.assertTrue(
-            causal_dag.constructive_backdoor_criterion(
-                proper_backdoor_graph, xs, ys, zs
-            )
-        )
+        self.assertTrue(causal_dag.constructive_backdoor_criterion(proper_backdoor_graph, xs, ys, zs))
 
     def test_constructive_backdoor_criterion_should_not_hold_not_d_separator_in_proper_backdoor_graph(
         self,
@@ -176,11 +168,7 @@ def test_constructive_backdoor_criterion_should_not_hold_not_d_separator_in_prop
         causal_dag = CausalDAG(self.dag_dot_path)
         xs, ys, zs = ["X1", "X2"], ["Y"], ["V"]
         proper_backdoor_graph = causal_dag.get_proper_backdoor_graph(xs, ys)
-        self.assertFalse(
-            causal_dag.constructive_backdoor_criterion(
-                proper_backdoor_graph, xs, ys, zs
-            )
-        )
+        self.assertFalse(causal_dag.constructive_backdoor_criterion(proper_backdoor_graph, xs, ys, zs))
 
     def test_constructive_backdoor_criterion_should_not_hold_descendent_of_proper_causal_path(
         self,
@@ -190,11 +178,7 @@ def test_constructive_backdoor_criterion_should_not_hold_descendent_of_proper_ca
         causal_dag = CausalDAG(self.dag_dot_path)
         xs, ys, zs = ["X1", "X2"], ["Y"], ["D1"]
         proper_backdoor_graph = causal_dag.get_proper_backdoor_graph(xs, ys)
-        self.assertFalse(
-            causal_dag.constructive_backdoor_criterion(
-                proper_backdoor_graph, xs, ys, zs
-            )
-        )
+        self.assertFalse(causal_dag.constructive_backdoor_criterion(proper_backdoor_graph, xs, ys, zs))
 
     def test_is_min_adjustment_for_min_adjustment(self):
         """Test whether is_min_adjustment can correctly test whether the minimum adjustment set is minimal."""
@@ -262,9 +246,7 @@ def test_enumerate_minimal_adjustment_sets_multiple(self):
         )
         xs, ys = ["X1", "X2"], ["Y"]
         adjustment_sets = causal_dag.enumerate_minimal_adjustment_sets(xs, ys)
-        set_of_adjustment_sets = set(
-            frozenset(min_separator) for min_separator in adjustment_sets
-        )
+        set_of_adjustment_sets = set(frozenset(min_separator) for min_separator in adjustment_sets)
         self.assertEqual(
             {frozenset({"Z1"}), frozenset({"Z2"}), frozenset({"Z3"})},
             set_of_adjustment_sets,
@@ -291,9 +273,7 @@ def test_enumerate_minimal_adjustment_sets_two_adjustments(self):
         )
         xs, ys = ["X1", "X2"], ["Y"]
         adjustment_sets = causal_dag.enumerate_minimal_adjustment_sets(xs, ys)
-        set_of_adjustment_sets = set(
-            frozenset(min_separator) for min_separator in adjustment_sets
-        )
+        set_of_adjustment_sets = set(frozenset(min_separator) for min_separator in adjustment_sets)
         self.assertEqual(
             {frozenset({"Z1", "Z4"}), frozenset({"Z2", "Z4"}), frozenset({"Z3", "Z4"})},
             set_of_adjustment_sets,
@@ -304,20 +284,20 @@ def test_dag_with_non_character_nodes(self):
         causal_dag = CausalDAG()
         causal_dag.graph.add_edges_from(
             [
-                ('va', 'ba'),
-                ('ba', 'ia'),
-                ('ba', 'da'),
-                ('ba', 'ra'),
-                ('la', 'va'),
-                ('la', 'aa'),
-                ('aa', 'ia'),
-                ('aa', 'da'),
-                ('aa', 'ra'),
+                ("va", "ba"),
+                ("ba", "ia"),
+                ("ba", "da"),
+                ("ba", "ra"),
+                ("la", "va"),
+                ("la", "aa"),
+                ("aa", "ia"),
+                ("aa", "da"),
+                ("aa", "ra"),
             ]
         )
-        xs, ys = ['ba'], ['da']
+        xs, ys = ["ba"], ["da"]
         adjustment_sets = causal_dag.enumerate_minimal_adjustment_sets(xs, ys)
-        self.assertEqual(adjustment_sets, [{'aa'}, {'la'}, {'va'}])
+        self.assertEqual(adjustment_sets, [{"aa"}, {"la"}, {"va"}])
 
     def tearDown(self) -> None:
         remove_temp_dir_if_existent()
@@ -385,9 +365,7 @@ class TestUndirectedGraphAlgorithms(unittest.TestCase):
 
     def setUp(self) -> None:
         self.graph = nx.Graph()
-        self.graph.add_edges_from(
-            [("a", 2), ("a", 3), (2, 4), (3, 5), (3, 4), (4, "b"), (5, "b")]
-        )
+        self.graph.add_edges_from([("a", 2), ("a", 3), (2, 4), (3, 5), (3, 4), (4, "b"), (5, "b")])
         self.treatment_node = "a"
         self.outcome_node = "b"
         self.treatment_node_set = {"a"}
@@ -396,9 +374,7 @@ def setUp(self) -> None:
 
     def test_close_separator(self):
         """Test whether close_separator correctly identifies the close separator of {2,3} in the undirected graph."""
-        result = close_separator(
-            self.graph, self.treatment_node, self.outcome_node, self.treatment_node_set
-        )
+        result = close_separator(self.graph, self.treatment_node, self.outcome_node, self.treatment_node_set)
         self.assertEqual({2, 3}, result)
 
     def test_list_all_min_sep(self):
@@ -414,12 +390,8 @@ def test_list_all_min_sep(self):
         )
 
         # Convert list of sets to set of frozen sets for comparison
-        min_separators = set(
-            frozenset(min_separator) for min_separator in min_separators
-        )
-        self.assertEqual(
-            {frozenset({2, 3}), frozenset({3, 4}), frozenset({4, 5})}, min_separators
-        )
+        min_separators = set(frozenset(min_separator) for min_separator in min_separators)
+        self.assertEqual({frozenset({2, 3}), frozenset({3, 4}), frozenset({4, 5})}, min_separators)
 
     def tearDown(self) -> None:
         remove_temp_dir_if_existent()