Local check runner

CHANGELOG.rst

@@ -6,6 +6,12 @@ foursight-core
 Change Log
 ----------
 
+4.4.1
+=====
+* 2023-07-17
+* Support local-check-runner utility.
+
+
 4.4.0
 =====
 * 2023-06-20

foursight_core/captured_output.py

@@ -0,0 +1,87 @@
+import builtins
+from collections import namedtuple
+from contextlib import contextmanager
+import io
+import sys
+from typing import Optional
+
+_original_print = builtins.print
+_original_stdout = sys.stdout
+_original_stderr = sys.stderr
+
+@contextmanager
+def captured_output(capture: bool = True):
+    """
+    Context manager to capture any/all output to stdout or stderr, and not actually output it to stdout
+    or stderr. Yields and object with a get_captured_output() method to get the output captured thus far,
+    and another uncaptured_print() method to actually print the given output to stdout, even though output
+    to stdout is being captured. Can be useful, for example, in creating command-line scripts which invoke
+    code which outputs a lot of info, warning, error, etc to stdout or stderr, and we want to suprress that
+    output; but with the yielded uncaptured_print() method output specific to the script can actually be
+    output (to stdout); and/or can also optionally output any/all captured output, e.g. for debugging or
+    troubleshooting purposes. Disable this capture, without having to restructure your code WRT the usage
+    of the with-clause with this context manager, pass False as an argument to this context manager. 
+    """
+
+    save_print = _original_print
+    save_stdout = _original_stdout
+    save_stderr = _original_stderr
+    captured_output = io.StringIO()
+
+    def captured_print(*args, **kwargs) -> None:
+        captured_output.write(*args)
+        captured_output.write("\n")
+
+    def uncaptured_print(*args, **kwargs) -> None:
+        builtins.print = save_print
+        sys.stdout = save_stdout
+        sys.stderr = save_stderr
+        print(*args, **kwargs)
+        if capture:
+            builtins.print = captured_print
+            sys.stdout = captured_output
+            sys.stderr = captured_output
+
+    def uncaptured_input(message: str) -> str:
+        builtins.print = save_print
+        sys.stdout = save_stdout
+        sys.stderr = save_stderr
+        value = input(message)
+        if capture:
+            builtins.print = captured_print
+            sys.stdout = captured_output
+            sys.stderr = captured_output
+        return value
+
+    def get_captured_output() -> Optional[str]:
+        return captured_output.getvalue() if capture else None
+
+    if capture:
+        builtins.print = captured_print
+        sys.stdout = captured_output
+        sys.stderr = captured_output
+
+    Result = namedtuple("Result", ["get_captured_output", "uncaptured_print", "uncaptured_input"])
+
+    try:
+        yield Result(get_captured_output, uncaptured_print, uncaptured_input)
+    finally:
+        builtins.print = save_print
+        sys.stdout = save_stdout
+        sys.stderr = save_stderr
+
+
+@contextmanager
+def uncaptured_output():
+    save_print = builtins.print
+    save_stdout = sys.stdout
+    save_stderr = sys.stderr
+    builtins.print = _original_print
+    sys.stdout = _original_stdout
+    sys.stderr = _original_stderr
+    try:
+        yield
+    finally:
+        builtins.print = save_print
+        sys.stdout = save_stdout
+        sys.stderr = save_stderr

foursight_core/check_utils.py

@@ -1,8 +1,10 @@
-import os
-import importlib
+from collections import namedtuple
 import copy
+import importlib
 import json
 import logging
+import os
+from typing import Callable, Optional
 from dcicutils.env_base import EnvBase
 from dcicutils.env_utils import infer_foursight_from_env
 from dcicutils.misc_utils import json_leaf_subst
@@ -39,7 +41,7 @@ def __init__(self, foursight_prefix, check_package_name='foursight_core', check_
         # which calls back to the locate_check_setup_file function AppUtilsCore here in foursight-core).
         if not os.path.exists(check_setup_file):
             raise BadCheckSetup(f"Did not locate the specified check setup file: {check_setup_file}")
-        self.CHECK_SETUP_FILE = check_setup_file # for display/troubleshooting
+        self.CHECK_SETUP_FILE = check_setup_file  # for display/troubleshooting
         with open(check_setup_file, 'r') as jfile:
             self.CHECK_SETUP = json.load(jfile)
         logger.debug(f"foursight_core/CheckHandler: Loaded check_setup.json file: {check_setup_file} ...")
@@ -360,7 +362,7 @@ def get_grouped_check_results(self, connection):
         grouped_list = [group for group in grouped_results.values()]
         return sorted(grouped_list, key=lambda v: v['_name'])
 
-    def run_check_or_action(self, connection, check_str, check_kwargs):
+    def obsolete_run_check_or_action(self, connection, check_str, check_kwargs):
         """
         Does validation of provided check_str, it's module, and kwargs.
         Determines by decorator whether the method is a check or action, then runs
@@ -400,6 +402,153 @@ def run_check_or_action(self, connection, check_str, check_kwargs):
             return ' '.join(['ERROR. Check or action must use a decorator.', error_str])
         return check_method(connection, **check_kwargs)
 
+    def run_check_or_action(self, connection, check_str, check_kwargs):
+        """
+        Does validation of provided check_str, it's module, and kwargs.
+        Determines by decorator whether the method is a check or action, then runs
+        it. All errors are taken care of within the running of the check/action.
+
+        Takes a FS_connection object, a check string formatted as: <str check module/name>
+        and a dictionary of check arguments.
+        For example:
+        check_str: 'system_checks/my_check'
+        check_kwargs: '{"foo":123}'
+        Fetches the check function and runs it (returning whatever it returns)
+        Return a string for failed results, CheckResult/ActionResult object otherwise.
+        """
+        check_method = None
+        try:
+            check_method = self._get_check_or_action_function(check_str)
+        except Exception as e:
+            return f"ERROR: {str(e)}"
+        if not isinstance(check_kwargs, dict):
+            return "ERROR: Check kwargs must be a dictionary: {check_str}"
+        return check_method(connection, **check_kwargs)
+
+    def _get_check_or_action_function(self, check_or_action_string: str, check_or_action: str = "check") -> Callable:
+        if len(check_or_action_string.strip().split('/')) != 2:
+            raise Exception(f"{check_or_action.title()} string must be of form"
+                            "module_name/{check_or_action}_function_name: {check_or_action_string}")
+        module_name = check_or_action_string.strip().split('/')[0]
+        function_name = check_or_action_string.strip().split('/')[1]
+        module = None
+        for package_name in [self.check_package_name, 'foursight_core']:
+            try:
+                module = self.import_check_module(package_name, module_name)
+            except ModuleNotFoundError:
+                continue
+            except Exception as e:
+                raise e
+        if not module:
+            raise Exception(f"Cannot find check module: {module_name}")
+        function = module.__dict__.get(function_name)
+        if not function:
+            raise Exception(f"Cannot find check function: {module_name}/{function_name}")
+        if not self.check_method_deco(function, self.CHECK_DECO) and \
+           not self.check_method_deco(function, self.ACTION_DECO):
+            raise Exception(f"{check_or_action.title()} function must use"
+                            "@{check_or_action}_function decorator: {module_name}/{function_name}")
+        return function
+
+    @staticmethod
+    def get_checks_info(search: str = None) -> list:
+        checks = []
+        registry = Decorators.get_registry()
+        for item in registry:
+            info = CheckHandler._create_check_or_action_info(registry[item])
+            if search and search not in info.qualified_name.lower():
+                continue
+            if info.is_check:
+                checks.append(info)
+        return sorted(checks, key=lambda item: item.qualified_name)
+
+    @staticmethod
+    def get_actions_info(search: str = None) -> list:
+        actions = []
+        registry = Decorators.get_registry()
+        for item in registry:
+            info = CheckHandler._create_check_or_action_info(registry[item])
+            if search and search not in info.qualified_name.lower():
+                continue
+            if info.is_action:
+                actions.append(info)
+        return sorted(actions, key=lambda item: item.qualified_name)
+
+    @staticmethod
+    def get_check_info(check_function_name: str, check_module_name: str = None) -> Optional[namedtuple]:
+        return CheckHandler._get_check_or_action_info(check_function_name, check_module_name, "check")
+
+    @staticmethod
+    def get_action_info(action_function_name: str, action_module_name: str = None) -> Optional[namedtuple]:
+        return CheckHandler._get_check_or_action_info(action_function_name, action_module_name, "action")
+
+    @staticmethod
+    def _get_check_or_action_info(function_name: str,
+                                  module_name: str = None, kind: str = None) -> Optional[namedtuple]:
+
+        function_name = function_name.strip();
+        if module_name:
+            module_name = module_name.strip();
+        if not module_name:
+            if len(function_name.split("/")) == 2:
+                module_name = function_name.split("/")[0].strip()
+                function_name = function_name.split("/")[1].strip()
+            elif len(function_name.split(".")) == 2:
+                module_name = function_name.split(".")[0].strip()
+                function_name = function_name.split(".")[1].strip()
+        registry = Decorators.get_registry()
+        for name in registry:
+            if not kind or registry[name]["kind"] == kind:
+                item = registry[name]
+                if item["name"] == function_name:
+                    if not module_name:
+                        return CheckHandler._create_check_or_action_info(item)
+                    if item["module"].endswith("." + module_name):
+                        return CheckHandler._create_check_or_action_info(item)
+
+    @staticmethod
+    def _create_check_or_action_info(info: dict) -> Optional[namedtuple]:
+
+        def unqualified_module_name(module_name: str) -> str:
+            return module_name.rsplit(".", 1)[-1] if "." in module_name else module_name
+
+        def qualified_check_or_action_name(check_or_action_name: str, module_name: str) -> str:
+            unqualified_module = unqualified_module_name(module_name)
+            return f"{unqualified_module}/{check_or_action_name}" if unqualified_module else check_or_action_name
+
+        Info = namedtuple("CheckInfo", ["kind",
+                                        "is_check",
+                                        "is_action",
+                                        "name",
+                                        "qualified_name",
+                                        "file",
+                                        "line",
+                                        "module",
+                                        "unqualified_module",
+                                        "package",
+                                        "github_url",
+                                        "args",
+                                        "kwargs",
+                                        "function",
+                                        "associated_action",
+                                        "associated_check"])
+        return Info(info["kind"],
+                    info["kind"] == "check",
+                    info["kind"] == "action",
+                    info["name"],
+                    qualified_check_or_action_name(info["name"], info["module"]),
+                    info["file"],
+                    info["line"],
+                    info["module"],
+                    unqualified_module_name(info["module"]),
+                    info["package"],
+                    info["github_url"],
+                    info["args"],
+                    info["kwargs"],
+                    info["function"],
+                    info.get("action"),
+                    info.get("check"))
+
     def init_check_or_action_res(self, connection, check):
         """
         Use in cases where a string is provided that could be a check or an action

foursight_core/decorators.py

@@ -87,10 +87,16 @@ def create_registry_record(self, kind, func, default_args, default_kwargs) -> No
             "package": func_package,
             "github_url": get_github_url(func_package, func_file, func_line),
             "args": default_args,
-            "kwargs": default_kwargs
+            "kwargs": default_kwargs,
+            "function": func
         }
         if associated_action:
             registry_record["action"] = associated_action
+        elif kind == "action":
+            for name in _decorator_registry:
+                item = _decorator_registry[name]
+                if item.get("action") == func_name:
+                    registry_record["check"] = item["name"] 
         _decorator_registry[func_name] = registry_record
 
     def check_function(self, *default_args, **default_kwargs):

foursight_core/fs_connection.py

@@ -44,7 +44,7 @@ def __init__(self, fs_environ, fs_environ_info, test=False, use_es=True, host=No
         # FOURFRONT information
         self.ff_server = fs_environ_info['fourfront']
         self.ff_env = fs_environ_info['ff_env']
-        self.ff_es = fs_environ_info['es']
+        self.ff_es = fs_environ_info['es'] if not host else host
         self.ff_bucket = fs_environ_info['bucket']
         self.redis = None
         self.redis_url = None