Codecov

Author: jmafoster1

causal_testing/json_front/json_class.py

@@ -117,7 +117,7 @@ def generate_tests(self, effects: dict, mutates: dict, estimators: dict, f_flag:
                 base_test_case = BaseTestCase(
                     treatment_variable=next(self.scenario.variables[v] for v in test["mutations"]),
                     outcome_variable=next(self.scenario.variables[v] for v in test["expectedEffect"]),
-                    effect=test["effect"],
+                    effect=test.get("effect", "direct"),
                 )
                 assert len(test["expectedEffect"]) == 1, "Can only have one expected effect."
                 concrete_tests = [

causal_testing/specification/causal_dag.py

@@ -521,5 +521,8 @@ def identification(self, base_test_case: BaseTestCase):
         minimal_adjustment_set = min(minimal_adjustment_sets, key=len)
         return minimal_adjustment_set
 
+    def to_dot(self):
+        return str(nx.nx_pydot.to_pydot(self.graph))
+
     def __str__(self):
         return f"Nodes: {self.graph.nodes}\nEdges: {self.graph.edges}"

tests/json_front_tests/test_json_class.py

@@ -102,6 +102,32 @@ def test_f_flag(self):
         with self.assertRaises(StatisticsError):
             self.json_class.generate_tests(effects, mutates, estimators, True)
 
+    def test_generate_coefficient_tests_from_json(self):
+        example_test = {
+            "tests": [
+                {
+                    "name": "test1",
+                    "mutations": ["test_input"],
+                    "estimator": "LinearRegressionEstimator",
+                    "estimate_type": "coefficient",
+                    "effect_modifiers": [],
+                    "expectedEffect": {"test_output": "NoEffect"},
+                    "skip": False,
+                }
+            ]
+        }
+        print(self.json_class.causal_specification.causal_dag.to_dot())
+        self.json_class.test_plan = example_test
+        effects = {"NoEffect": NoEffect()}
+        estimators = {"LinearRegressionEstimator": LinearRegressionEstimator}
+
+        self.json_class.generate_tests(effects, {}, estimators, False)
+
+        # Test that the final log message prints that failed tests are printed, which is expected behaviour for this scenario
+        with open("temp_out.txt", "r") as reader:
+            temp_out = reader.readlines()
+        self.assertIn("failed", temp_out[-1])
+
     def test_generate_tests_from_json(self):
         example_test = {
             "tests": [

tests/resources/data/dag.dot

@@ -1 +1 @@
-digraph G { test_input_no_dist; est_input -> B; B -> C; test_output -> test_input; test_output -> C}
+digraph G { test_input_no_dist; test_input -> B; B -> C; test_input -> test_output; test_output -> C}

tests/testing_tests/test_causal_test_case.py

@@ -1,37 +1,24 @@
 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_specification import CausalSpecification, Scenario
 from causal_testing.specification.variable import Input, Output
-from causal_testing.specification.causal_dag import CausalDAG
 from causal_testing.testing.causal_test_case import CausalTestCase
 from causal_testing.testing.causal_test_outcome import ExactValue
 from causal_testing.testing.base_test_case import BaseTestCase
 
 
-class TestCausalTestEngineObservational(unittest.TestCase):
-    """Test the CausalTestEngine workflow using observational data.
+class TestCausalTestCase(unittest.TestCase):
+    """Test the CausalTestCase class.
 
-    The causal test engine (CTE) is the main workflow for the causal testing framework. The CTE takes a causal test case
-    and a causal specification and computes the causal effect of the intervention on the outcome of interest.
+    The base test case is a data class which contains the minimum information
+    necessary to perform identification. The CausalTestCase class represents
+    a causal test case. We here test the basic getter methods.
     """
 
     def setUp(self) -> None:
-        # 1. Create Causal DAG
-        temp_dir_path = create_temp_dir_if_non_existent()
-        dag_dot_path = os.path.join(temp_dir_path, "dag.dot")
-        dag_dot = """digraph G { A -> C; D -> A; D -> C}"""
-        f = open(dag_dot_path, "w")
-        f.write(dag_dot)
-        f.close()
-        self.causal_dag = CausalDAG(dag_dot_path)
-
         # 2. Create Scenario and Causal Specification
         A = Input("A", float)
         C = Output("C", float)
-        D = Output("D", float)
-        self.scenario = Scenario({A, C, D})
-        self.causal_specification = CausalSpecification(scenario=self.scenario, causal_dag=self.causal_dag)
 
         # 3. Create an intervention and causal test case
         self.expected_causal_effect = ExactValue(4)

tests/testing_tests/test_estimators.py

@@ -176,6 +176,22 @@ def test_estimate_coefficient(self):
         )
         self.assertEqual(iv_estimator.estimate_coefficient(self.df), 2)
 
+    def test_estimate_unit_ate(self):
+        """
+        Test we get the correct coefficient.
+        """
+        iv_estimator = InstrumentalVariableEstimator(
+            df=self.df,
+            treatment="X",
+            treatment_value=None,
+            control_value=None,
+            adjustment_set=set(),
+            outcome="Y",
+            instrument="Z",
+        )
+        unit_ate, [low, high] = iv_estimator.estimate_unit_ate()
+        self.assertEqual(unit_ate, 2)
+
 
 class TestLinearRegressionEstimator(unittest.TestCase):
     """Test the linear regression estimator against the programming exercises in Section 2 of Hernán and Robins [1].