Expanded cfengine lint with more linting rules and the ability to specify files / folders

src/cfengine_cli/commands.py

@@ -1,6 +1,7 @@
 import sys
 import os
 import re
+import itertools
 import json
 from cfengine_cli.profile import profile_cfengine, generate_callstack
 from cfengine_cli.dev import dispatch_dev_subcommand
@@ -94,26 +95,63 @@ def format(names, line_length) -> int:
     return 0
 
 
-def lint() -> int:
+def _lint_folder(folder):
     errors = 0
-    for filename in find(".", extension=".json"):
-        if filename.startswith(("./.", "./out/")):
+    while folder.endswith(("/.", "/")):
+        folder = folder[0:-1]
+    for filename in itertools.chain(
+        find(folder, extension=".json"), find(folder, extension=".cf")
+    ):
+        if filename.startswith(("./.", "./out/", folder + "/.", folder + "/out/")):
             continue
-        if filename.endswith("/cfbs.json"):
-            lint_cfbs_json(filename)
+        if filename.startswith(".") and not filename.startswith("./"):
             continue
-        errors += lint_json(filename)
+        errors += _lint_single_file(filename)
+    return errors
 
-    for filename in find(".", extension=".cf"):
-        if filename.startswith(("./.", "./out/")):
-            continue
-        errors += lint_policy_file(filename)
+
+def _lint_single_file(file):
+    assert os.path.isfile(file)
+    if file.endswith("/cfbs.json"):
+        return lint_cfbs_json(file)
+    if file.endswith(".json"):
+        return lint_json(file)
+    assert file.endswith(".cf")
+    return lint_policy_file(file)
+
+
+def _lint_single_arg(arg):
+    if os.path.isdir(arg):
+        return _lint_folder(arg)
+    assert os.path.isfile(arg)
+    _lint_single_file(arg)
+    return 0
+
+
+def _lint(files) -> int:
+
+    if not files:
+        return _lint_folder(".")
+
+    errors = 0
+
+    for file in files:
+        errors += _lint_single_arg(file)
 
     if errors == 0:
         return 0
     return 1
 
 
+def lint(files) -> int:
+    errors = _lint(files)
+    if errors == 0:
+        print("Success, no errors found.")
+    else:
+        print(f"Failure, {errors} errors in total.")
+    return errors
+
+
 def report() -> int:
     _require_cfhub()
     _require_cfagent()

src/cfengine_cli/lint.py

@@ -17,6 +17,9 @@
 from cfbs.validate import validate_config
 from cfbs.cfbs_config import CFBSConfig
 
+DEPRECATED_PROMISE_TYPES = ["defaults", "guest_environments"]
+ALLOWED_BUNDLE_TYPES = ["agent", "common", "monitor", "server", "edit_line"]
+
 
 def lint_cfbs_json(filename) -> int:
     assert os.path.isfile(filename)
@@ -72,34 +75,94 @@ def _text(node):
     return node.text.decode()
 
 
-def _walk(filename, lines, node) -> int:
+def _walk_generic(filename, lines, node, visitor):
+    visitor(node)
+    for node in node.children:
+        _walk_generic(filename, lines, node, visitor)
+
+
+def _find_node_type(filename, lines, node, node_type):
+    matches = []
+    visitor = lambda x: matches.extend([x] if x.type == node_type else [])
+    _walk_generic(filename, lines, node, visitor)
+    return matches
+
+
+def _find_nodes(filename, lines, node):
+    matches = []
+    visitor = lambda x: matches.append(x)
+    _walk_generic(filename, lines, node, visitor)
+    return matches
+
+
+def _single_node_checks(filename, lines, node):
+    """Things which can be checked by only looking at one node,
+    not needing to recurse into children."""
     line = node.range.start_point[0] + 1
-    column = node.range.start_point[1]
-    errors = 0
-    # Checking for syntax errors (already detected by parser / grammar).
-    # These are represented in the syntax tree as special ERROR nodes.
-    if node.type == "ERROR":
+    column = node.range.start_point[1] + 1
+    if node.type == "attribute_name" and _text(node) == "ifvarclass":
         _highlight_range(node, lines)
-        print(f"Error: Syntax error at {filename}:{line}:{column}")
-        errors += 1
-
-    if node.type == "attribute_name":
-        if _text(node) == "ifvarclass":
+        print(
+            f"Deprecation: Use 'if' instead of 'ifvarclass' at {filename}:{line}:{column}"
+        )
+        return 1
+    if node.type == "promise_guard":
+        assert _text(node) and len(_text(node)) > 1 and _text(node)[-1] == ":"
+        promise_type = _text(node)[0:-1]
+        if promise_type in DEPRECATED_PROMISE_TYPES:
+            _highlight_range(node, lines)
+            print(
+                f"Deprecation: Promise type '{promise_type}' is deprecated at {filename}:{line}:{column}"
+            )
+            return 1
+    if node.type == "bundle_block_name":
+        if _text(node) != _text(node).lower():
+            _highlight_range(node, lines)
+            print(
+                f"Convention: Bundle name should be lowercase at {filename}:{line}:{column}"
+            )
+            return 1
+    if node.type == "promise_block_name":
+        if _text(node) != _text(node).lower():
             _highlight_range(node, lines)
             print(
-                f"Error: Use 'if' instead of 'ifvarclass' (deprecated) at {filename}:{line}:{column}"
+                f"Convention: Promise type should be lowercase at {filename}:{line}:{column}"
             )
-            errors += 1
+            return 1
+    if node.type == "bundle_block_type":
+        if _text(node) not in ALLOWED_BUNDLE_TYPES:
+            _highlight_range(node, lines)
+            print(
+                f"Error: Bundle type must be one of ({', '.join(ALLOWED_BUNDLE_TYPES)}), not '{_text(node)}' at {filename}:{line}:{column}"
+            )
+            return 1
+    return 0
 
-    for node in node.children:
-        errors += _walk(filename, lines, node)
+
+def _walk(filename, lines, node) -> int:
+    error_nodes = _find_node_type(filename, lines, node, "ERROR")
+    if error_nodes:
+        for node in error_nodes:
+            line = node.range.start_point[0] + 1
+            column = node.range.start_point[1] + 1
+            _highlight_range(node, lines)
+            print(f"Error: Syntax error at {filename}:{line}:{column}")
+        return len(error_nodes)
+
+    line = node.range.start_point[0] + 1
+    column = node.range.start_point[1] + 1
+
+    errors = 0
+    for node in _find_nodes(filename, lines, node):
+        errors += _single_node_checks(filename, lines, node)
 
     return errors
 
 
 def lint_policy_file(
     filename, original_filename=None, original_line=None, snippet=None, prefix=None
 ):
+    print(f"Linting: {filename}")
     assert original_filename is None or type(original_filename) is str
     assert original_line is None or type(original_line) is int
     assert snippet is None or type(snippet) is int

src/cfengine_cli/main.py

@@ -48,10 +48,11 @@ def _get_arg_parser():
     fmt = subp.add_parser("format", help="Autoformat .json and .cf files")
     fmt.add_argument("files", nargs="*", help="Files to format")
     fmt.add_argument("--line-length", default=80, type=int, help="Maximum line length")
-    subp.add_parser(
+    lnt = subp.add_parser(
         "lint",
         help="Look for syntax errors and other simple mistakes",
     )
+    lnt.add_argument("files", nargs="*", help="Files to format")
     subp.add_parser(
         "report",
         help="Run the agent and hub commands necessary to get new reporting data",
@@ -131,7 +132,7 @@ def run_command_with_args(args) -> int:
     if args.command == "format":
         return commands.format(args.files, args.line_length)
     if args.command == "lint":
-        return commands.lint()
+        return commands.lint(args.files)
     if args.command == "report":
         return commands.report()
     if args.command == "run":