feat: add jinja2 templating and plugin output

checker/__main__.py

@@ -206,6 +206,7 @@ def grade(
     except Exception as e:
         print_info("UNEXPECTED ERROR", color='red')
         print_info(e)
+        raise e
         exit(1)
     print_info("TESTING PASSED", color='green')
 

checker/configs/__init__.py

@@ -7,4 +7,3 @@
 )
 from .deadlines import DeadlinesConfig, DeadlinesGroupConfig, DeadlinesSettingsConfig, DeadlinesTaskConfig  # noqa: F401
 from .task import TaskConfig  # noqa: F401
-from .utils import ParametersResolver  # noqa: F401

checker/configs/checker.py

@@ -1,12 +1,16 @@
 from __future__ import annotations
 
 from enum import Enum
+from typing import Any, Union
 
 from pydantic import AnyUrl, Field, RootModel, ValidationError, field_validator
 
 from .utils import CustomBaseModel, YamlLoaderMixin
 
-ParamType = bool | int | float | str | list[int | float | str | None] | None
+
+# Note: old Union style in definition for backward compatibility
+ParamType = Union[bool, int, float, str, list[Union[int, float, str, None]], None]
+TTemplate = Union[str, list[Union[ParamType, str]], dict[str, Union[ParamType, str]]]
 
 
 class CheckerStructureConfig(CustomBaseModel):
@@ -44,6 +48,7 @@ class CheckerManytaskConfig(CustomBaseModel):
 
 
 class PipelineStageConfig(CustomBaseModel):
+
     class FailType(Enum):
         FAST = "fast"
         AFTER_ALL = "after_all"
@@ -52,13 +57,13 @@ class FailType(Enum):
     name: str
     run: str
 
-    args: dict[str, ParamType] = Field(default_factory=dict)
+    args: dict[str, ParamType | TTemplate] = Field(default_factory=dict)
 
-    run_if: str | None = None
+    run_if: bool | TTemplate | None = None
     fail: FailType = FailType.FAST
 
-    # save a score to the context with the name `register_score`
-    register_score: str | None = None
+    # save pipline stage result to context under this key
+    register_output: str | None = None
 
 
 class CheckerTestingConfig(CustomBaseModel):
@@ -81,7 +86,7 @@ class CheckerConfig(CustomBaseModel, YamlLoaderMixin):
     """
     Checker configuration.
     :ivar version: config version
-    :ivar default_params: default parameters for task pipeline
+    :ivar default_parameters: default parameters for task pipeline
     :ivar structure: describe the structure of the repo - private/public and allowed for change files
     :ivar export: describe export (publishing to public repo)
     :ivar manytask: describe connection to manytask
@@ -90,7 +95,7 @@ class CheckerConfig(CustomBaseModel, YamlLoaderMixin):
 
     version: int
 
-    default_params: CheckerParametersConfig = Field(default_factory=dict)
+    default_parameters: CheckerParametersConfig = Field(default_factory=dict)
 
     structure: CheckerStructureConfig
     export: CheckerExportConfig

checker/configs/task.py

@@ -12,6 +12,6 @@ class TaskConfig(CustomBaseModel, YamlLoaderMixin):
     version: int  # if config exists, version is always present
 
     structure: CheckerStructureConfig | None = None
-    params: CheckerParametersConfig | None = None
+    parameters: CheckerParametersConfig | None = None
     task_pipeline: list[PipelineStageConfig] | None = None
     report_pipeline: list[PipelineStageConfig] | None = None

checker/configs/utils.py

@@ -28,83 +28,3 @@ def from_yaml(cls, path: Path) -> "YamlLoaderMixin":
     def to_yaml(self, path: Path) -> None:
         with path.open("w") as f:
             yaml.dump(self.dict(), f)
-
-
-class ParametersResolver:
-    """
-    A simple resolver that can handle expressions in the config files like ${{ some expression here }}.
-    The following syntax is supported:
-    0. you can use bool, null/None, int, float, str and flat list types
-    1. context consists of parameters from the config file and environment variables
-
-    """
-
-    pattern = re.compile(r"\${{\s*(!?\w+)\s*}}", re.DOTALL)  # "some ${{ var }} string"
-    full_pattern = re.compile(r"^\${{\s*(!?\w+)\s*}}$", re.DOTALL)  # "${{ var }}"
-
-    def __init__(self, context: dict[str, Any]):
-        self.default_context = context
-        self._validate_context(context)
-
-    def _validate_context(self, context: dict[str, Any]) -> None:
-        for value in context.values():
-            if not isinstance(value, (bool, type(None), int, float, str, list)):
-                raise BadConfig(f"Expression resolver does not support {type(value)} type of {value}")
-
-    def _evaluate_single_expression(self, variable: Any, context: dict[str, Any]) -> Any:
-        """Evaluate a single variable from the context."""
-        # TODO: fix ! - not expression
-        not_in_expression = '!' in variable
-        variable = variable.replace("!", "")
-
-        if variable in context:
-            result = context[variable]
-            if not_in_expression:
-                # Note: cast to bool for not expression
-                # raise BadConfig(f"Expression resolver does not support {type(result)} type of {result}")
-                return not result
-            else:
-                return result
-        else:
-            raise BadConfig(f"Variable '{variable}' not found in context")
-
-    def resolve(self, arguments: dict[str, Any], extra_context: dict[str, Any]) -> dict[str, Any]:
-        """
-        Resolve the arguments.
-        :param arguments: arguments with expressions to resolve,
-                          some string with placeholders e.g. "Some ${{ var }} string"
-        :param extra_context: extra context to use for resolving
-        :return: resolved expression string, resolved if only expression found or original expression if no placeholders
-
-        Examples (individual expressions), where var = 1, var2 = 2 ints:
-        "Some ${{ var }} string" -> "Some 1 string"
-        "Some ${{ var }} string ${{ var2 }}" -> "Some 1 string 2"
-        "${{ var }}" -> 1 (int type)
-        "${{ var }} " -> "1 " (cast to str type)
-        """
-        return {key: self.resolve_single_param(value, extra_context) for key, value in arguments.items()}
-
-    def resolve_single_param(self, expression: Any | list[Any], extra_context: dict[str, Any]) -> Any:
-        if isinstance(expression, list):
-            return [self.resolve_single_param(item, extra_context) for item in expression]
-        else:
-            if isinstance(expression, str):
-                return self.resolve_single_string(expression, extra_context)
-            else:
-                return expression
-
-    def resolve_single_string(self, expression: str, extra_context: dict[str, Any]) -> Any:
-        self._validate_context(extra_context)
-        context = self.default_context | extra_context
-
-        # If the entire string is one placeholder, return its actual type
-        full_match = self.full_pattern.fullmatch(expression)
-        if full_match:
-            return self._evaluate_single_expression(full_match.group(1), context)
-
-        # If not, substitute and return the result as a string
-        def substitute(match: re.Match) -> str:
-            return str(self._evaluate_single_expression(match.group(1), context))
-
-        resolved_expression = self.pattern.sub(substitute, expression)
-        return resolved_expression

checker/exceptions.py

@@ -28,28 +28,26 @@ class BadStructure(CheckerValidationError):
 
 class ExportError(CheckerException):
     """Export stage exception"""
-
     pass
 
 
 class ReportError(CheckerException):
     """Report stage exception"""
-
     pass
 
 
 class TestingError(CheckerException):
     """All testers exceptions can occur during testing stage"""
-
     pass
 
 
 @dataclass
-class ExecutionFailedError(TestingError):
-    """Execution failed exception"""
+class PluginExecutionFailed(TestingError):
+    """Exception raised when plugin execution failed"""
 
     message: str = ""
     output: str | None = None
+    percentage: float = 0.0
 
     def __repr__(self) -> str:
         return f"{self.__class__.__name__}: {self.message}"

checker/plugins/aggregate.py

@@ -2,8 +2,8 @@
 
 from typing import Literal
 
-from .base import PluginABC
-from ..exceptions import ExecutionFailedError
+from .base import PluginABC, PluginOutput
+from ..exceptions import PluginExecutionFailed
 
 
 class AggregatePlugin(PluginABC):
@@ -18,17 +18,17 @@ class Args(PluginABC.Args):
         # TODO: validate for weights: len weights should be equal to len scores
         # TODO: validate not empty scores
 
-    def _run(self, args: Args, *, verbose: bool = False) -> str:
+    def _run(self, args: Args, *, verbose: bool = False) -> PluginOutput:
         weights = args.weights or ([1.0] * len(args.scores))
 
         if len(args.scores) != len(weights):
-            raise ExecutionFailedError(
+            raise PluginExecutionFailed(
                 f"Length of scores ({len(args.scores)}) and weights ({len(weights)}) does not match",
                 output=f"Length of scores ({len(args.scores)}) and weights ({len(weights)}) does not match",
             )
 
         if len(args.scores) == 0 or len(weights) == 0:
-            raise ExecutionFailedError(
+            raise PluginExecutionFailed(
                 f"Length of scores ({len(args.scores)}) or weights ({len(weights)}) is zero",
                 output=f"Length of scores ({len(args.scores)}) or weights ({len(weights)}) is zero",
             )
@@ -47,14 +47,17 @@ def _run(self, args: Args, *, verbose: bool = False) -> str:
             from functools import reduce
             score = reduce(lambda x, y: x * y, weighted_scores)
         else:
-            raise ExecutionFailedError(
+            raise PluginExecutionFailed(
                 f"Unknown strategy {args.strategy}",
                 output=f"Unknown strategy {args.strategy}",
             )
 
-        return (
-            f"Get scores:  {args.scores}\n"
-            f"Get weights: {args.weights}\n"
-            f"Aggregate weighted scores {weighted_scores} with strategy {args.strategy}\n"
-            f"Score: {score:.2f}"
+        return PluginOutput(
+            output=(
+                f"Get scores:  {args.scores}\n"
+                f"Get weights: {args.weights}\n"
+                f"Aggregate weighted scores {weighted_scores} with strategy {args.strategy}\n"
+                f"Score: {score:.2f}"
+            ),
+            percentage=score,
         )

checker/plugins/base.py

@@ -1,11 +1,25 @@
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from typing import Any
 
 from pydantic import BaseModel, ValidationError
 
 from checker.exceptions import BadConfig, BadStructure
 
 
+@dataclass
+class PluginOutput:
+    """Plugin output dataclass.
+    :ivar output: str plugin output
+    :ivar percentage: float plugin percentage
+    """
+
+    output: str
+    percentage: float | None = None
+
+
 class PluginABC(ABC):
     """Abstract base class for plugins.
     :ivar name: str plugin name, searchable by this name
@@ -19,13 +33,13 @@ class Args(BaseModel):
         """
         pass
 
-    def run(self, args: dict[str, Any], *, verbose: bool = False) -> str:
+    def run(self, args: dict[str, Any], *, verbose: bool = False) -> PluginOutput:
         """Run the plugin.
         :param args: dict plugin arguments to pass to subclass Args
         :param verbose: if True should print teachers debug info, if False student mode
         :raises BadConfig: if plugin arguments are invalid
         :raises ExecutionFailedError: if plugin failed
-        :return: plugin output
+        :return: PluginOutput with stdout/stderr and percentage
         """
         args_obj = self.Args(**args)
 
@@ -47,12 +61,13 @@ def validate(cls, args: dict[str, Any]) -> None:
             raise BadStructure(f"Plugin {cls.name} does not implement _run method")
 
     @abstractmethod
-    def _run(self, args: Args, *, verbose: bool = False) -> str:
+    def _run(self, args: Args, *, verbose: bool = False) -> PluginOutput:
         """Actual run the plugin.
         You have to implement this method in your plugin.
         In case of failure, raise ExecutionFailedError with an error message and output.
         :param args: plugin arguments, see Args subclass
         :param verbose: if True should print teachers debug info, if False student mode
+        :return: PluginOutput with stdout/stderr and percentage
         :raises ExecutionFailedError: if plugin failed
         """
         pass

checker/plugins/gitlab.py

@@ -1,6 +1,6 @@
 from pydantic import AnyUrl
 
-from .base import PluginABC
+from .base import PluginABC, PluginOutput
 
 
 class CheckGitlabMergeRequestPlugin(PluginABC):
@@ -15,6 +15,10 @@ class Args(PluginABC.Args):
         requre_approval: bool = False
         search_for_score: bool = False
 
-    def _run(self, args: Args, *, verbose: bool = False) -> str:
+    def _run(self, args: Args, *, verbose: bool = False) -> PluginOutput:
         # TODO: implement
-        print("TODO: implement")
+        assert NotImplementedError()
+
+        return PluginOutput(
+            output="",
+        )

checker/plugins/manytask.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from .base import PluginABC
+from .base import PluginABC, PluginOutput
 
 
 class AggregatePlugin(PluginABC):
@@ -15,6 +15,10 @@ class Args(PluginABC.Args):
         task_name: str
         score: float  # TODO: validate score is in [0, 1] (bonus score?)
 
-    def _run(self, args: Args, *, verbose: bool = False) -> str:
+    def _run(self, args: Args, *, verbose: bool = False) -> PluginOutput:
         # TODO: report score to the manytask
-        return f"DRY_RUN: Report score {args.score} for task '{args.task_name}' for user '{args.username}'"
+        assert NotImplementedError()
+
+        return PluginOutput(
+            output=f"Report score {args.score} for task '{args.task_name}' for user '{args.username}'"
+        )

checker/plugins/regex.py

@@ -1,5 +1,5 @@
-from .base import PluginABC
-from ..exceptions import ExecutionFailedError
+from .base import PluginABC, PluginOutput
+from ..exceptions import PluginExecutionFailed
 
 
 class CheckRegexpsPlugin(PluginABC):
@@ -13,14 +13,14 @@ class Args(PluginABC.Args):
         regexps: list[str]
         # TODO: Add validation for patterns and regexps
 
-    def _run(self, args: Args, *, verbose: bool = False) -> str:
+    def _run(self, args: Args, *, verbose: bool = False) -> PluginOutput:
         # TODO: add verbose output with files list
         import re
         from pathlib import Path
 
         # TODO: move to Args validation
         if not Path(args.origin).exists():
-            raise ExecutionFailedError(
+            raise PluginExecutionFailed(
                 f"Origin '{args.origin}' does not exist",
                 output=f"Origin {args.origin} does not exist",
             )
@@ -33,8 +33,10 @@ def _run(self, args: Args, *, verbose: bool = False) -> str:
 
                     for regexp in args.regexps:
                         if re.search(regexp, file_content, re.MULTILINE):
-                            raise ExecutionFailedError(
+                            raise PluginExecutionFailed(
                                 f"File '{file.name}' matches regexp '{regexp}'",
                                 output=f"File '{file}' matches regexp '{regexp}'",
                             )
-        return "No forbidden regexps found"
+        return PluginOutput(
+            output="No forbidden regexps found",
+        )