[Feature] Extract static analysis tests and add multi-template support (#305)

autograder/autograder.py

@@ -137,7 +137,7 @@ def build_pipeline(
     Build the AutograderPipeline object based on configuration.
 
     Args:
-        template_name: Name of the template to use
+        template_name: Name of the template to use (string or list of strings)
         include_feedback: Whether to include feedback generation
         grading_criteria: Criteria configuration dictionary
         feedback_config: Configuration for feedback generation

autograder/models/dataclass/structural_analysis_result.py

@@ -12,5 +12,9 @@ class StructuralAnalysisResult:
     Attributes:
         roots: A dictionary mapping filenames to their corresponding ast-grep root nodes.
                If a file could not be parsed, the value is None.
+        available: Whether structural analysis infrastructure was available and attempted.
+        reason: Optional reason explaining why analysis was unavailable/skipped.
     """
     roots: Dict[str, Optional['SgRoot']]
+    available: bool = True
+    reason: Optional[str] = None

autograder/models/pipeline_execution.py

@@ -99,14 +99,24 @@ def _require_step_data(self, step_name: StepName, artifact_name: str) -> Any:
             )
         return step_result.data
 
+    def get_loaded_templates(self) -> List["Template"]:
+        """
+        Retrieves the Grading Template objects loaded for this execution.
+        """
+        data = self._require_step_data(StepName.LOAD_TEMPLATE, "templates")
+        if isinstance(data, list):
+            return cast(List["Template"], data)
+        return [cast("Template", data)]
+
     def get_loaded_template(self) -> "Template":
         """
-        Retrieves the Grading Template object loaded for this execution.
+        Retrieves the first Grading Template object loaded for this execution.
+        (Maintained for backward compatibility)
         """
-        return cast(
-            "Template",
-            self._require_step_data(StepName.LOAD_TEMPLATE, "template"),
-        )
+        templates = self.get_loaded_templates()
+        if not templates:
+            raise ValueError("No templates loaded in the pipeline.")
+        return templates[0]
 
     def get_built_criteria_tree(self) -> "CriteriaTree":
         """

autograder/services/criteria_tree_service.py

@@ -27,26 +27,26 @@ class CriteriaTreeService:
 
     def __init__(self):
         self.logger = logging.getLogger("CriteriaTreeService")
-        self.__template = None
+        self.__templates: List[Template] = []
 
     def build_tree(
-        self, criteria_config: CriteriaConfig, template: Template
+        self, criteria_config: CriteriaConfig, templates: List[Template]
     ) -> CriteriaTree:
         """
         Build a complete criteria tree from validated configuration.
 
         Args:
             criteria_config: Validated criteria configuration
-            template: Template containing test functions
+            templates: List of templates containing test functions
 
         Returns:
             Complete CriteriaTree with embedded test functions
 
         Raises:
-            ValueError: If test function not found in template
+            ValueError: If test function not found in any template
         """
 
-        self.__template = template
+        self.__templates = templates
 
         base_category = self.__parse_category("base", criteria_config.base)
         tree = CriteriaTree(base_category)
@@ -88,10 +88,12 @@ def __parse_tests(self, test_configs: List[TestConfig]) -> List[TestNode]:
         return [self.__parse_test(test_item) for test_item in test_configs]
 
     def __find_test_function(self, name: str) -> Optional[TestFunction]:
-        try:
-            return self.__template.get_test(name)
-        except (AttributeError, KeyError):
-            return None
+        for template in self.__templates:
+            try:
+                return template.get_test(name)
+            except (AttributeError, KeyError):
+                continue
+        return None
 
     def __parse_test(self, config: TestConfig) -> TestNode:
         # Use technical 'type' for function lookup, falling back to 'name' for legacy support

autograder/services/grader/criteria_grader.py

@@ -139,18 +139,22 @@ def process_test(self, test: TestNode) -> TestResultNode:
             )
             test_params['program_command'] = resolved
 
-        # Inject the actual submission language so tests like forbidden_import
-        # always operate on the real language rather than a config-time guess.
-        if self.submission_language and 'submission_language' in test_params:
-            test_params['submission_language'] = self.submission_language
+        # Ensure submission_language is passed only once.
+        # Runtime language always takes precedence over config-specified language.
+        config_submission_language = test_params.pop('submission_language', None)
+        effective_submission_language = (
+            self.submission_language
+            if self.submission_language is not None
+            else config_submission_language
+        )
 
         test_result = test.test_function.execute(
             files=file_target,
             sandbox=self.sandbox,
             locale=self.locale,
             pre_computed_results=self.pre_computed_results,
             structural_analysis=self.structural_analysis,
-            submission_language=self.submission_language,
+            submission_language=effective_submission_language,
             **test_params,
         )
         return TestResultNode(

autograder/services/template_library_service.py

@@ -13,6 +13,7 @@
 from autograder.template_library.web_dev.template import WebDevTemplate
 from autograder.template_library.api_testing import ApiTestingTemplate
 from autograder.template_library.input_output import InputOutputTemplate
+from autograder.template_library.static_analysis import StaticAnalysisTemplate
 
 logger = logging.getLogger(__name__)
 
@@ -35,6 +36,7 @@ class TemplateLibraryService:
         "webdev": WebDevTemplate,
         "api": ApiTestingTemplate,
         "input_output": InputOutputTemplate,
+        "static_analysis": StaticAnalysisTemplate,
     }
 
     def __new__(cls):

autograder/steps/build_tree_step.py

@@ -43,11 +43,11 @@ def _execute(self, pipeline_exec: PipelineExecution) -> PipelineExecution:
         logger.info("Building criteria tree (external_user_id=%s)", pipeline_exec.submission.user_id)
         # Validate criteria configuration
         criteria_config = CriteriaConfig.from_dict(self._criteria_json)
-        template = pipeline_exec.get_loaded_template()
-        # Build the criteria tree with embedded test functions
+        templates = pipeline_exec.get_loaded_templates()
+        # Build the criteria tree with embedded test functions from multiple templates
         criteria_tree = self._criteria_tree_service.build_tree(
             criteria_config,
-            template
+            templates
         )
         logger.info(
             "Criteria tree built successfully (external_user_id=%s)",

autograder/steps/grade_step.py

@@ -41,13 +41,14 @@ def _execute(self, pipeline_exec: PipelineExecution) -> PipelineExecution:
         logger.info("Grading submission (external_user_id=%s)", pipeline_exec.submission.user_id)
 
         # If submission is sandboxed, feed grading template with container ref
-        template = pipeline_exec.get_loaded_template()
+        templates = pipeline_exec.get_loaded_templates()
+        requires_sandbox = any(t.requires_sandbox for t in templates)
 
         # Check if PRE_FLIGHT step was executed (only if setup_config was provided)
         sandbox = pipeline_exec.get_sandbox()
 
-        if not sandbox and template.requires_sandbox:
-            raise RuntimeError("Grading template requires a sandbox environment, but no sandbox was created")
+        if not sandbox and requires_sandbox:
+            raise RuntimeError("One or more grading templates require a sandbox environment, but no sandbox was created")
 
         criteria_tree = pipeline_exec.get_built_criteria_tree()
 

autograder/steps/load_template_step.py

@@ -1,4 +1,5 @@
 import logging
+from typing import List, Union
 
 from autograder.models.dataclass.step_result import StepResult, StepName, StepStatus
 from autograder.models.pipeline_execution import PipelineExecution
@@ -10,49 +11,77 @@
 
 class TemplateLoaderStep(Step):
     """
-    Step that loads a grading template, which contains test functions and helper code used for grading.
-    It can load either a built-in template from the library or a custom template provided by the user.
-    If the template is custom, it should be loaded in a sandboxed environment to ensure security and isolation.
+    Step that loads one or more grading templates, which contain test functions 
+    and helper code used for grading.
     """
-    def __init__(self, template_name: str, custom_template = None):
+    def __init__(self, template_name: Union[str, List[str], None], custom_template=None):
         """
         Initialize the template loader step.
         """
-        self._template_name = template_name
+        self._template_names = self._normalize_template_names(template_name)
         self._custom_template = custom_template
         self._template_service = TemplateLibraryService.get_instance()
+        if not self._custom_template and not self._template_names:
+            raise ValueError(
+                "template_name must contain at least one non-empty template identifier."
+            )
+
+    @staticmethod
+    def _normalize_template_names(template_name: Union[str, List[str], None]) -> List[str]:
+        if template_name is None:
+            return []
+
+        if isinstance(template_name, str):
+            raw_names = template_name.split(",")
+        elif isinstance(template_name, list):
+            raw_names = template_name
+        else:
+            raise ValueError(
+                "template_name must be a comma-separated string or a list of strings."
+            )
+
+        normalized: List[str] = []
+        for name in raw_names:
+            if not isinstance(name, str):
+                raise ValueError("template_name entries must be strings.")
+            stripped = name.strip()
+            if stripped:
+                normalized.append(stripped)
+        return normalized
 
     @property
     def step_name(self) -> StepName:
         return StepName.LOAD_TEMPLATE
 
     def _execute(self, pipeline_exec: PipelineExecution) -> PipelineExecution:
         """
-        Load the grading template, either built-in or custom, and return it as part of the step result.
+        Load the grading templates and return them as part of the step result.
         """
+        templates = []
+
         if self._custom_template:
             logger.info("Loading custom template (external_user_id=%s)", pipeline_exec.submission.user_id)
-            template = self._template_service.load_custom_template(self._custom_template) #TODO: Implement Custom Template Loading with Sandboxed Env
+            template = self._template_service.load_custom_template(self._custom_template)
+            templates.append(template)
         else:
-            logger.info(
-                "Loading built-in template: template=%s (external_user_id=%s)",
-                self._template_name,
-                pipeline_exec.submission.user_id,
-            )
-            template = self._template_service.load_builtin_template(self._template_name) # Load built-in template similar to custom to avoid code duplication
+            for name in self._template_names:
+                logger.info(
+                    "Loading built-in template: template=%s (external_user_id=%s)",
+                    name,
+                    pipeline_exec.submission.user_id,
+                )
+                template = self._template_service.load_builtin_template(name)
+                templates.append(template)
 
         logger.info(
-            "Template loaded successfully: template=%s (external_user_id=%s)",
-            self._template_name,
+            "Templates loaded successfully: count=%d (external_user_id=%s)",
+            len(templates),
             pipeline_exec.submission.user_id,
         )
         return pipeline_exec.add_step_result(
             StepResult(
                 step=StepName.LOAD_TEMPLATE,
-                data=template,
+                data=templates,
                 status=StepStatus.SUCCESS
             )
         )
-
-
-

autograder/steps/sandbox_step.py

@@ -36,10 +36,12 @@ def _execute(self, pipeline_exec: PipelineExecution) -> PipelineExecution:
         Returns:
             PipelineExecution with updated sandbox and step result.
         """
-        grading_template = pipeline_exec.get_loaded_template()
+        grading_templates = pipeline_exec.get_loaded_templates()
 
-        if not grading_template.requires_sandbox:
-            logger.info("Template does not require a sandbox. Skipping SandboxStep.")
+        requires_sandbox = any(t.requires_sandbox for t in grading_templates)
+
+        if not requires_sandbox:
+            logger.info("None of the templates require a sandbox. Skipping SandboxStep.")
             return pipeline_exec.add_step_result(StepResult.success(self.step_name, None))
 
         logger.info("Creating sandbox for submission (external_user_id=%s)", pipeline_exec.submission.user_id)

autograder/steps/structural_analysis_step.py

@@ -27,19 +27,46 @@ def step_name(self) -> StepName:
     def _execute(self, pipeline_exec: PipelineExecution) -> PipelineExecution:
         submission = pipeline_exec.submission
         language = submission.language
-        
+
         if not language:
             logger.warning("No language specified for submission; skipping structural analysis.")
-            return pipeline_exec.add_step_result(StepResult.success(self.step_name, StructuralAnalysisResult(roots={})))
+            return pipeline_exec.add_step_result(
+                StepResult.success(
+                    self.step_name,
+                    StructuralAnalysisResult(
+                        roots={},
+                        available=False,
+                        reason="missing_submission_language",
+                    ),
+                )
+            )
 
         if SgRoot is None:
-            logger.error("ast-grep-py is not installed; structural analysis will be skipped.")
-            return pipeline_exec.add_step_result(StepResult.fail(self.step_name, "ast-grep-py not installed"))
+            logger.warning("ast-grep-py is not installed; skipping structural analysis.")
+            return pipeline_exec.add_step_result(
+                StepResult.success(
+                    self.step_name,
+                    StructuralAnalysisResult(
+                        roots={},
+                        available=False,
+                        reason="ast_grep_unavailable",
+                    ),
+                )
+            )
 
         ast_grep_lang = self._map_language(language)
         if not ast_grep_lang:
             logger.warning(f"Language {language.value} is not supported by ast-grep; skipping.")
-            return pipeline_exec.add_step_result(StepResult.success(self.step_name, StructuralAnalysisResult(roots={})))
+            return pipeline_exec.add_step_result(
+                StepResult.success(
+                    self.step_name,
+                    StructuralAnalysisResult(
+                        roots={},
+                        available=False,
+                        reason=f"unsupported_language:{language.value}",
+                    ),
+                )
+            )
 
         roots: Dict[str, Optional[SgRoot]] = {}
         for filename, sub_file in submission.submission_files.items():
@@ -53,7 +80,7 @@ def _execute(self, pipeline_exec: PipelineExecution) -> PipelineExecution:
                 logger.warning(f"Failed to parse {filename} with ast-grep: {e}")
                 roots[filename] = None
 
-        result = StructuralAnalysisResult(roots=roots)
+        result = StructuralAnalysisResult(roots=roots, available=True)
         return pipeline_exec.add_step_result(StepResult.success(self.step_name, result))
 
     def _map_language(self, language: Language) -> Optional[str]: