feat: new parser for template tags

docs/changelog.rst

@@ -8,6 +8,12 @@ Unreleased
 +++++++++++++++++++++++++++++++++++
 
 * New rule set :doc:`rules/dag` to validate DAG templates and their tasks.
+* Redesign :doc:`rules/workflow-variable` to improve validation of Argo workflow variable references:
+
+  * Rule ``VAR001`` has been renamed to :rule:`var101`.
+  * Rule ``VAR002`` has been renamed to :rule:`var201`.
+  * New rule :rule:`var102` to flag invalid usage of expr-lang format in simple tags.
+  * New rule :rule:`var202` to flag non-Argo variable references.
 
 :octicon:`plug` Enhancements
 ++++++++++++++++++++++++++++

docs/rules/variable.rst

@@ -0,0 +1,147 @@
+Workflow Variable Rules (``VAR``)
+=================================
+
+The code ``VAR`` identifies potential issues with `workflow variables`_, including syntax errors, misused functions, and other related problems.
+
+.. _workflow variables: https://argo-workflows.readthedocs.io/en/latest/variables/
+
+.. admonition:: Known Limitations
+   :class: caution
+
+   Currently, Tugboat supports only `simple tags`_, and does not support `expression tags`_.
+
+   This means:
+
+   * ``{{ inputs.parameters.simple }}`` can be checked
+
+   * ``{{= inputs.parameters.expression }}`` will be ignored
+
+   We are planning to add support for expression tags in an upcoming release.
+
+   .. _simple tags: https://argo-workflows.readthedocs.io/en/latest/variables/#simple
+   .. _expression tags: https://argo-workflows.readthedocs.io/en/latest/variables/#expression
+
+
+.. VAR1xx syntax errors
+
+.. rule:: VAR101 Syntax error
+
+   This error occurs when a workflow variable fails to parse due to a syntax error.
+
+   Note that most parsers report syntax errors as soon as they encounter the first issue.
+   This means the error message might not always indicate the exact location of the problem, but it provides a useful starting point for debugging.
+
+.. rule:: VAR102 Incorrect template tag format
+
+   This error occurs when a reference uses expression tag syntax inside a simple template tag.
+
+   Argo Workflows supports two types of template tags:
+
+   - **Simple tags**: ``{{ inputs.parameters.foo }}``
+   - **Expression tags**: ``{{= inputs.parameters.foo }}``
+
+   The syntax for referencing values differs between these two formats.
+   Simple tags use dot notation only (e.g., ``inputs.parameters.foo``), while expression tags use `expr-lang`_ syntax, which supports both dot notation and bracket notation for member access (e.g., ``inputs.parameters['foo']`` or ``inputs["parameters"]["foo"]``).
+
+   This error is reported when expression tag syntax is mistakenly used inside a simple tag.
+
+   .. code-block:: yaml
+      :emphasize-lines: 14
+
+      apiVersion: argoproj.io/v1alpha1
+      kind: Workflow
+      metadata:
+        generateName: test-
+      spec:
+        entrypoint: whalesay
+        templates:
+          - name: whalesay
+            inputs:
+              parameters:
+                - name: message
+            container:
+              image: docker/whalesay:latest
+              args: ["{{ inputs.parameters['message'] }}"]
+
+   In the example above, ``inputs.parameters['message']`` uses bracket notation, which is valid in expression tags but not in simple tags.
+   The correct format for a simple tag is ``{{ inputs.parameters.message }}``.
+
+   .. _expr-lang: https://expr-lang.org/
+
+
+.. VAR2xx invalid references
+
+.. rule:: VAR201 Unknown Argo workflow variable reference
+
+   This error occurs when a reference is not recognized as a valid Argo workflow variable in the current context.
+
+   Tugboat validates references against a list of known Argo workflow variables based on the template type and defined inputs/outputs.
+   If a reference cannot be matched to any known variable, this error is reported.
+
+   .. code-block:: yaml
+      :emphasize-lines: 18
+
+      apiVersion: argoproj.io/v1alpha1
+      kind: Workflow
+      metadata:
+        generateName: test-
+      spec:
+        entrypoint: main
+        templates:
+          - name: main
+            inputs:
+              parameters:
+                - name: message
+            steps:
+              - - name: step-1
+                  template: produce
+                  arguments:
+                    parameters:
+                      - name: msg
+                        value: "{{ inputs.parameters.msg }}"
+              - - name: step-2
+                  template: consume
+                  arguments:
+                    parameters:
+                      - name: msg
+                        value: "{{ steps.step-1.outputs.parameters.massage }}"
+
+   In the example above, ``inputs.parameters.msg`` is reported because the defined parameter is named ``message``, not ``msg``.
+
+   When possible, Tugboat will suggest the closest matching variable name as a fix.
+
+   .. note::
+
+      Tugboat can only validate references within the same manifest.
+      References to step outputs (e.g., ``steps.step-1.outputs.parameters.message``) cannot be fully validated because the output is defined by the referenced template, which may be a WorkflowTemplate defined elsewhere.
+
+.. rule:: VAR202 Not an Argo workflow variable reference
+
+   This warning occurs when a template tag contains a single identifier that does not match any known Argo workflow variable pattern.
+
+   Argo workflow variables typically have a dotted format like ``inputs.parameters.name`` or ``workflow.name``.
+   When a tag contains only a simple variable name (e.g., ``{{ foo }}``), it is unlikely to be an Argo variable.
+
+   .. code-block:: yaml
+      :emphasize-lines: 14
+
+      apiVersion: argoproj.io/v1alpha1
+      kind: Workflow
+      metadata:
+        generateName: test-
+      spec:
+        entrypoint: whalesay
+        templates:
+          - name: whalesay
+            container:
+              image: docker/whalesay:latest
+              args: ["{{ message }}"]
+
+   In the example above, ``{{ message }}`` is flagged because ``message`` is not a valid Argo workflow variable.
+
+   This warning is commonly triggered when:
+
+   - The manifest uses another templating engine (e.g., Jinja2, Helm) that shares the ``{{ }}`` syntax
+   - There is a typo or incomplete variable reference
+
+   If the tag is intended for another templating engine, you can suppress this warning by adding a ``# noqa: VAR202`` comment. See :doc:`../violations` for more information on suppressing warnings.

pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "cloup (>=3.0.5,<4.0.0)",
     "colorlog (>=6.9.0,<7.0.0)",
     "frozendict (>=2.4.6,<3.0.0)",
+    "lark (>=1.3.1,<2.0.0)",
     "pluggy (>=1.5.0,<2.0.0)",
     "pydantic (>=2.9.2,<3.0.0)",
     "pydantic-settings (>=2.7.1,<3.0.0)",

tests/analyzers/test_template_tag.py

@@ -0,0 +1,171 @@
+from unittest.mock import Mock
+
+import lark
+import pytest
+from dirty_equals import IsPartialDict, IsStr
+
+from tests.dirty_equals import HasSubstring
+from tugboat.analyzers.template_tag import (
+    check_simple_tag_reference,
+    check_template_tags,
+    parse_argo_template_tags,
+    split_expr_membership,
+)
+from tugboat.references import ReferenceCollection
+
+
+class TestParseArgoTemplateTags:
+
+    def test_pass(self):
+        tree = parse_argo_template_tags(
+            """
+            Hello {{ inputs.parameters.name }}
+            This is {{ inputs.parameters['bot'] }} speaking.
+
+            The result is {{= 1 + 2 }}.
+            """
+        )
+
+        (input_name_tag, bot_name_tag) = tree.find_data("simple_tag")
+
+        (input_name,) = input_name_tag.find_token("REF")
+        assert input_name == "inputs.parameters.name"
+
+        (bot_name,) = bot_name_tag.find_token("REF")
+        assert bot_name == "inputs.parameters['bot']"
+
+    def test_error(self):
+        with pytest.raises(lark.exceptions.UnexpectedCharacters):
+            parse_argo_template_tags("{{ inputs. parameters.name }}")
+        with pytest.raises(lark.exceptions.UnexpectedEOF):
+            parse_argo_template_tags("{{ inputs.parameters.name ")
+
+
+class TestCheckTemplateTags:
+
+    def test_pass(self):
+        references = ReferenceCollection()
+        references.add(("inputs", "parameters", "name"))
+
+        diagnoses = list(
+            check_template_tags("{{ inputs.parameters.name }}", references)
+        )
+
+        assert diagnoses == []
+
+    def test_simple_tag_picked(self):
+        diagnoses = list(check_template_tags("{{ inputs }}", ReferenceCollection()))
+        assert diagnoses == [IsPartialDict(code="VAR202")]
+
+    def test_unexpected_character(self):
+        diagnoses = list(
+            check_template_tags(
+                "{{ inputs. parameters.name }}",
+                Mock(ReferenceCollection),
+            )
+        )
+        assert diagnoses == [
+            IsPartialDict(
+                code="VAR101",
+                summary="Syntax error",
+                msg=(
+                    IsStr()
+                    & HasSubstring("The field contains a syntax error")
+                    & HasSubstring("{{ inputs. parameters.name }}")
+                    & HasSubstring("This error is usually caused by invalid characters")
+                ),
+            )
+        ]
+
+    def test_unexpected_eof(self):
+        diagnoses = list(
+            check_template_tags(
+                "{{ inputs",
+                Mock(ReferenceCollection),
+            )
+        )
+        assert diagnoses == [
+            IsPartialDict(
+                code="VAR101",
+                summary="Syntax error",
+                msg=(
+                    IsStr()
+                    & HasSubstring("The field contains a syntax error")
+                    & HasSubstring("{{ inputs")
+                    & HasSubstring(
+                        "This error is usually caused by an incomplete template tag"
+                    )
+                ),
+            )
+        ]
+
+
+class TestCheckSimpleTagReference:
+
+    def test_pass(self):
+        references = ReferenceCollection()
+        references.add(("inputs", "parameters", "name"))
+
+        diagnosis = check_simple_tag_reference("inputs.parameters.name", references)
+        assert diagnosis is None
+
+    def test_incorrect_format(self):
+        references = ReferenceCollection()
+        references.add(("inputs", "parameters", "name"))
+
+        diagnosis = check_simple_tag_reference("inputs.parameters['name']", references)
+
+        assert diagnosis == IsPartialDict(
+            code="VAR102",
+            summary="Incorrect template tag format",
+            fix="inputs.parameters.name",
+        )
+
+    def test_syntax_error(self):
+        references = ReferenceCollection()
+        diagnosis = check_simple_tag_reference("inputs.parameters['name", references)
+        assert diagnosis == IsPartialDict(code="VAR101")
+
+    def test_not_a_argo_variable_1(self):
+        references = ReferenceCollection()
+        diagnosis = check_simple_tag_reference("inputs", references)
+        assert diagnosis == IsPartialDict(code="VAR202")
+
+    def test_unknown_variable(self):
+        references = ReferenceCollection()
+        references.add(("demo",))
+
+        diagnosis = check_simple_tag_reference("inputs", references)
+        assert diagnosis == IsPartialDict(
+            code="VAR201",
+            fix="demo",
+            ctx={
+                "reference": {
+                    "found": ("inputs",),
+                    "found:str": "inputs",
+                    "closest": ("demo",),
+                    "closest:str": "demo",
+                }
+            },
+        )
+
+
+class TestSplitExprMembership:
+
+    @pytest.mark.parametrize(
+        ("source", "expected"),
+        [
+            # success
+            ("parameters", ("parameters",)),
+            ("parameters.item", ("parameters", "item")),
+            ("parameters['item-1'].subkey", ("parameters", "item-1", "subkey")),
+            ('parameters["item-2"].subkey', ("parameters", "item-2", "subkey")),
+            ("a.b.c['d'].e['f'].g", ("a", "b", "c", "d", "e", "f", "g")),
+            # failed
+            ("0", ()),
+            ("parameters[\"item']", ()),
+        ],
+    )
+    def test(self, source: str, expected: tuple[str, ...]):
+        parts = split_expr_membership(source)
+        assert parts == expected