Remove custom helpers and use tempfile for all tests

Author: christopher-wild

tests/data_collection_tests/test_observational_data_collector.py

@@ -1,13 +1,13 @@
 import unittest
 import os
+import shutil, tempfile
 import pandas as pd
 from causal_testing.data_collection.data_collector import ObservationalDataCollector
 from causal_testing.specification.causal_specification import Scenario
 from causal_testing.specification.variable import Input, Output, Meta
 from scipy.stats import uniform, rv_discrete
 from enum import Enum
 import random
-from tests.test_helpers import create_temp_dir_if_non_existent, remove_temp_dir_if_existent
 
 
 class TestObservationalDataCollector(unittest.TestCase):
@@ -17,9 +17,9 @@ class Color(Enum):
             GREEN = "GREEN"
             BLUE = "BLUE"
 
-        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")
+        self.temp_dir_path = tempfile.mkdtemp()
+        self.dag_dot_path = os.path.join(self.temp_dir_path, "dag.dot")
+        self.observational_df_path = os.path.join(self.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], "Y2": ["RED", "GREEN", "BLUE", "BLUE"]}
@@ -66,7 +66,7 @@ def populate_m(data):
         assert all((m == 2 * x1 for x1, m in zip(data["X1"], data["M"])))
 
     def tearDown(self) -> None:
-        remove_temp_dir_if_existent()
+        shutil.rmtree(self.temp_dir_path)
 
 
 if __name__ == "__main__":

tests/generation_tests/test_abstract_test_case.py

@@ -1,13 +1,13 @@
 import unittest
 import os
+import shutil, tempfile
 import pandas as pd
 import numpy as np
 from causal_testing.generation.abstract_causal_test_case import AbstractCausalTestCase
 from causal_testing.generation.enum_gen import EnumGen
 from causal_testing.specification.causal_specification import Scenario
 from causal_testing.specification.variable import Input, Output
 from scipy.stats import uniform, rv_discrete
-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
 from z3 import And
 from enum import Enum
@@ -29,9 +29,9 @@ class TestAbstractTestCase(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")
+        self.temp_dir_path = tempfile.mkdtemp()
+        self.dag_dot_path = os.path.join(self.temp_dir_path, "dag.dot")
+        self.observational_df_path = os.path.join(self.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(
@@ -192,7 +192,7 @@ def test_feasible_constraints(self):
         assert len(concrete_tests) < 1000
 
     def tearDown(self) -> None:
-        remove_temp_dir_if_existent()
+        shutil.rmtree(self.temp_dir_path)
 
 
 if __name__ == "__main__":

tests/json_front_tests/test_json_class.py

@@ -6,7 +6,6 @@
 
 from causal_testing.testing.estimators import LinearRegressionEstimator, Estimator
 from causal_testing.testing.causal_test_outcome import NoEffect, Positive
-from tests.test_helpers import remove_temp_dir_if_existent
 from causal_testing.json_front.json_class import JsonUtility, CausalVariables
 from causal_testing.specification.variable import Input, Output, Meta
 from causal_testing.specification.scenario import Scenario
@@ -321,7 +320,6 @@ def add_modelling_assumptions(self):
             self.json_class.run_json_tests(effects=effects, mutates=mutates, estimators=estimators, f_flag=False)
 
     def tearDown(self) -> None:
-        remove_temp_dir_if_existent()
         if os.path.exists("temp_out.txt"):
             os.remove("temp_out.txt")
 

tests/specification_tests/test_causal_dag.py

@@ -1,11 +1,13 @@
 import unittest
 import os
+import shutil, tempfile
 import networkx as nx
 from causal_testing.specification.causal_dag import CausalDAG, close_separator, list_all_min_sep
 from causal_testing.specification.scenario import Scenario
 from causal_testing.specification.variable import Input, Output
 from causal_testing.testing.base_test_case import BaseTestCase
-from tests.test_helpers import create_temp_dir_if_non_existent, remove_temp_dir_if_existent
+
+
 
 
 class TestCausalDAGIssue90(unittest.TestCase):
@@ -14,8 +16,8 @@ class TestCausalDAGIssue90(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.temp_dir_path = tempfile.mkdtemp()
+        self.dag_dot_path = os.path.join(self.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)
@@ -28,7 +30,7 @@ def test_enumerate_minimal_adjustment_sets(self):
         self.assertEqual([{"Z"}], adjustment_sets)
 
     def tearDown(self) -> None:
-        remove_temp_dir_if_existent()
+        shutil.rmtree(self.temp_dir_path)
 
 
 class TestIVAssumptions(unittest.TestCase):

tests/specification_tests/test_metamorphic_relations.py

@@ -1,10 +1,9 @@
 import unittest
 import os
-
+import shutil, tempfile
 import pandas as pd
 from itertools import combinations
 
-from tests.test_helpers import create_temp_dir_if_non_existent
 from causal_testing.specification.causal_dag import CausalDAG
 from causal_testing.specification.causal_specification import Scenario
 from causal_testing.specification.metamorphic_relation import (
@@ -69,8 +68,8 @@ def run_system_with_input_configuration(self, input_configuration: dict) -> pd.D
 
 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")
+        self.temp_dir_path = tempfile.mkdtemp()
+        self.dag_dot_path = os.path.join(self.temp_dir_path, "dag.dot")
         dag_dot = """digraph DAG { rankdir=LR; X1 -> Z; Z -> M; M -> Y; X2 -> Z; X3 -> M;}"""
         with open(self.dag_dot_path, "w") as f:
             f.write(dag_dot)
@@ -88,6 +87,9 @@ def setUp(self) -> None:
             self.scenario, self.default_control_input_config, self.default_treatment_input_config
         )
 
+    def tearDown(self) -> None:
+        shutil.rmtree(self.temp_dir_path)
+
     def test_should_cause_metamorphic_relations_correct_spec(self):
         """Test if the ShouldCause MR passes all metamorphic tests where the DAG perfectly represents the program."""
         causal_dag = CausalDAG(self.dag_dot_path)

tests/surrogate_tests/test_causal_surrogate_assisted.py

@@ -7,8 +7,9 @@
 from causal_testing.surrogate.causal_surrogate_assisted import SimulationResult, CausalSurrogateAssistedTestCase, Simulator
 from causal_testing.surrogate.surrogate_search_algorithms import GeneticSearchAlgorithm
 from causal_testing.testing.estimators import CubicSplineRegressionEstimator
-from tests.test_helpers import create_temp_dir_if_non_existent, remove_temp_dir_if_existent
+
 import os
+import shutil, tempfile
 import pandas as pd
 import numpy as np
 
@@ -43,8 +44,8 @@ def setUpClass(cls) -> None:
         cls.class_df = load_class_df()
 
     def setUp(self):
-        temp_dir_path = create_temp_dir_if_non_existent()
-        self.dag_dot_path = os.path.join(temp_dir_path, "dag.dot")
+        self.temp_dir_path = tempfile.mkdtemp()
+        self.dag_dot_path = os.path.join(self.temp_dir_path, "dag.dot")
         dag_dot = """digraph DAG { rankdir=LR; Z -> X; X -> M [included=1, expected=positive]; M -> Y [included=1, expected=negative]; Z -> M; }"""
         with open(self.dag_dot_path, "w") as f:
             f.write(dag_dot)
@@ -199,7 +200,7 @@ def test_causal_surrogate_assisted_execution_incorrect_search_config(self):
                           custom_data_aggregator=data_double_aggregator)
 
     def tearDown(self) -> None:
-        remove_temp_dir_if_existent()
+        shutil.rmtree(self.temp_dir_path)
 
 def load_class_df():
     """Get the testing data and put into a dataframe."""

tests/testing_tests/test_causal_test_adequacy.py

@@ -10,7 +10,6 @@
 from causal_testing.testing.causal_test_suite import CausalTestSuite
 from causal_testing.testing.causal_test_adequacy import DAGAdequacy
 from causal_testing.testing.causal_test_outcome import NoEffect, Positive
-from tests.test_helpers import remove_temp_dir_if_existent
 from causal_testing.json_front.json_class import JsonUtility, CausalVariables
 from causal_testing.specification.variable import Input, Output, Meta
 from causal_testing.specification.scenario import Scenario
@@ -255,6 +254,5 @@ def test_dag_adequacy_independent_other_way(self):
         )
 
     def tearDown(self) -> None:
-        remove_temp_dir_if_existent()
         if os.path.exists("temp_out.txt"):
             os.remove("temp_out.txt")

tests/testing_tests/test_causal_test_case.py

@@ -1,9 +1,10 @@
 import unittest
 import os
+import tempfile
+import shutil
 import pandas as pd
 import numpy as np
 
-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
@@ -44,9 +45,6 @@ def test_str(self):
             " {Output: C::float}: ExactValue: 4±0.2.",
         )
 
-    def tearDown(self) -> None:
-        remove_temp_dir_if_existent()
-
 
 class TestCausalTestExecution(unittest.TestCase):
     """Test the causal test execution workflow using observational data.
@@ -57,8 +55,8 @@ class TestCausalTestExecution(unittest.TestCase):
 
     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")
+        self.temp_dir_path = tempfile.mkdtemp()
+        dag_dot_path = os.path.join(self.temp_dir_path, "dag.dot")
         dag_dot = """digraph G { A -> C; D -> A; D -> C}"""
         with open(dag_dot_path, "w") as file:
             file.write(dag_dot)
@@ -88,7 +86,7 @@ def setUp(self) -> None:
         df = pd.DataFrame({"D": list(np.random.normal(60, 10, 1000))})  # D = exogenous
         df["A"] = [1 if d > 50 else 0 for d in df["D"]]
         df["C"] = df["D"] + (4 * (df["A"] + 2))  # C = (4*(A+2)) + D
-        self.observational_data_csv_path = os.path.join(temp_dir_path, "observational_data.csv")
+        self.observational_data_csv_path = os.path.join(self.temp_dir_path, "observational_data.csv")
         df.to_csv(self.observational_data_csv_path, index=False)
 
         # 5. Create observational data collector
@@ -101,6 +99,9 @@ def setUp(self) -> None:
         self.treatment_value = 1
         self.control_value = 0
 
+    def tearDown(self) -> None:
+        shutil.rmtree(self.temp_dir_path)
+
     def test_check_minimum_adjustment_set(self):
         """Check that the minimum adjustment set is correctly made"""
         minimal_adjustment_set = self.causal_dag.identification(self.base_test_case)
@@ -215,6 +216,3 @@ def test_execute_test_observational_linear_regression_estimator_squared_term(sel
         )
         causal_test_result = self.causal_test_case.execute_test(estimation_model, self.data_collector)
         pd.testing.assert_series_equal(causal_test_result.test_value.value, pd.Series(4.0), atol=1)
-
-    def tearDown(self) -> None:
-        remove_temp_dir_if_existent()