Merge pull request #189 from CITCOM-project/json_concrete_param

Expose concrete test generation parameters in JSON Frontend

Author: christopher-wild

causal_testing/json_front/json_class.py

@@ -123,54 +123,15 @@ def run_json_tests(self, effects: dict, estimators: dict, f_flag: bool = False,
         :param estimators: Dictionary mapping estimator classes to string representations.
         :param f_flag: Failure flag that if True the script will stop executing when a test fails.
         """
-        failures = 0
-        msg = ""
         for test in self.test_plan["tests"]:
             if "skip" in test and test["skip"]:
                 continue
             test["estimator"] = estimators[test["estimator"]]
             if "mutations" in test:
                 if test["estimate_type"] == "coefficient":
-                    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["expected_effect"]),
-                        effect=test.get("effect", "direct"),
-                    )
-                    assert len(test["expected_effect"]) == 1, "Can only have one expected effect."
-                    causal_test_case = CausalTestCase(
-                        base_test_case=base_test_case,
-                        expected_causal_effect=next(
-                            effects[effect] for variable, effect in test["expected_effect"].items()
-                        ),
-                        estimate_type="coefficient",
-                        effect_modifier_configuration={
-                            self.scenario.variables[v] for v in test.get("effect_modifiers", [])
-                        },
-                    )
-                    result = self._execute_test_case(causal_test_case=causal_test_case, test=test, f_flag=f_flag)
-                    msg = (
-                        f"Executing test: {test['name']} \n"
-                        + f"  {causal_test_case} \n"
-                        + "  "
-                        + ("\n  ").join(str(result[1]).split("\n"))
-                        + "==============\n"
-                        + f"  Result: {'FAILED' if result[0] else 'Passed'}"
-                    )
-                    print(msg)
+                    msg = self._run_coefficient_test(test=test, f_flag=f_flag, effects=effects)
                 else:
-                    abstract_test = self._create_abstract_test_case(test, mutates, effects)
-                    concrete_tests, _ = abstract_test.generate_concrete_tests(5, 0.05)
-                    failures, _ = self._execute_tests(concrete_tests, test, f_flag)
-
-                    msg = (
-                        f"Executing test: {test['name']} \n"
-                        + "  abstract_test \n"
-                        + f"  {abstract_test} \n"
-                        + f"  {abstract_test.treatment_variable.name},"
-                        + f"  {abstract_test.treatment_variable.distribution} \n"
-                        + f"  Number of concrete tests for test case: {str(len(concrete_tests))} \n"
-                        + f"  {failures}/{len(concrete_tests)} failed for {test['name']}"
-                    )
+                    msg = self._run_ate_test(test=test, f_flag=f_flag, effects=effects, mutates=mutates)
                 self._append_to_file(msg, logging.INFO)
             else:
                 outcome_variable = next(
@@ -197,8 +158,74 @@ def run_json_tests(self, effects: dict, estimators: dict, f_flag: bool = False,
                     + f"control value = {test['control_value']}, treatment value = {test['treatment_value']} \n"
                     + f"Result: {'FAILED' if failed else 'Passed'}"
                 )
+                print(msg)
                 self._append_to_file(msg, logging.INFO)
 
+    def _run_coefficient_test(self, test: dict, f_flag: bool, effects: dict):
+        """Builds structures and runs test case for tests with an estimate_type of 'coefficient'.
+
+        :param test: Single JSON test definition stored in a mapping (dict)
+        :param f_flag: Failure flag that if True the script will stop executing when a test fails.
+        :param effects: Dictionary mapping effect class instances to string representations.
+        :return: String containing the message to be outputted
+        """
+        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["expected_effect"]),
+            effect=test.get("effect", "direct"),
+        )
+        assert len(test["expected_effect"]) == 1, "Can only have one expected effect."
+        causal_test_case = CausalTestCase(
+            base_test_case=base_test_case,
+            expected_causal_effect=next(effects[effect] for variable, effect in test["expected_effect"].items()),
+            estimate_type="coefficient",
+            effect_modifier_configuration={self.scenario.variables[v] for v in test.get("effect_modifiers", [])},
+        )
+        result = self._execute_test_case(causal_test_case=causal_test_case, test=test, f_flag=f_flag)
+        msg = (
+            f"Executing test: {test['name']} \n"
+            + f"  {causal_test_case} \n"
+            + "  "
+            + ("\n  ").join(str(result[1]).split("\n"))
+            + "==============\n"
+            + f"  Result: {'FAILED' if result[0] else 'Passed'}"
+        )
+        return msg
+
+    def _run_ate_test(self, test: dict, f_flag: bool, effects: dict, mutates: dict):
+        """Builds structures and runs test case for tests with an estimate_type of 'ate'.
+
+        :param test: Single JSON test definition stored in a mapping (dict)
+        :param f_flag: Failure flag that if True the script will stop executing when a test fails.
+        :param effects: Dictionary mapping effect class instances to string representations.
+        :param mutates: Dictionary mapping mutation functions to string representations.
+        :return: String containing the message to be outputted
+        """
+        if "sample_size" in test:
+            sample_size = test["sample_size"]
+        else:
+            sample_size = 5
+        if "target_ks_score" in test:
+            target_ks_score = test["target_ks_score"]
+        else:
+            target_ks_score = 0.05
+        abstract_test = self._create_abstract_test_case(test, mutates, effects)
+        concrete_tests, _ = abstract_test.generate_concrete_tests(
+            sample_size=sample_size, target_ks_score=target_ks_score
+        )
+        failures, _ = self._execute_tests(concrete_tests, test, f_flag)
+
+        msg = (
+            f"Executing test: {test['name']} \n"
+            + "  abstract_test \n"
+            + f"  {abstract_test} \n"
+            + f"  {abstract_test.treatment_variable.name},"
+            + f"  {abstract_test.treatment_variable.distribution} \n"
+            + f"  Number of concrete tests for test case: {str(len(concrete_tests))} \n"
+            + f"  {failures}/{len(concrete_tests)} failed for {test['name']}"
+        )
+        return msg
+
     def _execute_tests(self, concrete_tests, test, f_flag):
         failures = 0
         details = []

tests/json_front_tests/test_json_class.py

@@ -239,6 +239,37 @@ def test_run_concrete_json_testcase(self):
         with open("temp_out.txt", "r") as reader:
             temp_out = reader.readlines()
         self.assertIn("FAILED", temp_out[-1])
+    def test_concrete_generate_params(self):
+        example_test = {
+            "tests": [
+                {
+                    "name": "test1",
+                    "mutations": {"test_input": "Increase"},
+                    "estimator": "LinearRegressionEstimator",
+                    "estimate_type": "ate",
+                    "effect_modifiers": [],
+                    "expected_effect": {"test_output": "NoEffect"},
+                    "sample_size": 5,
+                    "target_ks_score": 0.05,
+                    "skip": False,
+                }
+            ]
+        }
+        self.json_class.test_plan = example_test
+        effects = {"NoEffect": NoEffect()}
+        mutates = {
+            "Increase": lambda x: self.json_class.scenario.treatment_variables[x].z3
+                                  > self.json_class.scenario.variables[x].z3
+        }
+        estimators = {"LinearRegressionEstimator": LinearRegressionEstimator}
+
+        self.json_class.run_json_tests(effects=effects, estimators=estimators, f_flag=False, mutates=mutates)
+
+        # 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_no_data_provided(self):
         example_test = {