#!/usr/bin/env python3 """ Test Templates System Tests for the template service and API endpoints. """ import json import os import tempfile import unittest from datetime import datetime import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from models.template import WorkflowTemplate, TemplateParameter, TemplateDifficulty from models.visual_workflow import VisualWorkflow, VisualNode, VisualEdge, Position, Size, Port, WorkflowSettings, ParameterType from services.template_service import TemplateService class TestTemplateModels(unittest.TestCase): """Test template data models""" def setUp(self): """Set up test data""" self.sample_workflow = VisualWorkflow( id="test_workflow", name="Test Workflow", description="A test workflow", version="1.0.0", created_at=datetime.now(), updated_at=datetime.now(), created_by="test_user", nodes=[ VisualNode( id="node1", type="click", position=Position(100, 100), size=Size(150, 80), parameters={"target": "{{button_selector}}"}, input_ports=[Port("in", "input", "input")], output_ports=[Port("out", "output", "output")] ) ], edges=[], variables=[], settings=WorkflowSettings(), is_template=True ) def test_template_parameter_serialization(self): """Test template parameter to_dict and from_dict""" param = TemplateParameter( name="button_selector", type=ParameterType.TARGET, description="Button to click", node_id="node1", parameter_name="target", label="Button" ) # Test serialization param_dict = param.to_dict() self.assertEqual(param_dict['name'], "button_selector") self.assertEqual(param_dict['type'], "target") self.assertEqual(param_dict['node_id'], "node1") # Test deserialization param2 = TemplateParameter.from_dict(param_dict) self.assertEqual(param2.name, param.name) self.assertEqual(param2.type, param.type) self.assertEqual(param2.node_id, param.node_id) def test_workflow_template_serialization(self): """Test workflow template to_dict and from_dict""" template = WorkflowTemplate( id="test_template", name="Test Template", description="A test template", category="Test", workflow=self.sample_workflow, parameters=[ TemplateParameter( name="button_selector", type=ParameterType.TARGET, description="Button to click", node_id="node1", parameter_name="target" ) ], tags=["test", "example"], difficulty=TemplateDifficulty.BEGINNER ) # Test serialization template_dict = template.to_dict() self.assertEqual(template_dict['id'], "test_template") self.assertEqual(template_dict['name'], "Test Template") self.assertEqual(len(template_dict['parameters']), 1) self.assertIn('workflow', template_dict) # Test deserialization template2 = WorkflowTemplate.from_dict(template_dict) self.assertEqual(template2.id, template.id) self.assertEqual(template2.name, template.name) self.assertEqual(len(template2.parameters), 1) self.assertEqual(template2.workflow.id, template.workflow.id) def test_template_instantiation(self): """Test creating a workflow from a template""" template = WorkflowTemplate( id="test_template", name="Test Template", description="A test template", category="Test", workflow=self.sample_workflow, parameters=[ TemplateParameter( name="button_selector", type=ParameterType.TARGET, description="Button to click", node_id="node1", parameter_name="target" ) ] ) # Instantiate template parameters = {"button_selector": "button.submit"} workflow = template.instantiate(parameters, "My Workflow", "test_user") # Verify workflow self.assertEqual(workflow.name, "My Workflow") self.assertEqual(workflow.created_by, "test_user") self.assertFalse(workflow.is_template) self.assertNotEqual(workflow.id, template.workflow.id) # Should have new ID # Verify parameter substitution self.assertEqual(len(workflow.nodes), 1) self.assertEqual(workflow.nodes[0].parameters["target"], "button.submit") def test_template_validation(self): """Test template validation""" # Valid template template = WorkflowTemplate( id="test_template", name="Test Template", description="A test template", category="Test", workflow=self.sample_workflow, parameters=[] ) errors = template.validate() self.assertEqual(len(errors), 0) # Invalid template - missing required fields template.id = "" template.name = "" errors = template.validate() self.assertGreater(len(errors), 0) self.assertTrue(any("ID is required" in err for err in errors)) self.assertTrue(any("name is required" in err for err in errors)) class TestTemplateService(unittest.TestCase): """Test template service""" def setUp(self): """Set up test environment""" self.temp_dir = tempfile.mkdtemp() self.service = TemplateService(data_dir=self.temp_dir) # Create sample workflow self.sample_workflow = VisualWorkflow( id="test_workflow", name="Test Workflow", description="A test workflow", version="1.0.0", created_at=datetime.now(), updated_at=datetime.now(), created_by="test_user", nodes=[ VisualNode( id="node1", type="click", position=Position(100, 100), size=Size(150, 80), parameters={"target": "{{button_selector}}"}, input_ports=[Port("in", "input", "input")], output_ports=[Port("out", "output", "output")] ) ], edges=[], variables=[], settings=WorkflowSettings(), is_template=True ) def tearDown(self): """Clean up test environment""" import shutil shutil.rmtree(self.temp_dir, ignore_errors=True) def test_create_and_get_template(self): """Test creating and retrieving a template""" template_data = { 'name': 'Test Template', 'description': 'A test template', 'category': 'Test', 'workflow': self.sample_workflow.to_dict(), 'parameters': [ { 'name': 'button_selector', 'type': 'target', 'description': 'Button to click', 'node_id': 'node1', 'parameter_name': 'target', 'required': True } ], 'tags': ['test'], 'difficulty': 'beginner' } # Create template template = self.service.create_template(template_data) self.assertIsNotNone(template.id) self.assertEqual(template.name, 'Test Template') # Retrieve template retrieved = self.service.get_template(template.id) self.assertIsNotNone(retrieved) self.assertEqual(retrieved.name, template.name) self.assertEqual(len(retrieved.parameters), 1) def test_list_templates(self): """Test listing templates""" # Should have default templates templates = self.service.list_templates() self.assertGreater(len(templates), 0) # Test filtering by category web_templates = self.service.list_templates(category="Web Automation") self.assertGreater(len(web_templates), 0) for template in web_templates: self.assertEqual(template.category, "Web Automation") # Test filtering by difficulty beginner_templates = self.service.list_templates(difficulty="beginner") self.assertGreater(len(beginner_templates), 0) for template in beginner_templates: self.assertEqual(template.difficulty, TemplateDifficulty.BEGINNER) def test_template_instantiation(self): """Test instantiating a template""" # Get a default template templates = self.service.list_templates() self.assertGreater(len(templates), 0) template = templates[0] # Create parameters for instantiation parameters = {} for param in template.parameters: if param.type == ParameterType.STRING: parameters[param.name] = f"test_{param.name}" elif param.type == ParameterType.TARGET: parameters[param.name] = f"#{param.name}" # Instantiate template workflow = self.service.instantiate_template( template.id, parameters, "Test Workflow", "test_user" ) self.assertIsNotNone(workflow) self.assertEqual(workflow.name, "Test Workflow") self.assertEqual(workflow.created_by, "test_user") self.assertFalse(workflow.is_template) def test_create_template_from_workflow(self): """Test creating a template from an existing workflow""" # Create a workflow workflow = VisualWorkflow( id="source_workflow", name="Source Workflow", description="Source for template", version="1.0.0", created_at=datetime.now(), updated_at=datetime.now(), created_by="test_user", nodes=[ VisualNode( id="node1", type="type", position=Position(100, 100), size=Size(150, 80), parameters={"target": "input.username", "text": "testuser"}, input_ports=[Port("in", "input", "input")], output_ports=[Port("out", "output", "output")] ) ], edges=[], variables=[], settings=WorkflowSettings(), tags=["login", "test"] ) # Define template parameters parameters = [ { 'name': 'username', 'type': 'string', 'description': 'Username to enter', 'node_id': 'node1', 'parameter_name': 'text', 'required': True } ] # Create template from workflow template = self.service.create_template_from_workflow( workflow, "Login Template", "Template for login", "Authentication", parameters ) self.assertIsNotNone(template.id) self.assertEqual(template.name, "Login Template") self.assertEqual(template.category, "Authentication") self.assertEqual(len(template.parameters), 1) self.assertTrue(template.workflow.is_template) def test_update_template(self): """Test updating a template""" # Create a template first template_data = { 'name': 'Original Template', 'description': 'Original description', 'category': 'Test', 'workflow': self.sample_workflow.to_dict(), 'parameters': [], 'tags': ['original'] } template = self.service.create_template(template_data) original_id = template.id # Update the template updated_data = template_data.copy() updated_data['name'] = 'Updated Template' updated_data['description'] = 'Updated description' updated_data['tags'] = ['updated'] updated_template = self.service.update_template(original_id, updated_data) self.assertIsNotNone(updated_template) self.assertEqual(updated_template.id, original_id) self.assertEqual(updated_template.name, 'Updated Template') self.assertEqual(updated_template.description, 'Updated description') self.assertEqual(updated_template.tags, ['updated']) def test_delete_template(self): """Test deleting a template""" # Create a template first template_data = { 'name': 'Template to Delete', 'description': 'Will be deleted', 'category': 'Test', 'workflow': self.sample_workflow.to_dict(), 'parameters': [] } template = self.service.create_template(template_data) template_id = template.id # Verify it exists self.assertIsNotNone(self.service.get_template(template_id)) # Delete it success = self.service.delete_template(template_id) self.assertTrue(success) # Verify it's gone self.assertIsNone(self.service.get_template(template_id)) # Try to delete non-existent template success = self.service.delete_template("non_existent") self.assertFalse(success) class TestTemplateAPI(unittest.TestCase): """Test template API endpoints""" def setUp(self): """Set up test Flask app""" import tempfile from app import app self.temp_dir = tempfile.mkdtemp() app.config['TESTING'] = True app.config['TEMPLATE_DATA_DIR'] = self.temp_dir self.client = app.test_client() self.app_context = app.app_context() self.app_context.push() def tearDown(self): """Clean up test environment""" import shutil self.app_context.pop() shutil.rmtree(self.temp_dir, ignore_errors=True) def test_list_templates_endpoint(self): """Test GET /api/templates/""" response = self.client.get('/api/templates/') self.assertEqual(response.status_code, 200) data = json.loads(response.data) self.assertIn('templates', data) self.assertIn('count', data) self.assertIsInstance(data['templates'], list) def test_get_template_endpoint(self): """Test GET /api/templates/""" # First get list of templates response = self.client.get('/api/templates/') data = json.loads(response.data) if data['templates']: template_id = data['templates'][0]['id'] # Get specific template response = self.client.get(f'/api/templates/{template_id}') self.assertEqual(response.status_code, 200) template_data = json.loads(response.data) self.assertEqual(template_data['id'], template_id) self.assertIn('workflow', template_data) # Test non-existent template response = self.client.get('/api/templates/non_existent') self.assertEqual(response.status_code, 404) def test_create_template_endpoint(self): """Test POST /api/templates/""" template_data = { 'name': 'API Test Template', 'description': 'Created via API', 'category': 'Test', 'workflow': { 'id': 'test_workflow', 'name': 'Test Workflow', 'description': 'Test', 'version': '1.0.0', 'created_at': datetime.now().isoformat(), 'updated_at': datetime.now().isoformat(), 'created_by': 'test', 'nodes': [], 'edges': [], 'variables': [], 'settings': { 'timeout': 300000, 'retry_on_failure': True, 'max_retries': 3, 'enable_self_healing': True, 'enable_analytics': True }, 'tags': [], 'is_template': True }, 'parameters': [], 'tags': ['api', 'test'] } response = self.client.post('/api/templates/', data=json.dumps(template_data), content_type='application/json') self.assertEqual(response.status_code, 201) created_template = json.loads(response.data) self.assertEqual(created_template['name'], 'API Test Template') self.assertIn('id', created_template) def test_template_filtering(self): """Test template filtering by category and difficulty""" # Test category filtering response = self.client.get('/api/templates/?category=Web Automation') self.assertEqual(response.status_code, 200) data = json.loads(response.data) for template in data['templates']: self.assertEqual(template['category'], 'Web Automation') # Test difficulty filtering response = self.client.get('/api/templates/?difficulty=beginner') self.assertEqual(response.status_code, 200) data = json.loads(response.data) for template in data['templates']: self.assertEqual(template['difficulty'], 'beginner') def test_get_template_categories(self): """Test GET /api/templates/categories""" response = self.client.get('/api/templates/categories') self.assertEqual(response.status_code, 200) data = json.loads(response.data) self.assertIn('categories', data) self.assertIsInstance(data['categories'], list) def run_tests(): """Run all template tests""" print("๐Ÿงช Running Template System Tests...") # Create test suite suite = unittest.TestSuite() # Add test cases suite.addTest(unittest.makeSuite(TestTemplateModels)) suite.addTest(unittest.makeSuite(TestTemplateService)) suite.addTest(unittest.makeSuite(TestTemplateAPI)) # Run tests runner = unittest.TextTestRunner(verbosity=2) result = runner.run(suite) # Print summary if result.wasSuccessful(): print("โœ… All template tests passed!") return True else: print(f"โŒ {len(result.failures)} test(s) failed, {len(result.errors)} error(s)") return False if __name__ == '__main__': success = run_tests() exit(0 if success else 1)