141 lines
5 KiB
Python
141 lines
5 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
End-to-End Validation Tests for New Architecture
|
|
Validates that the new architecture works correctly in real scenarios.
|
|
"""
|
|
|
|
import pytest
|
|
import sys
|
|
import os
|
|
|
|
# Add gateway to path
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
|
|
|
|
from modules.datamodels.datamodelWorkflow import ActionDefinition, AiResponse
|
|
from modules.datamodels.datamodelDocref import DocumentReferenceList, DocumentListReference
|
|
from modules.datamodels.datamodelChat import ChatWorkflow
|
|
from modules.shared.jsonUtils import parseJsonWithModel
|
|
|
|
|
|
class TestArchitectureValidation:
|
|
"""End-to-end validation of new architecture"""
|
|
|
|
def test_actionDefinition_stage1_to_stage2_flow(self):
|
|
"""Validate Stage 1 → Stage 2 flow"""
|
|
# Stage 1: Action selection with resources
|
|
stage1 = ActionDefinition(
|
|
action="ai.process",
|
|
actionObjective="Process documents",
|
|
documentList=DocumentReferenceList(references=[
|
|
DocumentListReference(label="input_docs")
|
|
])
|
|
)
|
|
assert stage1.needsStage2() is True # Parameters not set
|
|
|
|
# Stage 2: Add parameters
|
|
stage1.parameters = {"resultType": "pdf", "aiPrompt": "Generate report"}
|
|
assert stage1.needsStage2() is False # Parameters now set
|
|
|
|
def test_documentReferenceList_round_trip(self):
|
|
"""Validate DocumentReferenceList string conversion round-trip"""
|
|
# Create typed references
|
|
refList = DocumentReferenceList(references=[
|
|
DocumentListReference(messageId="msg123", label="task1_results"),
|
|
DocumentListReference(label="task2_results")
|
|
])
|
|
|
|
# Convert to strings
|
|
stringList = refList.to_string_list()
|
|
assert len(stringList) == 2
|
|
assert "docList:msg123:task1_results" in stringList
|
|
assert "docList:task2_results" in stringList
|
|
|
|
# Parse back from strings
|
|
parsedList = DocumentReferenceList.from_string_list(stringList)
|
|
assert len(parsedList.references) == 2
|
|
assert parsedList.references[0].messageId == "msg123"
|
|
assert parsedList.references[1].messageId is None
|
|
|
|
def test_parseJsonWithModel_actionDefinition(self):
|
|
"""Validate parseJsonWithModel with ActionDefinition"""
|
|
jsonStr = '''
|
|
{
|
|
"action": "ai.process",
|
|
"actionObjective": "Process documents",
|
|
"documentList": {
|
|
"references": [
|
|
{"messageId": "msg123", "label": "task1_results"}
|
|
]
|
|
}
|
|
}
|
|
'''
|
|
|
|
# Should parse successfully
|
|
result = parseJsonWithModel(jsonStr, ActionDefinition)
|
|
assert isinstance(result, ActionDefinition)
|
|
assert result.action == "ai.process"
|
|
assert result.actionObjective == "Process documents"
|
|
|
|
def test_workflow_state_management(self):
|
|
"""Validate workflow state management"""
|
|
workflow = ChatWorkflow(
|
|
id="test123",
|
|
name="Test",
|
|
mandateId="test_mandate"
|
|
)
|
|
|
|
# Test state increments
|
|
workflow.incrementAction()
|
|
assert workflow.getActionIndex() == 1
|
|
|
|
workflow.incrementTask()
|
|
assert workflow.getTaskIndex() == 1
|
|
assert workflow.getActionIndex() == 0 # Reset
|
|
|
|
workflow.incrementRound()
|
|
assert workflow.getRoundIndex() == 1
|
|
assert workflow.getTaskIndex() == 0 # Reset
|
|
assert workflow.getActionIndex() == 0 # Reset
|
|
|
|
def test_aiResponse_structure(self):
|
|
"""Validate AiResponse structure"""
|
|
response = AiResponse(
|
|
content='{"result": "success"}',
|
|
metadata=None,
|
|
documents=None
|
|
)
|
|
|
|
# Test toJson conversion
|
|
jsonResult = response.toJson()
|
|
assert isinstance(jsonResult, dict)
|
|
assert jsonResult["result"] == "success"
|
|
|
|
|
|
class TestBackwardCompatibilityRemoved:
|
|
"""Validate that backward compatibility has been removed"""
|
|
|
|
def test_no_string_document_references(self):
|
|
"""Validate that string document references are not supported"""
|
|
# DocumentReferenceList.from_string_list() should work
|
|
# But direct string usage should be converted
|
|
stringList = ["docList:task1_results"]
|
|
refList = DocumentReferenceList.from_string_list(stringList)
|
|
assert isinstance(refList, DocumentReferenceList)
|
|
assert len(refList.references) == 1
|
|
|
|
def test_no_snake_case_fields(self):
|
|
"""Validate that only camelCase fields are used"""
|
|
actionDef = ActionDefinition(
|
|
action="ai.process",
|
|
actionObjective="Test objective"
|
|
)
|
|
# Should use camelCase
|
|
assert hasattr(actionDef, "actionObjective")
|
|
assert not hasattr(actionDef, "action_objective") # snake_case removed
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|
|
|