Formatted with black

Author: AndrewC19

causal_testing/data_collection/data_collector.py

@@ -106,9 +106,9 @@ def collect_data(self, **kwargs) -> pd.DataFrame:
         executions.
         """
         control_results_df = self.run_system_with_input_configuration(self.control_input_configuration)
-        control_results_df.rename('control_{}'.format, inplace=True)
+        control_results_df.rename("control_{}".format, inplace=True)
         treatment_results_df = self.run_system_with_input_configuration(self.treatment_input_configuration)
-        treatment_results_df.rename('treatment_{}'.format, inplace=True)
+        treatment_results_df.rename("treatment_{}".format, inplace=True)
         results_df = pd.concat([control_results_df, treatment_results_df], ignore_index=False)
         return results_df
 

causal_testing/specification/metamorphic_relation.py

@@ -13,50 +13,42 @@
 @dataclass(order=True)
 class MetamorphicRelation:
     """Class representing a metamorphic relation."""
+
     treatment_var: Node
     output_var: Node
     adjustment_vars: Iterable[Node]
     dag: CausalDAG
     tests: Iterable = None
 
-    def generate_follow_up(self,
-                           n_tests: int,
-                           min_val: float,
-                           max_val: float,
-                           seed: int = 0):
+    def generate_follow_up(self, n_tests: int, min_val: float, max_val: float, seed: int = 0):
         """Generate numerical follow-up input configurations."""
         np.random.seed(seed)
 
         # Get set of variables to change, excluding the treatment itself
-        variables_to_change = set([node for node in self.dag.graph.nodes if
-                                   self.dag.graph.in_degree(node) == 0])
+        variables_to_change = set([node for node in self.dag.graph.nodes if self.dag.graph.in_degree(node) == 0])
         if self.adjustment_vars:
             variables_to_change |= set(self.adjustment_vars)
         if self.treatment_var in variables_to_change:
             variables_to_change.remove(self.treatment_var)
 
         # Assign random numerical values to the variables to change
         test_inputs = pd.DataFrame(
-            np.random.randint(min_val, max_val,
-                              size=(n_tests, len(variables_to_change))
-                              ),
-            columns=sorted(variables_to_change)
+            np.random.randint(min_val, max_val, size=(n_tests, len(variables_to_change))),
+            columns=sorted(variables_to_change),
         )
 
         # Enumerate the possible source, follow-up pairs for the treatment
-        candidate_source_follow_up_pairs = np.array(
-            list(combinations(range(int(min_val), int(max_val+1)), 2))
-        )
+        candidate_source_follow_up_pairs = np.array(list(combinations(range(int(min_val), int(max_val + 1)), 2)))
 
         # Sample without replacement from the possible source, follow-up pairs
         sampled_source_follow_up_indices = np.random.choice(
             candidate_source_follow_up_pairs.shape[0], n_tests, replace=False
         )
 
-        follow_up_input = f"{self.treatment_var}\'"
+        follow_up_input = f"{self.treatment_var}'"
         source_follow_up_test_inputs = pd.DataFrame(
             candidate_source_follow_up_pairs[sampled_source_follow_up_indices],
-            columns=sorted([self.treatment_var] + [follow_up_input])
+            columns=sorted([self.treatment_var] + [follow_up_input]),
         )
         source_test_inputs = source_follow_up_test_inputs[[self.treatment_var]]
         follow_up_test_inputs = source_follow_up_test_inputs[[follow_up_input]]
@@ -69,12 +61,13 @@ def generate_follow_up(self,
             other_test_inputs_record = [{}] * len(source_test_inputs)
         metamorphic_tests = []
         for i in range(len(source_test_inputs_record)):
-            metamorphic_test = MetamorphicTest(source_test_inputs_record[i],
-                                               follow_up_test_inputs_record[i],
-                                               other_test_inputs_record[i],
-                                               self.output_var,
-                                               str(self)
-                                               )
+            metamorphic_test = MetamorphicTest(
+                source_test_inputs_record[i],
+                follow_up_test_inputs_record[i],
+                other_test_inputs_record[i],
+                self.output_var,
+                str(self),
+            )
             metamorphic_tests.append(metamorphic_test)
         self.tests = metamorphic_tests
 
@@ -124,8 +117,9 @@ def assertion(self, source_output, follow_up_output):
 
     def test_oracle(self, test_results):
         """A single passing test is sufficient to show presence of a causal effect."""
-        assert len(test_results["fail"]) < len(self.tests),\
-            f"{str(self)}: {len(test_results['fail'])}/{len(self.tests)} tests failed."
+        assert len(test_results["fail"]) < len(
+            self.tests
+        ), f"{str(self)}: {len(test_results['fail'])}/{len(self.tests)} tests failed."
 
     def __str__(self):
         formatted_str = f"{self.treatment_var} --> {self.output_var}"
@@ -143,8 +137,9 @@ def assertion(self, source_output, follow_up_output):
 
     def test_oracle(self, test_results):
         """A single passing test is sufficient to show presence of a causal effect."""
-        assert len(test_results["fail"]) == 0,\
-            f"{str(self)}: {len(test_results['fail'])}/{len(self.tests)} tests failed."
+        assert (
+            len(test_results["fail"]) == 0
+        ), f"{str(self)}: {len(test_results['fail'])}/{len(self.tests)} tests failed."
 
     def __str__(self):
         formatted_str = f"{self.treatment_var} _||_ {self.output_var}"
@@ -156,18 +151,21 @@ def __str__(self):
 @dataclass(order=True)
 class MetamorphicTest:
     """Class representing a metamorphic test case."""
+
     source_inputs: dict
     follow_up_inputs: dict
     other_inputs: dict
     output: str
     relation: str
 
     def __str__(self):
-        return f"Source inputs: {self.source_inputs}\n" \
-               f"Follow-up inputs: {self.follow_up_inputs}\n" \
-               f"Other inputs: {self.other_inputs}\n" \
-               f"Output: {self.output}" \
-               f"Metamorphic Relation: {self.relation}"
+        return (
+            f"Source inputs: {self.source_inputs}\n"
+            f"Follow-up inputs: {self.follow_up_inputs}\n"
+            f"Other inputs: {self.other_inputs}\n"
+            f"Output: {self.output}"
+            f"Metamorphic Relation: {self.relation}"
+        )
 
 
 def generate_metamorphic_relations(dag: CausalDAG) -> list[MetamorphicRelation]:

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()