rpa_vision_v3/visual_workflow_builder/backend/test_templates.py

#!/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/<id>"""
        # 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)