abstract success criteria constants and magic strings

lib/ramble/ramble/analysis/backwards.py

@@ -14,6 +14,7 @@
 import ramble.util.lock as lk
 from ramble.analysis.base import AnalysisStrategyBase
 from ramble.experiment_result import ExperimentStatus
+from ramble.success_criteria import SuccessCriteriaResult
 from ramble.util.foms import NULL_CONTEXT as _NULL_CONTEXT
 from ramble.util.logger import logger
 
@@ -246,9 +247,9 @@ def __call__(self, workspace):
             else:
                 criteria_name = f"config::{criteria_scope}::{criteria_obj.name}"
             if criteria_obj.ok():
-                app.result.success_criteria[criteria_name] = "PASSED"
+                app.result.success_criteria[criteria_name] = SuccessCriteriaResult.PASSED
             else:
-                app.result.success_criteria[criteria_name] = "FAILED"
+                app.result.success_criteria[criteria_name] = SuccessCriteriaResult.FAILED
 
         for context_key, fom_map in fom_values.items():
             metadata = context_metadata[context_key]

lib/ramble/ramble/analysis/forward.py

@@ -15,6 +15,7 @@
 import ramble.util.lock as lk
 from ramble.analysis.base import AnalysisStrategyBase
 from ramble.experiment_result import ExperimentStatus
+from ramble.success_criteria import SuccessCriteriaResult
 from ramble.util.foms import NULL_CONTEXT as _NULL_CONTEXT
 from ramble.util.logger import logger
 
@@ -231,9 +232,9 @@ def format_context(context_match, context_format):
             else:
                 criteria_name = f"config::{criteria_scope}::{criteria_obj.name}"
             if criteria_obj.ok():
-                app.result.success_criteria[criteria_name] = "PASSED"
+                app.result.success_criteria[criteria_name] = SuccessCriteriaResult.PASSED
             else:
-                app.result.success_criteria[criteria_name] = "FAILED"
+                app.result.success_criteria[criteria_name] = SuccessCriteriaResult.FAILED
 
         for context_key, fom_map in fom_values.items():
             metadata = context_metadata[context_key]

lib/ramble/ramble/success_criteria.py

@@ -13,6 +13,36 @@
 from ramble.util.logger import logger
 
 
+class SuccessCriteriaMode:
+    """Valid modes for a success criteria.
+
+    These values are part of the user-facing API (the ``success_criteria``
+    directive and workspace configuration), so they must not change.
+    """
+
+    STRING = "string"
+    APPLICATION_FUNCTION = "application_function"
+    FOM_COMPARISON = "fom_comparison"
+
+
+class SuccessCriteriaScope:
+    """Scopes a success criteria can be defined in."""
+
+    OBJECT_DEFINITIONS = "object_definitions"
+    EXPERIMENT = "experiment"
+
+
+class SuccessCriteriaResult:
+    """Result values recorded for each success criteria during analysis."""
+
+    PASSED = "PASSED"
+    FAILED = "FAILED"
+
+
+#: Name of the implicit criteria that calls the application's success function
+APPLICATION_FUNCTION_CRITERIA_NAME = "_application_function"
+
+
 class ScopedCriteriaList:
     """A scoped list of success criteria
 
@@ -21,20 +51,18 @@ class ScopedCriteriaList:
 
     Possible scopes are:
      - object_definitions
-     - application
-     - workload
      - experiment
 
     To see if success was met, all criteria will be checked and are AND-ed together.
     """
 
     _valid_scopes = [
-        "object_definitions",
-        "experiment",
+        SuccessCriteriaScope.OBJECT_DEFINITIONS,
+        SuccessCriteriaScope.EXPERIMENT,
     ]
     _flush_scopes = {
-        "experiment": ["experiment"],
-        "object_definitions": ["object_definitions"],
+        SuccessCriteriaScope.EXPERIMENT: [SuccessCriteriaScope.EXPERIMENT],
+        SuccessCriteriaScope.OBJECT_DEFINITIONS: [SuccessCriteriaScope.OBJECT_DEFINITIONS],
     }
 
     def __init__(self):
@@ -106,7 +134,11 @@ class SuccessCriteria:
     experiment.
     """
 
-    _valid_modes = ["string", "application_function", "fom_comparison"]
+    _valid_modes = [
+        SuccessCriteriaMode.STRING,
+        SuccessCriteriaMode.APPLICATION_FUNCTION,
+        SuccessCriteriaMode.FOM_COMPARISON,
+    ]
     _success_function = "evaluate_success"
 
     def __init__(
@@ -137,7 +169,7 @@ def __init__(
         self.owner = owning_object
         self.pre_filter = ""
 
-        if mode == "string":
+        if mode == SuccessCriteriaMode.STRING:
             if match is None and anti_match is None:
                 logger.die(
                     f'Success criteria with mode="{mode}" '
@@ -156,7 +188,7 @@ def __init__(
                 self.pre_filter = get_literal_from_regex(anti_match)
             self.file = file
 
-        elif mode == "fom_comparison":
+        elif mode == SuccessCriteriaMode.FOM_COMPARISON:
             if formula is None or fom_name is None:
                 logger.die(
                     f'Success criteria with mode="{mode}" '
@@ -168,18 +200,18 @@ def __init__(
 
     def passed(self, test=None, app_inst=None, fom_values=None):
         logger.debug(f"Testing criteria {self.name} mode = {self.mode}")
-        if self.mode == "string":
+        if self.mode == SuccessCriteriaMode.STRING:
             if self.match is not None:
                 if self.pre_filter and self.pre_filter not in test:
                     return False
                 match_obj = self.match.match(test)
                 if match_obj:
                     return True
-        elif self.mode == "application_function":
+        elif self.mode == SuccessCriteriaMode.APPLICATION_FUNCTION:
             if hasattr(app_inst, self._success_function):
                 func = getattr(app_inst, self._success_function)
                 return func()
-        elif self.mode == "fom_comparison":
+        elif self.mode == SuccessCriteriaMode.FOM_COMPARISON:
             if fom_values is None:
                 logger.die(
                     f'Success criteria of mode="{self.mode}" requires '
@@ -238,7 +270,7 @@ def passed(self, test=None, app_inst=None, fom_values=None):
 
     def anti_matched(self, test=None):
         logger.debug(f"Testing anti-criterion {self.name} mode = {self.mode}")
-        if self.mode == "string":
+        if self.mode == SuccessCriteriaMode.STRING:
             if self.anti_match is not None:
                 if self.pre_filter and self.pre_filter not in test:
                     return False

lib/ramble/ramble/test/success_criteria.py

@@ -174,3 +174,33 @@ def evaluate_predicate(expr, extra_vars=None):
         formula="{value} > 0",
     )
     assert not crit4.passed(app_inst=app_inst, fom_values=fom_values)
+
+
+def test_success_criteria_constant_values():
+    """Mode, scope, and result values are user-facing and must remain stable."""
+    from ramble.success_criteria import (
+        ScopedCriteriaList,
+        SuccessCriteria,
+        SuccessCriteriaMode,
+        SuccessCriteriaResult,
+        SuccessCriteriaScope,
+    )
+
+    # Modes are used in the success_criteria directive and workspace configs
+    assert SuccessCriteriaMode.STRING == "string"
+    assert SuccessCriteriaMode.APPLICATION_FUNCTION == "application_function"
+    assert SuccessCriteriaMode.FOM_COMPARISON == "fom_comparison"
+    assert set(SuccessCriteria._valid_modes) == {
+        "string",
+        "application_function",
+        "fom_comparison",
+    }
+
+    # Scopes appear in reported criteria names (config::<scope>::<name>)
+    assert SuccessCriteriaScope.OBJECT_DEFINITIONS == "object_definitions"
+    assert SuccessCriteriaScope.EXPERIMENT == "experiment"
+    assert set(ScopedCriteriaList._valid_scopes) == {"object_definitions", "experiment"}
+
+    # Results are written to results files
+    assert SuccessCriteriaResult.PASSED == "PASSED"
+    assert SuccessCriteriaResult.FAILED == "FAILED"

var/ramble/repos/builtin/base_classes/application-base/base_class.py

@@ -60,6 +60,11 @@
     variant,
 )
 from ramble.pipeline import pipelines as pipeline_enum
+from ramble.success_criteria import (
+    APPLICATION_FUNCTION_CRITERIA_NAME,
+    SuccessCriteriaMode,
+    SuccessCriteriaScope,
+)
 from ramble.util import cleaner, conversions, json_util
 from ramble.util.foms import NULL_CONTEXT as _NULL_CONTEXT
 from ramble.util.foms import (
@@ -654,7 +659,9 @@ def set_success_list(self, success_criteria):
 
         if success_criteria:
             for conf in success_criteria:
-                self.success_list.add_criteria("experiment", **conf)
+                self.success_list.add_criteria(
+                    SuccessCriteriaScope.EXPERIMENT, **conf
+                )
 
     def build_phase_order(self):
         if self._pipeline_graphs is not None:
@@ -1214,11 +1221,11 @@ def build_used_variables(self):
         # Add variables from success criteria
         criteria_list = self.success_list
         for criteria, _ in criteria_list.all_criteria():
-            if criteria.mode == "fom_comparison":
+            if criteria.mode == SuccessCriteriaMode.FOM_COMPARISON:
                 self.expander.expand_var(criteria.fom_formula)
                 self.expander.expand_var(criteria.fom_name)
                 self.expander.expand_var(criteria.fom_context)
-            elif criteria.mode == "application_function":
+            elif criteria.mode == SuccessCriteriaMode.APPLICATION_FUNCTION:
                 self.evaluate_success()
 
         if self.package_manager is not None:
@@ -3974,7 +3981,7 @@ def analysis_dicts(self, criteria_list):
         inmem_fom_defs = {}
 
         # Add the object defined criteria
-        criteria_list.flush_scope("object_definitions")
+        criteria_list.flush_scope(SuccessCriteriaScope.OBJECT_DEFINITIONS)
 
         resolved_criteria = {
             crit.name for crit, _ in criteria_list.all_criteria()
@@ -4009,7 +4016,7 @@ def analysis_dicts(self, criteria_list):
                         continue
 
                     resolved_criteria.add(criteria)
-                    if conf["mode"] == "string":
+                    if conf["mode"] == SuccessCriteriaMode.STRING:
                         match = (
                             self.expander.expand_var(conf["match"])
                             if conf["match"] is not None
@@ -4021,17 +4028,17 @@ def analysis_dicts(self, criteria_list):
                             else None
                         )
                         criteria_list.add_criteria(
-                            "object_definitions",
+                            SuccessCriteriaScope.OBJECT_DEFINITIONS,
                             criteria,
                             mode=conf["mode"],
                             match=match,
                             file=conf["file"],
                             anti_match=anti_match,
                             owning_object=obj_inst,
                         )
-                    elif conf["mode"] == "fom_comparison":
+                    elif conf["mode"] == SuccessCriteriaMode.FOM_COMPARISON:
                         criteria_list.add_criteria(
-                            "object_definitions",
+                            SuccessCriteriaScope.OBJECT_DEFINITIONS,
                             criteria,
                             conf["mode"],
                             fom_name=conf["fom_name"],
@@ -4040,11 +4047,11 @@ def analysis_dicts(self, criteria_list):
                             owning_object=obj_inst,
                         )
 
-        if "_application_function" not in resolved_criteria:
+        if APPLICATION_FUNCTION_CRITERIA_NAME not in resolved_criteria:
             criteria_list.add_criteria(
-                scope="object_definitions",
-                name="_application_function",
-                mode="application_function",
+                scope=SuccessCriteriaScope.OBJECT_DEFINITIONS,
+                name=APPLICATION_FUNCTION_CRITERIA_NAME,
+                mode=SuccessCriteriaMode.APPLICATION_FUNCTION,
                 owning_object=self,
             )