ShouldCause metamorphic relation gen and testing works

Author: AndrewC19

causal_testing/data_collection/data_collector.py

@@ -106,8 +106,10 @@ 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)
         treatment_results_df = self.run_system_with_input_configuration(self.treatment_input_configuration)
-        results_df = pd.concat([control_results_df, treatment_results_df], ignore_index=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
 
     @abstractmethod

causal_testing/specification/metamorphic_relation.py

@@ -6,6 +6,7 @@
 import pandas as pd
 
 from causal_testing.specification.causal_specification import CausalDAG, Node
+from causal_testing.data_collection.data_collector import ExperimentalDataCollector
 
 @dataclass(order=True)
 class MetamorphicRelation:
@@ -57,43 +58,91 @@ def generate_follow_up(self,
         )
         source_test_inputs = source_follow_up_test_inputs[[self.treatment_var]]
         follow_up_test_inputs = source_follow_up_test_inputs[[follow_up_input]]
-        follow_up_test_inputs.rename({follow_up_input: self.treatment_var})
-
-        # TODO: Add a metamorphic test dataclass that stores these attributes
-        self.tests = list(
-            zip(
-                source_test_inputs.to_dict(orient="records"),
-                follow_up_test_inputs.to_dict(orient="records"),
-                test_inputs.to_dict(orient="records") if not test_inputs.empty
-                else [{}] * len(source_test_inputs),
-                [self.output_var] * len(source_test_inputs),
-                [str(self)] * len(source_test_inputs)
-            )
-        )
+        follow_up_test_inputs = follow_up_test_inputs.rename(columns={follow_up_input: self.treatment_var})
+        source_test_inputs_record = source_test_inputs.to_dict(orient="records")
+        follow_up_test_inputs_record = follow_up_test_inputs.to_dict(orient="records")
+        if not test_inputs.empty:
+            other_test_inputs_record = test_inputs.to_dict(orient="records")
+        else:
+            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_tests.append(metamorphic_test)
+        self.tests = metamorphic_tests
+
+    def execute_tests(self, data_collector: ExperimentalDataCollector):
+        """Execute the generated list of metamorphic tests, returning a dictionary of tests that pass and fail.
+
+        :param data_collector: An experimental data collector for the system-under-test.
+        """
+        test_results = {"pass": [], "fail": []}
+        for metamorphic_test in self.tests:
+            # Update the control and treatment configuration to take generated values for source and follow-up tests
+            control_input_config = metamorphic_test.source_inputs | metamorphic_test.other_inputs
+            treatment_input_config = metamorphic_test.follow_up_inputs | metamorphic_test.other_inputs
+            data_collector.control_input_configuration = control_input_config
+            data_collector.treatment_input_configuration = treatment_input_config
+            metamorphic_test_results_df = data_collector.collect_data()
+            print(metamorphic_test_results_df)
+            # Compare control and treatment results
+            control_output = metamorphic_test_results_df.loc["control_0"][metamorphic_test.output]
+            treatment_output = metamorphic_test_results_df.loc["treatment_0"][metamorphic_test.output]
+            if not self.assertion(control_output, treatment_output):
+                test_results["fail"].append(metamorphic_test)
+            else:
+                test_results["pass"].append(metamorphic_test)
+            return test_results
 
     @abstractmethod
-    def test_oracle(self):
-        """A test oracle i.e. a method that checks correctness of a test."""
+    def assertion(self, source_output, follow_up_output):
+        """An assertion that should be applied to an individual metamorphic test run."""
         ...
 
     @abstractmethod
-    def execute_test(self):
-        """Execute a test for this metamorphic relation."""
+    def test_oracle(self, test_results):
+        """A test oracle that assert whether the MR holds or not based on ALL test results.
+
+        This method must raise an assertion, not return a bool."""
         ...
 
 
 @dataclass(order=True)
 class ShouldCause(MetamorphicRelation):
     """Class representing a should cause metamorphic relation."""
 
-    def test_oracle(self):
-        pass
+    def assertion(self, source_output, follow_up_output):
+        """If there is a causal effect, the outputs should not be the same."""
+        return source_output != follow_up_output
+
+    def test_oracle(self, test_results):
+        ...
 
-    def execute_test(self):
-        pass
 
     def __str__(self):
         formatted_str = f"{self.treatment_var} --> {self.output_var}"
         if self.adjustment_vars:
             formatted_str += f" | {self.adjustment_vars}"
         return formatted_str
+
+
+@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}"

tests/specification_tests/test_metamorphic_relations.py

@@ -1,23 +1,63 @@
 import unittest
 import os
 
+import pandas as pd
+
 from tests.test_helpers import create_temp_dir_if_non_existent, remove_temp_dir_if_existent
 from causal_testing.specification.causal_dag import CausalDAG
+from causal_testing.specification.causal_specification import Scenario
 from causal_testing.specification.metamorphic_relation import ShouldCause
+from causal_testing.data_collection.data_collector import ExperimentalDataCollector
+from causal_testing.specification.variable import Input, Output
+
+
+def program_under_test(X1, X2, X3, Z=None, M=None, Y=None):
+    if Z is None:
+        Z = 2*X1 + -3*X2 + 10
+    if M is None:
+        M = 3*Z + X3
+    if Y is None:
+        Y = M/2
+    return {'Z': Z, 'M': M, 'Y': Y}
+
+
+class ProgramUnderTestEDC(ExperimentalDataCollector):
+
+    def run_system_with_input_configuration(self, input_configuration: dict) -> pd.DataFrame:
+        print(input_configuration)
+        results_dict = program_under_test(**input_configuration)
+        print(results_dict)
+        results_df = pd.DataFrame(results_dict, index=[0])
+        return results_df
+
 
 class TestMetamorphicRelation(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 DAG { rankdir=LR; Z -> X; X -> M; M -> Y; Z -> M; }"""
+        dag_dot = """digraph DAG { rankdir=LR; X1 -> Z; Z -> M; M -> Y; X1 -> M; X2 -> Z; X3 -> M;}"""
         with open(self.dag_dot_path, "w") as f:
             f.write(dag_dot)
 
+        X1 = Input('X1', float)
+        X2 = Input('X2', float)
+        X3 = Input('X3', float)
+        Z = Output('Z', float)
+        M = Output('M', float)
+        Y = Output('Y', float)
+        scenario = Scenario(variables={X1, X2, X3, Z, M, Y})
+        default_control_input_config = {'X1': 1, 'X2': 2, 'X3': 3}
+        default_treatment_input_config = {'X1': 2, 'X2': 3, 'X3': 3}
+        self.data_collector = ProgramUnderTestEDC(scenario,
+                                                  default_control_input_config,
+                                                  default_treatment_input_config)
+
     def test_metamorphic_relation(self):
         causal_dag = CausalDAG(self.dag_dot_path)
         for edge in causal_dag.graph.edges:
             (u, v) = edge
             should_cause_MR = ShouldCause(u, v, None, causal_dag)
-            should_cause_MR.generate_follow_up(1, -10.0, 10.0, 1)
-            print(should_cause_MR.tests)
+            should_cause_MR.generate_follow_up(10, -10.0, 10.0, 1)
+            test_results = should_cause_MR.execute_tests(self.data_collector)
+            print(test_results)