Provide even more context to the lint context

lib/galaxy/tool_util/cwl/schema.py

@@ -41,12 +41,6 @@ def raw_process_reference(self, path, loading_context=None):
             suffix = ''
             if '#' in path:
                 path, suffix = path.split('#')
-            print(f""" -------
-
-{open(path).read()}
-
-            -------
-            """)
             processed_path = os.path.join(output_dir, os.path.basename(path))
             path = os.path.abspath(path)
             uri = f"file://{path}"
@@ -56,12 +50,6 @@ def raw_process_reference(self, path, loading_context=None):
                 exit_code = cwl_expression_refactor.main([output_dir, path, '--skip-some1', '--skip-some2'])
                 if exit_code == 0:
                     uri = f"file://{processed_path}"
-                    print(f""" -------
-
-{open(processed_path).read()}
-
-            -------
-                    """)
             if suffix:
                 uri = f"{uri}#{suffix}"
             loading_context, process_object, uri = load_tool.fetch_document(uri, loadingContext=loading_context)

lib/galaxy/tool_util/lint.py

@@ -12,15 +12,36 @@
 
 
 def lint_tool_source(tool_source, level=LEVEL_ALL, fail_level=LEVEL_WARN, extra_modules=None, skip_types=None, name=None):
+    """
+    apply all (applicable) linters from the linters submodule
+    and the ones in extramodules
+
+    immediately print linter messages (wrt level) and return if linting failed (wrt fail_level)
+    """
     extra_modules = extra_modules or []
     skip_types = skip_types or []
     lint_context = LintContext(level=level, skip_types=skip_types, object_name=name)
     lint_tool_source_with(lint_context, tool_source, extra_modules)
-
     return not lint_context.failed(fail_level)
 
 
+def lint_context_for_tool_source(tool_source, extra_modules=None, skip_types=None, name=None):
+    """
+    this is the silent variant of lint_tool_source
+    it returns the LintContext from which all linter messages
+    and the status can be obtained
+    """
+    extra_modules = extra_modules or []
+    skip_types = skip_types or []
+    lint_context = LintContext(level="all", skip_types=skip_types, object_name=name)
+    lint_tool_source_with(lint_context, tool_source, extra_modules, silent=True)
+    return lint_context
+
+
 def lint_xml(tool_xml, level=LEVEL_ALL, fail_level=LEVEL_WARN, extra_modules=None, skip_types=None, name=None):
+    """
+    silently lint an xml tool
+    """
     extra_modules = extra_modules or []
     skip_types = skip_types or []
     lint_context = LintContext(level=level, skip_types=skip_types, object_name=name)
@@ -29,7 +50,7 @@ def lint_xml(tool_xml, level=LEVEL_ALL, fail_level=LEVEL_WARN, extra_modules=Non
     return not lint_context.failed(fail_level)
 
 
-def lint_tool_source_with(lint_context, tool_source, extra_modules=None):
+def lint_tool_source_with(lint_context, tool_source, extra_modules=None, silent=False):
     extra_modules = extra_modules or []
     import galaxy.tool_util.linters
     tool_xml = getattr(tool_source, "xml_tree", None)
@@ -52,31 +73,45 @@ def lint_tool_source_with(lint_context, tool_source, extra_modules=None):
                         # XML linter and non-XML tool, skip for now
                         continue
                     else:
-                        lint_context.lint(name, value, tool_xml)
+                        lint_context.lint(name, value, tool_xml, silent)
                 else:
-                    lint_context.lint(name, value, tool_source)
+                    lint_context.lint(name, value, tool_source, silent)
+    return lint_context
 
 
 def lint_xml_with(lint_context, tool_xml, extra_modules=None):
     extra_modules = extra_modules or []
     tool_source = get_tool_source(xml_tree=tool_xml)
-    return lint_tool_source_with(lint_context, tool_source, extra_modules=extra_modules)
+    lint_tool_source_with(lint_context, tool_source, extra_modules=extra_modules)
 
 
 class LintMessage:
-    def __init__(self, level, message, line, xpath=None):
+    """
+    a message from the linter
+    """
+    def __init__(self, level, message, line, fname, xpath=None):
         self.level = level
         self.message = message
-        self.line = line
+        self.line = line  # 1 based
+        self.fname = fname
         self.xpath = xpath
 
     def __str__(self):
-        rval = f".. {self.level.upper()}: {self.message}"
-        if self.line is not None:
-            rval += f" (line {self.line})"
-        if self.xpath is not None:
+        return f".. {self.level.upper()}: {self.message}"
+
+    def pretty_print(self, level=True, fname=True, xpath=False):
+        rval = ""
+        if level:
+            rval += f".. {self.level.upper()}: "
+        rval += "{self.message}"
+        if fname and self.line is not None:
+            rval += " ("
+            if self.fname:
+                rval += f"{self.fname}:"
+            rval += str(self.line)
+            rval += ")"
+        if xpath and self.xpath is not None:
             rval += f" [{self.xpath}]"
-
         return rval
 
 
@@ -89,59 +124,67 @@ def __init__(self, level, skip_types=None, object_name=None):
         self.skip_types = skip_types or []
         self.level = level
         self.object_name = object_name
-        self.found_errors = False
-        self.found_warns = False
+        self.message_list = []
+
+    @property
+    def found_errors(self):
+        return len(self.error_messages) > 0
 
-    def lint(self, name, lint_func, lint_target):
+    @property
+    def found_warns(self):
+        return len(self.warn_messages) > 0
+
+    def lint(self, name, lint_func, lint_target, silent=False):
         name = name.replace("tsts", "tests")[len("lint_"):]
         if name in self.skip_types:
             return
-        self.printed_linter_info = False
-        self.message_list = []
+
+        if not silent:
+            # this is a relict from the past where the lint context
+            # was reset when called with a new lint_func, as workaround
+            # we safe the message list, apply the lint_func (which then
+            # adds to the message_list) and restore the message list
+            # at the end (+ append the new messages)
+            tmp_message_list = list(self.message_list)
+            self.message_list = []
 
         # call linter
         lint_func(lint_target, self)
-        # TODO: colorful emoji if in click CLI.
-        if self.error_messages:
-            status = "FAIL"
-        elif self.warn_messages:
-            status = "WARNING"
-        else:
-            status = "CHECK"
-
-        def print_linter_info():
-            if self.printed_linter_info:
-                return
-            self.printed_linter_info = True
-            print(f"Applying linter {name}... {status}")
-
-        for message in self.message_list:
-            if message.level != "error":
-                continue
-            self.found_errors = True
-            print_linter_info()
-            print(f"{message}")
-
-        if self.level != LEVEL_ERROR:
-            for message in self.message_list:
-                if message.level != "warning":
-                    continue
-                self.found_warns = True
-                print_linter_info()
-                print(f"{message}")
 
-        if self.level == LEVEL_ALL:
-            for message in self.message_list:
-                if message.level != "info":
-                    continue
-                print_linter_info()
-                print(f"{message}")
-            for message in self.message_list:
-                if message.level != "check":
-                    continue
-                print_linter_info()
+        if not silent:
+            # TODO: colorful emoji if in click CLI.
+            if self.error_messages:
+                status = "FAIL"
+            elif self.warn_messages:
+                status = "WARNING"
+            else:
+                status = "CHECK"
+
+            def print_linter_info(printed_linter_info):
+                if printed_linter_info:
+                    return True
+                print(f"Applying linter {name}... {status}")
+                return True
+
+            plf = False
+            for message in self.error_messages:
+                plf = print_linter_info(plf)
                 print(f"{message}")
 
+            if self.level != LEVEL_ERROR:
+                for message in self.warn_messages:
+                    plf = print_linter_info(plf)
+                    print(f"{message}")
+
+            if self.level == LEVEL_ALL:
+                for message in self.info_messages:
+                    plf = print_linter_info(plf)
+                    print(f"{message}")
+                for message in self.valid_messages:
+                    plf = print_linter_info(plf)
+                    print(f"{message}")
+            self.message_list = tmp_message_list + self.message_list
+
     @property
     def valid_messages(self):
         return [x.message for x in self.message_list if x.level == "check"]
@@ -158,22 +201,23 @@ def warn_messages(self):
     def error_messages(self):
         return [x.message for x in self.message_list if x.level == "error"]
 
-    def __handle_message(self, level, message, line, xpath, *args):
+    def __handle_message(self, level, message, line, fname, xpath, *args):
         if args:
             message = message % args
-        self.message_list.append(LintMessage(level=level, message=message, line=line, xpath=xpath))
+        self.message_list.append(LintMessage(level=level, message=message,
+                                 line=line, fname=fname, xpath=xpath))
 
-    def valid(self, message, line=None, xpath=None, *args):
-        self.__handle_message("check", message, line, xpath, *args)
+    def valid(self, message, line=None, fname=None, xpath=None, *args):
+        self.__handle_message("check", message, line, fname, xpath, *args)
 
-    def info(self, message, line=None, xpath=None, *args):
-        self.__handle_message("info", message, line, xpath, *args)
+    def info(self, message, line=None, fname=None, xpath=None, *args):
+        self.__handle_message("info", message, line, fname, xpath, *args)
 
-    def error(self, message, line=None, xpath=None, *args):
-        self.__handle_message("error", message, line, xpath, *args)
+    def error(self, message, line=None, fname=None, xpath=None, *args):
+        self.__handle_message("error", message, line, fname, xpath, *args)
 
-    def warn(self, message, line=None, xpath=None, *args):
-        self.__handle_message("warning", message, line, xpath, *args)
+    def warn(self, message, line=None, fname=None, xpath=None, *args):
+        self.__handle_message("warning", message, line, fname, xpath, *args)
 
     def failed(self, fail_level):
         found_warns = self.found_warns

lib/galaxy/tool_util/linters/_util.py

@@ -1,6 +1,19 @@
 import re
 
 
+def node_props_factory(tool_xml):
+
+    def node_props(node):
+        if node is None:
+            return {"line": 0, "fname": None, "xpath": None}
+        else:
+            return {"line": node.sourceline,
+                    "fname": node.base,
+                    "xpath": tool_xml.getpath(node)}
+
+    return node_props
+
+
 def is_datasource(tool_xml):
     """Returns true if the tool is a datasource tool"""
     return tool_xml.getroot().attrib.get('tool_type', '') == 'data_source'

lib/galaxy/tool_util/linters/citations.py

@@ -3,41 +3,37 @@
 Citations describe references that should be used when consumers
 of the tool publish results.
 """
+from ._util import node_props_factory
 
 
 def lint_citations(tool_xml, lint_ctx):
     """Ensure tool contains at least one valid citation."""
     root = tool_xml.getroot()
-    if root is not None:
-        root_line = root.sourceline
-        root_xpath = tool_xml.getpath(root)
-    else:
-        root_line = 1
-        root_xpath = None
+    node_props = node_props_factory(tool_xml)
     citations = root.findall("citations")
     if len(citations) > 1:
-        lint_ctx.error("More than one citation section found, behavior undefined.", line=citations[1].sourceline, xpath=tool_xml.getpath(citations[1]))
+        lint_ctx.error("More than one citation section found, behavior undefined.", **node_props(citations[1]))
         return
 
     if len(citations) == 0:
-        lint_ctx.warn("No citations found, consider adding citations to your tool.", line=root_line, xpath=root_xpath)
+        lint_ctx.warn("No citations found, consider adding citations to your tool.", **node_props(root))
         return
 
     valid_citations = 0
     for citation in citations[0]:
         if citation.tag != "citation":
-            lint_ctx.warn(f"Unknown tag discovered in citations block [{citation.tag}], will be ignored.", line=citation.sourceline, xpath=tool_xml.getpath(citation))
+            lint_ctx.warn(f"Unknown tag discovered in citations block [{citation.tag}], will be ignored.", **node_props(citation))
             continue
         citation_type = citation.attrib.get("type")
         if citation_type not in ('bibtex', 'doi'):
-            lint_ctx.warn(f"Unknown citation type discovered [{citation_type}], will be ignored.", line=citation.sourceline, xpath=tool_xml.getpath(citation))
+            lint_ctx.warn(f"Unknown citation type discovered [{citation_type}], will be ignored.", **node_props(citation))
             continue
         if citation.text is None or not citation.text.strip():
-            lint_ctx.error(f'Empty {citation_type} citation.', line=citation.sourceline, xpath=tool_xml.getpath(citation))
+            lint_ctx.error(f'Empty {citation_type} citation.', **node_props(citation))
             continue
         valid_citations += 1
 
     if valid_citations > 0:
-        lint_ctx.valid(f"Found {valid_citations} likely valid citations.", line=root_line, xpath=root_xpath)
+        lint_ctx.valid(f"Found {valid_citations} likely valid citations.", **node_props(root))
     else:
-        lint_ctx.warn("Found no valid citations.", line=root_line, xpath=root_xpath)
+        lint_ctx.warn("Found no valid citations.", **node_props(root))