Added test generation to base metamorphic relation class

Author: AndrewC19

causal_testing/specification/metamorphic_relation.py

@@ -1,16 +1,75 @@
 from dataclasses import dataclass
 from abc import abstractmethod
+from typing import Iterable
+from itertools import combinations
+import numpy as np
+import pandas as pd
 
+from causal_testing.specification.causal_specification import CausalDAG, Node
 
-@dataclass(order=True, frozen=True)
+@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
 
-    @abstractmethod
-    def generate_follow_up(self, source_input_configuration):
-        """Generate a follow-up input configuration from a given source input
-           configuration."""
-        ...
+    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])
+        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)
+        )
+
+        # 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))
+        )
+
+        # 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}\'"
+        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])
+        )
+        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)
+            )
+        )
 
     @abstractmethod
     def test_oracle(self):
@@ -23,16 +82,18 @@ def execute_test(self):
         ...
 
 
-@dataclass(order=True, frozen=True)
+@dataclass(order=True)
 class ShouldCause(MetamorphicRelation):
     """Class representing a should cause metamorphic relation."""
 
-    def generate_follow_up(self, source_input_configuration):
-        pass
-
     def test_oracle(self):
         pass
 
     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

tests/specification_tests/test_metamorphic_relations.py

@@ -0,0 +1,23 @@
+import unittest
+import os
+
+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.metamorphic_relation import ShouldCause
+
+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; }"""
+        with open(self.dag_dot_path, "w") as f:
+            f.write(dag_dot)
+
+    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)