Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 39 additions & 1 deletion backend/openedx_ai_extensions/admin.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,12 @@

from openedx_ai_extensions.models import PromptTemplate
from openedx_ai_extensions.workflows.models import AIWorkflowProfile, AIWorkflowScope, AIWorkflowSession
from openedx_ai_extensions.workflows.template_utils import discover_templates, parse_json5_string
from openedx_ai_extensions.workflows.template_utils import (
discover_templates,
get_effective_config,
parse_json5_string,
validate_workflow_config,
)


@admin.register(PromptTemplate)
Expand Down Expand Up @@ -126,6 +131,39 @@ def clean_content_patch(self):

return content_patch_raw

def clean(self):
"""Validate the effective configuration after merging base template with patch."""
cleaned_data = super().clean()

base_filepath = cleaned_data.get('base_filepath')
content_patch_raw = cleaned_data.get('content_patch', '')

if not base_filepath:
return cleaned_data

# Parse the content patch
content_patch = {}
if content_patch_raw and content_patch_raw.strip():
try:
content_patch = parse_json5_string(content_patch_raw)
except Exception: # pylint: disable=broad-exception-caught
# Already caught in clean_content_patch
return cleaned_data

# Get effective config and validate it
effective_config = get_effective_config(base_filepath, content_patch)
if effective_config is None:
return cleaned_data

is_valid, errors = validate_workflow_config(effective_config)
if not is_valid:
raise ValidationError(
'Effective configuration is invalid: %(errors)s',
params={'errors': '; '.join(errors)},
)

return cleaned_data


@admin.register(AIWorkflowProfile)
class AIWorkflowProfileAdmin(admin.ModelAdmin):
Expand Down
18 changes: 18 additions & 0 deletions backend/openedx_ai_extensions/workflows/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,24 @@ def processor_config(self) -> dict:
return {}
return self.config.get("processor_config", {})

def clean(self):
"""Validate the effective configuration before saving."""
super().clean()
effective_config = get_effective_config(self.base_filepath, self.content_patch_dict)
if effective_config is not None:
is_valid, errors = validate_workflow_config(effective_config)
if not is_valid:
raise ValidationError({
"content_patch": errors,
})

def save(self, *args, **kwargs):
"""Override save to validate and clear cached config."""
self.full_clean()
# Invalidate cached_property so it's recomputed after save
self.__dict__.pop("config", None)
super().save(*args, **kwargs)


class AIWorkflowScope(models.Model):
"""
Expand Down
38 changes: 38 additions & 0 deletions backend/openedx_ai_extensions/workflows/template_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -313,6 +313,9 @@ def _validate_semantics(config: dict) -> list[str]:
if not isinstance(processor_value, dict):
errors.append(f"processor_config.{processor_name} must be an object")

# Validate that prompt_template references exist in the database
errors.extend(_validate_prompt_templates(processor_config))

# Check actuator_config structure (required by schema 1.0)
actuator_config = config.get("actuator_config", {})
if not isinstance(actuator_config, dict):
Expand All @@ -335,6 +338,41 @@ def _validate_semantics(config: dict) -> list[str]:
return errors


def _validate_prompt_templates(processor_config: dict) -> list[str]:
"""
Validate that all prompt_template references in processor configs exist in the database.

Checks each processor's configuration for a ``prompt_template`` field
and verifies that the referenced PromptTemplate exists (by slug or UUID).

Args:
processor_config: The processor_config dict from the workflow configuration.

Returns:
List of error messages for any missing prompt templates.
"""
from openedx_ai_extensions.models import PromptTemplate # pylint: disable=import-outside-toplevel

errors = []

for processor_name, processor_value in processor_config.items():
if not isinstance(processor_value, dict):
continue

template_id = processor_value.get("prompt_template")
if not template_id:
continue

prompt = PromptTemplate.load_prompt(template_id)
if prompt is None:
errors.append(
f"processor_config.{processor_name}.prompt_template: "
f"PromptTemplate '{template_id}' does not exist"
)

return errors


def get_effective_config(base_filepath: str, content_patch: dict) -> Optional[dict]:
"""
Get the effective configuration by merging base template with patch.
Expand Down
210 changes: 210 additions & 0 deletions backend/tests/test_template_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,10 @@

from django.test import TestCase, override_settings

from openedx_ai_extensions.models import PromptTemplate
from openedx_ai_extensions.workflows.template_utils import (
WORKFLOW_SCHEMA,
_validate_prompt_templates,
_validate_semantics,
discover_templates,
get_effective_config,
Expand Down Expand Up @@ -640,6 +642,139 @@ def test_invalid_identifier_special_chars(self):
self.assertGreater(len(errors), 0)


class TestValidatePromptTemplates(TestCase):
"""Tests for _validate_prompt_templates function."""

def test_no_prompt_template_no_errors(self):
"""Test that configs without prompt_template produce no errors."""
processor_config = {
"LLMProcessor": {"function": "summarize_content", "provider": "default"},
"OpenEdXProcessor": {"function": "get_location_content"},
}
errors = _validate_prompt_templates(processor_config)
self.assertEqual(errors, [])

def test_existing_slug_no_errors(self):
"""Test that referencing an existing prompt template by slug produces no errors."""
template = PromptTemplate.objects.create(
slug="test-existing-slug",
body="You are a helpful AI assistant."
)
processor_config = {
"LLMProcessor": {"prompt_template": template.slug},
}
errors = _validate_prompt_templates(processor_config)
self.assertEqual(errors, [])

def test_existing_uuid_no_errors(self):
"""Test that referencing an existing prompt template by UUID produces no errors."""
template = PromptTemplate.objects.create(
slug="test-existing-uuid",
body="You are a helpful AI assistant."
)
processor_config = {
"LLMProcessor": {"prompt_template": str(template.id)},
}
errors = _validate_prompt_templates(processor_config)
self.assertEqual(errors, [])

def test_nonexistent_slug_produces_error(self):
"""Test that referencing a nonexistent slug produces a validation error."""
processor_config = {
"LLMProcessor": {"prompt_template": "slug-que-no-existe"},
}
errors = _validate_prompt_templates(processor_config)
self.assertEqual(len(errors), 1)
self.assertIn("slug-que-no-existe", errors[0])
self.assertIn("does not exist", errors[0])
self.assertIn("LLMProcessor", errors[0])

def test_nonexistent_uuid_produces_error(self):
"""Test that referencing a nonexistent UUID produces a validation error."""
processor_config = {
"LLMProcessor": {"prompt_template": "12345678-1234-1234-1234-123456789abc"},
}
errors = _validate_prompt_templates(processor_config)
self.assertEqual(len(errors), 1)
self.assertIn("12345678-1234-1234-1234-123456789abc", errors[0])
self.assertIn("does not exist", errors[0])

def test_multiple_processors_with_invalid_templates(self):
"""Test that errors are reported for each processor with an invalid prompt_template."""
processor_config = {
"LLMProcessor": {"prompt_template": "nonexistent-1"},
"EducatorAssistantProcessor": {"prompt_template": "nonexistent-2"},
}
errors = _validate_prompt_templates(processor_config)
self.assertEqual(len(errors), 2)
self.assertTrue(any("nonexistent-1" in e for e in errors))
self.assertTrue(any("nonexistent-2" in e for e in errors))

def test_mixed_valid_and_invalid_templates(self):
"""Test mix of valid and invalid prompt_template references."""
template = PromptTemplate.objects.create(
slug="valid-template",
body="A valid prompt."
)
processor_config = {
"LLMProcessor": {"prompt_template": template.slug},
"EducatorAssistantProcessor": {"prompt_template": "invalid-template"},
}
errors = _validate_prompt_templates(processor_config)
self.assertEqual(len(errors), 1)
self.assertIn("invalid-template", errors[0])
self.assertIn("EducatorAssistantProcessor", errors[0])

def test_empty_prompt_template_no_errors(self):
"""Test that empty string or null prompt_template is skipped (no error)."""
processor_config = {
"LLMProcessor": {"prompt_template": ""},
"OpenEdXProcessor": {"prompt_template": None},
}
errors = _validate_prompt_templates(processor_config)
self.assertEqual(errors, [])

def test_non_dict_processor_value_skipped(self):
"""Test that non-dict processor values are skipped without error."""
processor_config = {
"LLMProcessor": "not a dict",
}
errors = _validate_prompt_templates(processor_config)
self.assertEqual(errors, [])

def test_full_validation_catches_nonexistent_prompt_template(self):
"""Test that validate_workflow_config catches nonexistent prompt_template (integration)."""
config = {
"schema_version": "1.0",
"orchestrator_class": "DirectLLMResponse",
"processor_config": {
"LLMProcessor": {"prompt_template": "does-not-exist"},
},
"actuator_config": {"UIComponents": {"request": {}, "response": {}}},
}
is_valid, errors = validate_workflow_config(config)
self.assertFalse(is_valid)
self.assertTrue(any("does-not-exist" in e for e in errors))

def test_full_validation_passes_with_existing_prompt_template(self):
"""Test that validate_workflow_config passes with an existing prompt_template."""
PromptTemplate.objects.create(
slug="real-template",
body="This prompt exists."
)
config = {
"schema_version": "1.0",
"orchestrator_class": "DirectLLMResponse",
"processor_config": {
"LLMProcessor": {"prompt_template": "real-template"},
},
"actuator_config": {"UIComponents": {"request": {}, "response": {}}},
}
is_valid, errors = validate_workflow_config(config)
self.assertTrue(is_valid)
self.assertEqual(errors, [])


class TestGetEffectiveConfig(TestCase):
"""Tests for get_effective_config function."""

Expand Down Expand Up @@ -827,3 +962,78 @@ def test_ui_components_requires_request_and_response(self):
# Both must be objects
self.assertEqual(ui_components_props["properties"]["request"]["type"], "object")
self.assertEqual(ui_components_props["properties"]["response"]["type"], "object")


class TestAdminFormPromptTemplateValidation(TestCase):
"""Tests that the Admin form blocks saving when prompt_template references are invalid."""

def setUp(self):
"""Set up a temp directory with a valid base template."""
self.tmpdir = tempfile.mkdtemp()
self.temp_path = Path(self.tmpdir)

base_template = self.temp_path / "base_valid.json"
base_template.write_text('''{
"schema_version": "1.0",
"orchestrator_class": "DirectLLMResponse",
"processor_config": {
"LLMProcessor": {"provider": "default"}
},
"actuator_config": {
"UIComponents": {
"request": {"component": "AIRequestComponent"},
"response": {"component": "AIResponseComponent"}
}
}
}''')

def tearDown(self):
"""Clean up temporary directory."""
shutil.rmtree(self.tmpdir)

def test_admin_form_rejects_nonexistent_prompt_template(self):
"""Test that the admin form raises ValidationError for nonexistent prompt_template."""
from openedx_ai_extensions.admin import AIWorkflowProfileAdminForm # pylint: disable=import-outside-toplevel

form_data = {
'slug': 'test-profile',
'base_filepath': 'base_valid.json',
'content_patch': '{"processor_config": {"LLMProcessor": {"prompt_template": "no-existe"}}}',
}

with override_settings(WORKFLOW_TEMPLATE_DIRS=[self.tmpdir]):
form = AIWorkflowProfileAdminForm(data=form_data)
self.assertFalse(form.is_valid())
# The error should mention the nonexistent prompt template
all_errors = str(form.errors)
self.assertIn("no-existe", all_errors)

def test_admin_form_accepts_existing_prompt_template(self):
"""Test that the admin form accepts a valid prompt_template reference."""
from openedx_ai_extensions.admin import AIWorkflowProfileAdminForm # pylint: disable=import-outside-toplevel

PromptTemplate.objects.create(slug="valid-admin-template", body="A real prompt.")

form_data = {
'slug': 'test-profile-valid',
'base_filepath': 'base_valid.json',
'content_patch': '{"processor_config": {"LLMProcessor": {"prompt_template": "valid-admin-template"}}}',
}

with override_settings(WORKFLOW_TEMPLATE_DIRS=[self.tmpdir]):
form = AIWorkflowProfileAdminForm(data=form_data)
self.assertTrue(form.is_valid(), f"Form errors: {form.errors}")

def test_admin_form_accepts_no_prompt_template(self):
"""Test that the admin form accepts config without prompt_template."""
from openedx_ai_extensions.admin import AIWorkflowProfileAdminForm # pylint: disable=import-outside-toplevel

form_data = {
'slug': 'test-profile-no-pt',
'base_filepath': 'base_valid.json',
'content_patch': '',
}

with override_settings(WORKFLOW_TEMPLATE_DIRS=[self.tmpdir]):
form = AIWorkflowProfileAdminForm(data=form_data)
self.assertTrue(form.is_valid(), f"Form errors: {form.errors}")