refactor: centralize test and test set type validation

apps/backend/src/rhesis/backend/alembic/versions/1776e6dd47d3_add_model_type_column_to_model_table.py

@@ -1,8 +1,7 @@
-from alembic import op
-import sqlalchemy as sa
-from typing import Union, Sequence
-import rhesis
+from typing import Sequence, Union
 
+import sqlalchemy as sa
+from alembic import op
 
 # revision identifiers, used by Alembic.
 revision: str = "1776e6dd47d3"

apps/backend/src/rhesis/backend/app/constants.py

@@ -34,8 +34,39 @@ def get_value(cls, entity_type):
         return entity_type
 
 
+# TestType Enum - DB-level test classification aligned with initial_data.json type_lookup
+class TestType(str, Enum):
+    """
+    Enum for test types.
+
+    These are reserved test type values in the TypeLookup table:
+    - SINGLE_TURN: Traditional single request-response tests
+    - MULTI_TURN: Agentic multi-turn conversation tests using Penelope
+    """
+
+    SINGLE_TURN = "Single-Turn"
+    MULTI_TURN = "Multi-Turn"
+
+    @classmethod
+    def get_value(cls, test_type):
+        """Get the string value of a test type"""
+        if isinstance(test_type, cls):
+            return test_type.value
+        return test_type
+
+    @classmethod
+    def from_string(cls, value: str):
+        """Get enum from string value (case-sensitive, accepts only allowed values)."""
+        if not value:
+            return None
+        for test_type in cls:
+            if test_type.value == value:
+                return test_type
+        return None
+
+
 # TestSetType Enum - DB-level test set classification aligned with initial_data.json type_lookup
-class TestSetType(Enum):
+class TestSetType(str, Enum):
     SINGLE_TURN = "Single-Turn"
     MULTI_TURN = "Multi-Turn"
 
@@ -48,17 +79,11 @@ def get_value(cls, test_set_type):
 
     @classmethod
     def from_string(cls, value: str):
-        """Get enum from string value (case-insensitive, accepts snake_case and hyphenated).
-
-        Maps both 'single_turn' and 'Single-Turn' to TestSetType.SINGLE_TURN, so API
-        clients that use snake_case conventions are handled transparently.
-        """
+        """Get enum from string value (case-sensitive, accepts only allowed values)."""
         if not value:
             return None
-        # Normalize underscores to hyphens so snake_case aliases work (single_turn → single-turn)
-        normalized = value.lower().replace("_", "-")
         for test_set_type in cls:
-            if test_set_type.value.lower() == normalized:
+            if test_set_type.value == value:
                 return test_set_type
         return None
 

apps/backend/src/rhesis/backend/app/models/test.py

@@ -116,7 +116,7 @@ def to_searchable_text(self) -> str:
         - Single-turn: Uses prompt.content and expected_response
         - Multi-turn: Uses test_configuration (goal, instructions, scenario, etc.)
         """
-        from rhesis.backend.tasks.enums import TestType
+        from rhesis.backend.app.constants import TestType
         from rhesis.backend.tasks.execution.modes import get_test_type
 
         test_type = get_test_type(self)

apps/backend/src/rhesis/backend/app/schemas/test.py

@@ -1,10 +1,9 @@
 from datetime import datetime
 from typing import Any, Dict, List, Optional, Union
 
-from pydantic import UUID4, BaseModel, ConfigDict, ValidationError, field_validator
+from pydantic import UUID4, BaseModel, ConfigDict, field_validator
 
 from rhesis.backend.app.schemas import Base
-from rhesis.backend.app.schemas.multi_turn_test_config import validate_multi_turn_config
 from rhesis.backend.app.schemas.user import UserReference
 
 
@@ -113,36 +112,19 @@ class TestCreate(TestBase):
     prompt: Optional[TestPromptCreate] = None
     test_type: Optional[str] = None
 
+    @field_validator("test_type")
+    @classmethod
+    def validate_test_type(cls, v: Optional[str]) -> Optional[str]:
+        from rhesis.backend.app.schemas.validators import format_test_type
+
+        return format_test_type(v)
+
     @field_validator("test_configuration")
     @classmethod
     def validate_test_configuration(cls, v: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
-        """
-        Validate test_configuration JSON based on content.
-
-        For multi-turn tests (when goal is present), validates against MultiTurnTestConfig schema.
-        """
-        if v is None:
-            return None
+        from rhesis.backend.app.schemas.validators import validate_test_config_content
 
-        # If 'goal' is present, this is a multi-turn test configuration
-        if "goal" in v:
-            try:
-                # Validate using multi-turn config schema
-                validated_config = validate_multi_turn_config(v)
-                # Return as dict for storage
-                return validated_config.model_dump(exclude_none=True)
-            except ValidationError as e:
-                # Re-raise with more context
-                error_messages = []
-                for error in e.errors():
-                    field = " -> ".join(str(loc) for loc in error["loc"])
-                    error_messages.append(f"{field}: {error['msg']}")
-                raise ValueError(
-                    f"Invalid multi-turn test configuration: {'; '.join(error_messages)}"
-                )
-
-        # For other configurations, allow any valid JSON
-        return v
+        return validate_test_config_content(v)
 
 
 class TestUpdate(TestBase):
@@ -153,36 +135,19 @@ class TestUpdate(TestBase):
     prompt: Optional[TestPromptCreate] = None
     test_type: Optional[str] = None
 
+    @field_validator("test_type")
+    @classmethod
+    def validate_test_type(cls, v: Optional[str]) -> Optional[str]:
+        from rhesis.backend.app.schemas.validators import format_test_type
+
+        return format_test_type(v)
+
     @field_validator("test_configuration")
     @classmethod
     def validate_test_configuration(cls, v: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
-        """
-        Validate test_configuration JSON based on content.
+        from rhesis.backend.app.schemas.validators import validate_test_config_content
 
-        For multi-turn tests (when goal is present), validates against MultiTurnTestConfig schema.
-        """
-        if v is None:
-            return None
-
-        # If 'goal' is present, this is a multi-turn test configuration
-        if "goal" in v:
-            try:
-                # Validate using multi-turn config schema
-                validated_config = validate_multi_turn_config(v)
-                # Return as dict for storage
-                return validated_config.model_dump(exclude_none=True)
-            except ValidationError as e:
-                # Re-raise with more context
-                error_messages = []
-                for error in e.errors():
-                    field = " -> ".join(str(loc) for loc in error["loc"])
-                    error_messages.append(f"{field}: {error['msg']}")
-                raise ValueError(
-                    f"Invalid multi-turn test configuration: {'; '.join(error_messages)}"
-                )
-
-        # For other configurations, allow any valid JSON
-        return v
+        return validate_test_config_content(v)
 
 
 class Test(TestBase):
@@ -221,6 +186,14 @@ class TestBulkCreate(BaseModel):
     owner_id: Optional[UUID4] = None
     status: Optional[str] = None
     priority: Optional[int] = None
+    test_type: Optional[str] = None
+
+    @field_validator("test_type")
+    @classmethod
+    def validate_test_type(cls, v: Optional[str]) -> Optional[str]:
+        from rhesis.backend.app.schemas.validators import format_test_type
+
+        return format_test_type(v)
 
     @field_validator("assignee_id", "owner_id")
     @classmethod
@@ -259,21 +232,9 @@ def validate_multi_turn_or_prompt(cls, v, info):
                 "or 'test_configuration' with 'goal' must be provided (for multi-turn tests)"
             )
 
-        # If 'goal' is present, validate as multi-turn config
-        if "goal" in v:
-            try:
-                validated_config = validate_multi_turn_config(v)
-                return validated_config.model_dump(exclude_none=True)
-            except ValidationError as e:
-                error_messages = []
-                for error in e.errors():
-                    field = " -> ".join(str(loc) for loc in error["loc"])
-                    error_messages.append(f"{field}: {error['msg']}")
-                raise ValueError(
-                    f"Invalid multi-turn test configuration: {'; '.join(error_messages)}"
-                )
+        from rhesis.backend.app.schemas.validators import validate_test_config_content
 
-        return v
+        return validate_test_config_content(v)
 
 
 class TestBulkCreateRequest(BaseModel):
@@ -339,23 +300,20 @@ class TestExecuteRequest(BaseModel):
     # Optional: Explicitly specify test type (otherwise auto-detected)
     test_type: Optional[str] = None  # "Single-Turn" or "Multi-Turn"
 
+    @field_validator("test_type")
+    @classmethod
+    def validate_test_type(cls, v: Optional[str]) -> Optional[str]:
+        from rhesis.backend.app.schemas.validators import format_test_type
+
+        return format_test_type(v)
+
     @field_validator("test_configuration")
     @classmethod
     def validate_test_configuration(cls, v, info):
         """Validate multi-turn test configuration if provided."""
-        if v and "goal" in v:
-            try:
-                validated_config = validate_multi_turn_config(v)
-                return validated_config.model_dump(exclude_none=True)
-            except ValidationError as e:
-                error_messages = []
-                for error in e.errors():
-                    field = " -> ".join(str(loc) for loc in error["loc"])
-                    error_messages.append(f"{field}: {error['msg']}")
-                raise ValueError(
-                    f"Invalid multi-turn test configuration: {'; '.join(error_messages)}"
-                )
-        return v
+        from rhesis.backend.app.schemas.validators import validate_test_config_content
+
+        return validate_test_config_content(v)
 
     def model_post_init(self, __context):
         """Validate that either test_id or test definition is provided."""
@@ -421,6 +379,13 @@ class ConversationToTestRequest(BaseModel):
     endpoint_id: Optional[UUID4] = None
     test_type: Optional[str] = "Multi-Turn"  # "Single-Turn" or "Multi-Turn"
 
+    @field_validator("test_type")
+    @classmethod
+    def validate_test_type(cls, v: Optional[str]) -> Optional[str]:
+        from rhesis.backend.app.schemas.validators import format_test_type
+
+        return format_test_type(v)
+
 
 class SingleTurnTestExtraction(BaseModel):
     """LLM-extracted metadata for a single-turn test."""

apps/backend/src/rhesis/backend/app/schemas/test_set.py

@@ -81,6 +81,13 @@ class TestData(BaseModel):
     priority: Optional[int] = None
     metadata: Optional[Dict[str, Any]] = {}
 
+    @field_validator("test_type")
+    @classmethod
+    def validate_test_type(cls, v: Optional[str]) -> Optional[str]:
+        from rhesis.backend.app.schemas.validators import format_test_type
+
+        return format_test_type(v)
+
     @field_validator("assignee_id", "owner_id")
     @classmethod
     def validate_uuid(cls, v):
@@ -105,9 +112,6 @@ def validate_multi_turn_or_prompt(cls, v, info):
         - prompt is provided (single-turn test), OR
         - test_configuration with goal is provided (multi-turn test)
         """
-        from pydantic import ValidationError
-
-        from rhesis.backend.app.schemas.multi_turn_test_config import validate_multi_turn_config
 
         prompt = info.data.get("prompt")
 
@@ -122,21 +126,9 @@ def validate_multi_turn_or_prompt(cls, v, info):
                 "or 'test_configuration' with 'goal' must be provided (for multi-turn tests)"
             )
 
-        # If 'goal' is present, validate as multi-turn config
-        if "goal" in v:
-            try:
-                validated_config = validate_multi_turn_config(v)
-                return validated_config.model_dump(exclude_none=True)
-            except ValidationError as e:
-                error_messages = []
-                for error in e.errors():
-                    field = " -> ".join(str(loc) for loc in error["loc"])
-                    error_messages.append(f"{field}: {error['msg']}")
-                raise ValueError(
-                    f"Invalid multi-turn test configuration: {'; '.join(error_messages)}"
-                )
+        from rhesis.backend.app.schemas.validators import validate_test_config_content
 
-        return v
+        return validate_test_config_content(v)
 
 
 class TestSetBulkCreate(BaseModel):
@@ -150,6 +142,13 @@ class TestSetBulkCreate(BaseModel):
     tests: List[TestData]
     metadata: Optional[Dict[str, Any]] = None
 
+    @field_validator("test_set_type")
+    @classmethod
+    def validate_test_set_type(cls, v: str) -> str:
+        from rhesis.backend.app.schemas.validators import format_test_set_type
+
+        return format_test_set_type(v)
+
     @field_validator("owner_id", "assignee_id")
     @classmethod
     def validate_uuid_fields(cls, v):

apps/backend/src/rhesis/backend/app/schemas/validators.py

@@ -0,0 +1,64 @@
+from typing import Any, Dict, Optional
+
+from pydantic import ValidationError
+
+from rhesis.backend.app.constants import TestSetType, TestType
+from rhesis.backend.app.schemas.multi_turn_test_config import validate_multi_turn_config
+
+
+def format_test_type(v: Optional[str]) -> Optional[str]:
+    """Format test type to title case and validate against allowed types."""
+    if v is None:
+        return None
+
+    formatted = v.title()
+    allowed_types = [t.value for t in TestType]
+
+    if formatted not in allowed_types:
+        raise ValueError(f"Invalid test type '{v}'. Allowed values are: {', '.join(allowed_types)}")
+
+    return formatted
+
+
+def format_test_set_type(v: Optional[str]) -> Optional[str]:
+    """Format test set type to title case and validate against allowed types."""
+    if v is None:
+        return None
+
+    formatted = v.title()
+    allowed_types = [t.value for t in TestSetType]
+
+    if formatted not in allowed_types:
+        raise ValueError(
+            f"Invalid test set type '{v}'. Allowed values are: {', '.join(allowed_types)}"
+        )
+
+    return formatted
+
+
+def validate_test_config_content(v: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
+    """
+    Validate test_configuration JSON based on content.
+
+    For multi-turn tests (when goal is present), validates against MultiTurnTestConfig schema.
+    """
+    if v is None:
+        return None
+
+    # If 'goal' is present, this is a multi-turn test configuration
+    if "goal" in v:
+        try:
+            # Validate using multi-turn config schema
+            validated_config = validate_multi_turn_config(v)
+            # Return as dict for storage
+            return validated_config.model_dump(exclude_none=True)
+        except ValidationError as e:
+            # Re-raise with more context
+            error_messages = []
+            for error in e.errors():
+                field = " -> ".join(str(loc) for loc in error["loc"])
+                error_messages.append(f"{field}: {error['msg']}")
+            raise ValueError(f"Invalid multi-turn test configuration: {'; '.join(error_messages)}")
+
+    # For other configurations, allow any valid JSON
+    return v