#!/usr/bin/env python3 """ Real Integration Test for Template Message API Tests template creation, retrieval, update, and deletion with real API calls Data is actually written to MongoDB for comprehensive testing """ import requests import json import sys import os import uuid from datetime import datetime # Import existing configuration from common.config.app_settings import app_settings from webapi.config.site_settings import site_settings from common.token.token_manager import TokenManager # API base URL - use site_settings host = 'localhost' if site_settings.SERVER_HOST == '0.0.0.0' else site_settings.SERVER_HOST port = site_settings.SERVER_PORT BASE_URL = f"http://{host}:{port}/api/notification" class TemplateMessageRealIntegrationTest: def __init__(self): self.base_url = BASE_URL # Use tenant_id from token instead of generating random one self.test_tenant_id = "test_tenant_user" # Keep consistent with token tenant_id self.test_template_id = f"test_template_{uuid.uuid4().hex[:8]}" # Generate tokens using existing TokenManager print("🔑 Generating tokens using existing TokenManager...") self.admin_token, self.tenant_token = self._generate_tokens() print(f"🧪 Starting Template Message Real Token Integration Test - Tenant ID: {self.test_tenant_id}") print(f"📊 Test data will be written to MongoDB") print(f"📝 Test template ID: {self.test_template_id}") print(f"🌐 API Base URL: {self.base_url}") def _generate_tokens(self): """Generate admin and tenant tokens using existing TokenManager""" token_manager = TokenManager() # Generate admin token with subject format (like authentication service) admin_subject = {"subject": {"id": "admin_user", "role": 8}} # ADMINISTRATOR = 8 admin_token = token_manager.create_access_token(admin_subject) # Generate tenant token with subject format (like authentication service) tenant_subject = {"subject": {"id": self.test_tenant_id, "role": 2}} # BUSINESS = 2 tenant_token = token_manager.create_access_token(tenant_subject) print(f"✅ Generated admin token: {admin_token[:20]}...") print(f"✅ Generated tenant token: {tenant_token[:20]}...") return admin_token, tenant_token def _get_headers(self, token): """Get request headers with authorization token""" return { "Content-Type": "application/json", "Authorization": f"Bearer {token}" } def test_create_template(self): """Test creating a new template""" print("\n🚀 Testing template creation...") # Test data for template creation template_data = { "template_id": self.test_template_id, "subject": "Welcome to {{company_name}}", "body": """

Welcome {{new_employee_name}}!

Welcome to {{company_name}}. Your employee ID is {{employee_id}}.

Position: {{position}}

Department: {{department}}

Start Date: {{start_date}}

Email: {{email_address}}

Best regards,
{{company_name}} Team

""", "region": 0, # English template "tenant_id": self.test_tenant_id, "is_active": True, "description": "Welcome email template for new employees" } try: response = requests.post( f"{self.base_url}/tenant/templates/create", headers=self._get_headers(self.tenant_token), json=template_data ) print(f"📤 Create Template Response Status: {response.status_code}") print(f"📤 Create Template Response: {response.text}") if response.status_code in [200, 201]: # 201 is also success for creation result = response.json() print(f"✅ Template created successfully!") print(f"📝 Template ID: {result.get('result', {}).get('template_id')}") print(f"📝 Subject: {result.get('result', {}).get('subject')}") return True else: print(f"❌ Failed to create template: {response.status_code}") return False except Exception as e: print(f"❌ Exception during template creation: {str(e)}") return False def test_get_template(self): """Test retrieving a template (not available - only list API exists)""" print("\n📖 Testing template retrieval...") print("ℹ️ Individual template get API not available - only list API exists") return True # Skip this test as API doesn't exist def test_update_template(self): """Test updating a template""" print("\n✏️ Testing template update...") # Updated template data updated_data = { "subject": "Updated: Welcome to {{company_name}} - {{new_employee_name}}", "body": """

🎉 Welcome {{new_employee_name}}!

Welcome to {{company_name}}. Your employee ID is {{employee_id}}.

Employee Details:

Position: {{position}}
Department: {{department}}
Start Date: {{start_date}}
Email: {{email_address}}

Best regards,
{{company_name}} Team

""", "region": 0 # Add region parameter } try: response = requests.put( f"{self.base_url}/tenant/templates/update/{self.test_template_id}", headers=self._get_headers(self.tenant_token), json=updated_data ) print(f"📤 Update Template Response Status: {response.status_code}") print(f"📤 Update Template Response: {response.text}") if response.status_code == 200: result = response.json() print(f"✅ Template updated successfully!") print(f"📝 Updated Subject: {result.get('result', {}).get('subject')}") print(f"📝 Updated Body length: {len(result.get('result', {}).get('body', ''))} characters") return True else: print(f"❌ Failed to update template: {response.status_code}") return False except Exception as e: print(f"❌ Exception during template update: {str(e)}") return False def test_list_templates(self): """Test listing templates for tenant""" print("\n📋 Testing template listing...") try: response = requests.get( f"{self.base_url}/tenant/templates/list?region=0", headers=self._get_headers(self.tenant_token) ) print(f"📤 List Templates Response Status: {response.status_code}") print(f"📤 List Templates Response: {response.text}") if response.status_code == 200: result = response.json() templates = result.get('templates', []) print(f"✅ Templates listed successfully!") print(f"📝 Found {len(templates)} templates for tenant {self.test_tenant_id}") for i, template in enumerate(templates[:3]): # Show first 3 templates print(f" {i+1}. {template.get('template_id')}: {template.get('subject')}") if len(templates) > 3: print(f" ... and {len(templates) - 3} more templates") return True else: print(f"❌ Failed to list templates: {response.status_code}") return False except Exception as e: print(f"❌ Exception during template listing: {str(e)}") return False def test_render_template(self): """Test rendering a template with properties""" print("\n🎨 Testing template rendering...") # Properties for template rendering render_properties = { "company_name": "TestCorp", "new_employee_name": "John Doe", "employee_id": "EMP123", "position": "Software Engineer", "department": "Engineering", "start_date": "2024-01-15", "email_address": "john.doe@testcorp.com" } render_data = { "template_id": self.test_template_id, "properties": render_properties, "region": 0 } try: response = requests.post( f"{self.base_url}/tenant/templates/render/{self.test_template_id}?region=0", headers=self._get_headers(self.tenant_token), json=render_data ) print(f"📤 Render Template Response Status: {response.status_code}") print(f"📤 Render Template Response: {response.text}") if response.status_code == 200: result = response.json() rendered_result = result.get('result', {}) print(f"✅ Template rendered successfully!") print(f"📝 Rendered Subject: {rendered_result.get('subject')}") print(f"📝 Rendered Body length: {len(rendered_result.get('body', ''))} characters") # Show a preview of the rendered content subject_preview = rendered_result.get('subject', '')[:50] + "..." if len(rendered_result.get('subject', '')) > 50 else rendered_result.get('subject', '') body_preview = rendered_result.get('body', '')[:100] + "..." if len(rendered_result.get('body', '')) > 100 else rendered_result.get('body', '') print(f"📝 Subject Preview: {subject_preview}") print(f"📝 Body Preview: {body_preview}") return True else: print(f"❌ Failed to render template: {response.status_code}") return False except Exception as e: print(f"❌ Exception during template rendering: {str(e)}") return False def test_delete_template(self): """Test deleting a template""" print("\n🗑️ Testing template deletion...") try: response = requests.delete( f"{self.base_url}/tenant/templates/delete/{self.test_template_id}?region=0", headers=self._get_headers(self.tenant_token) ) print(f"📤 Delete Template Response Status: {response.status_code}") print(f"📤 Delete Template Response: {response.text}") if response.status_code == 200: result = response.json() print(f"✅ Template deleted successfully!") print(f"📝 Deleted Template ID: {result.get('result', {}).get('template_id')}") return True else: print(f"❌ Failed to delete template: {response.status_code}") return False except Exception as e: print(f"❌ Exception during template deletion: {str(e)}") return False def test_admin_create_global_template(self): """Test creating a global template (Admin only)""" print("\n👑 Testing admin global template creation...") # Test data for global template creation global_template_data = { "template_id": f"global_template_{uuid.uuid4().hex[:8]}", "subject": "Global Welcome to {{company_name}}", "body": """

Global Welcome {{new_employee_name}}!

Welcome to {{company_name}} - This is a global template.

Employee ID: {{employee_id}}

Position: {{position}}

Department: {{department}}

Best regards,
{{company_name}} Team

""", "region": 0, # English template "is_active": True } try: response = requests.post( f"{self.base_url}/admin/global_templates/create", headers=self._get_headers(self.admin_token), json=global_template_data ) print(f"📤 Admin Create Global Template Response Status: {response.status_code}") print(f"📤 Admin Create Global Template Response: {response.text}") if response.status_code in [200, 201]: result = response.json() print(f"✅ Global template created successfully!") print(f"📝 Template ID: {result.get('template_id')}") print(f"📝 Action: {result.get('action')}") return True else: print(f"❌ Failed to create global template: {response.status_code}") return False except Exception as e: print(f"❌ Exception during global template creation: {str(e)}") return False def test_admin_list_global_templates(self): """Test listing global templates (Admin and Tenant)""" print("\n📋 Testing global template listing...") try: response = requests.get( f"{self.base_url}/global_templates/list?region=0", headers=self._get_headers(self.admin_token) ) print(f"📤 List Global Templates Response Status: {response.status_code}") print(f"📤 List Global Templates Response: {response.text}") if response.status_code == 200: result = response.json() templates = result.get('templates', []) print(f"✅ Global templates listed successfully!") print(f"📝 Found {len(templates)} global templates") for i, template in enumerate(templates[:3]): # Show first 3 templates print(f" {i+1}. {template.get('template_id')}: {template.get('subject')}") if len(templates) > 3: print(f" ... and {len(templates) - 3} more templates") return True else: print(f"❌ Failed to list global templates: {response.status_code}") return False except Exception as e: print(f"❌ Exception during global template listing: {str(e)}") return False def test_assign_templates_to_tenant(self): """Test assigning global templates to tenant""" print("\n🔗 Testing template assignment to tenant...") # First, let's get some global templates to assign try: list_response = requests.get( f"{self.base_url}/global_templates/list?region=0", headers=self._get_headers(self.admin_token) ) if list_response.status_code == 200: global_templates = list_response.json().get('templates', []) if global_templates: # Use the first global template for assignment template_to_assign = global_templates[0].get('template_id') assign_data = { "template_ids": [template_to_assign], "region": 0 } response = requests.post( f"{self.base_url}/tenant/templates/assign", headers=self._get_headers(self.tenant_token), json=assign_data ) print(f"📤 Assign Templates Response Status: {response.status_code}") print(f"📤 Assign Templates Response: {response.text}") if response.status_code == 200: result = response.json() print(f"✅ Templates assigned successfully!") print(f"📝 Assigned template: {template_to_assign}") print(f"📝 Results: {result.get('results')}") return True else: print(f"❌ Failed to assign templates: {response.status_code}") return False else: print("ℹ️ No global templates available for assignment") return True # Skip this test if no global templates else: print(f"❌ Failed to get global templates for assignment: {list_response.status_code}") return False except Exception as e: print(f"❌ Exception during template assignment: {str(e)}") return False def test_admin_update_global_template(self): """Test updating a global template (Admin only)""" print("\n✏️ Testing admin global template update...") # First, let's get a global template to update try: list_response = requests.get( f"{self.base_url}/global_templates/list?region=0", headers=self._get_headers(self.admin_token) ) if list_response.status_code == 200: global_templates = list_response.json().get('templates', []) if global_templates: # Use the first global template for update template_to_update = global_templates[0].get('template_id') update_data = { "subject": "Updated Global: Welcome to {{company_name}} - {{new_employee_name}}", "body": """

🎉 Updated Global Welcome {{new_employee_name}}!

Welcome to {{company_name}} - This is an updated global template.

Employee Information:

Employee ID: {{employee_id}}
Position: {{position}}
Department: {{department}}

Best regards,
{{company_name}} Team

""", "region": 0 } response = requests.put( f"{self.base_url}/admin/global_templates/update/{template_to_update}", headers=self._get_headers(self.admin_token), json=update_data ) print(f"📤 Admin Update Global Template Response Status: {response.status_code}") print(f"📤 Admin Update Global Template Response: {response.text}") if response.status_code == 200: result = response.json() print(f"✅ Global template updated successfully!") print(f"📝 Updated template: {template_to_update}") return True else: print(f"❌ Failed to update global template: {response.status_code}") return False else: print("ℹ️ No global templates available for update") return True # Skip this test if no global templates else: print(f"❌ Failed to get global templates for update: {list_response.status_code}") return False except Exception as e: print(f"❌ Exception during global template update: {str(e)}") return False def test_admin_delete_global_template(self): """Test deleting a global template (Admin only)""" print("\n🗑️ Testing admin global template deletion...") # First, let's get a global template to delete try: list_response = requests.get( f"{self.base_url}/global_templates/list?region=0", headers=self._get_headers(self.admin_token) ) if list_response.status_code == 200: global_templates = list_response.json().get('templates', []) if global_templates: # Use the first global template for deletion template_to_delete = global_templates[0].get('template_id') response = requests.delete( f"{self.base_url}/admin/global_templates/delete/{template_to_delete}", headers=self._get_headers(self.admin_token) ) print(f"📤 Admin Delete Global Template Response Status: {response.status_code}") print(f"📤 Admin Delete Global Template Response: {response.text}") if response.status_code == 200: result = response.json() print(f"✅ Global template deleted successfully!") print(f"📝 Deleted template: {template_to_delete}") return True else: print(f"❌ Failed to delete global template: {response.status_code}") return False else: print("ℹ️ No global templates available for deletion") return True # Skip this test if no global templates else: print(f"❌ Failed to get global templates for deletion: {list_response.status_code}") return False except Exception as e: print(f"❌ Exception during global template deletion: {str(e)}") return False def test_error_scenarios(self): """Test error scenarios and edge cases""" print("\n⚠️ Testing error scenarios...") # Test 1: Create template with invalid data print("\n📝 Test 1: Create template with missing required fields") invalid_data = { "template_id": f"invalid_template_{uuid.uuid4().hex[:8]}", # Missing subject and body "region": 0 } try: response = requests.post( f"{self.base_url}/tenant/templates/create", headers=self._get_headers(self.tenant_token), json=invalid_data ) print(f"📤 Invalid Create Response Status: {response.status_code}") if response.status_code == 400: print("✅ Correctly rejected invalid template data") else: print(f"⚠️ Unexpected response for invalid data: {response.status_code}") except Exception as e: print(f"❌ Exception during invalid template creation: {str(e)}") # Test 2: Get non-existent template (not available - only list API exists) print("\n📝 Test 2: Get non-existent template (not available)") print("ℹ️ Individual template get API not available - skipping this test") # Test 3: Update non-existent template print("\n📝 Test 3: Update non-existent template") non_existent_id = f"non_existent_{uuid.uuid4().hex[:8]}" update_data = { "subject": "This should fail", "body": "This template doesn't exist", "region": 0 } try: response = requests.put( f"{self.base_url}/tenant/templates/update/{non_existent_id}", headers=self._get_headers(self.tenant_token), json=update_data ) print(f"📤 Update Non-existent Response Status: {response.status_code}") if response.status_code == 404: print("✅ Correctly handled update of non-existent template") else: print(f"⚠️ Unexpected response for updating non-existent template: {response.status_code}") except Exception as e: print(f"❌ Exception during non-existent template update: {str(e)}") # Test 4: Tenant trying to access admin API print("\n📝 Test 4: Tenant trying to access admin API") admin_data = { "template_id": f"unauthorized_template_{uuid.uuid4().hex[:8]}", "subject": "Unauthorized template", "body": "This should fail", "region": 0 } try: response = requests.post( f"{self.base_url}/admin/global_templates/create", headers=self._get_headers(self.tenant_token), # Using tenant token instead of admin json=admin_data ) print(f"📤 Unauthorized Admin Access Response Status: {response.status_code}") if response.status_code == 403: print("✅ Correctly rejected tenant access to admin API") else: print(f"⚠️ Unexpected response for unauthorized admin access: {response.status_code}") except Exception as e: print(f"❌ Exception during unauthorized admin access: {str(e)}") # Test 5: Assign non-existent templates to tenant print("\n📝 Test 5: Assign non-existent templates to tenant") assign_data = { "template_ids": [f"non_existent_template_{uuid.uuid4().hex[:8]}"], "region": 0 } try: response = requests.post( f"{self.base_url}/tenant/templates/assign", headers=self._get_headers(self.tenant_token), json=assign_data ) print(f"📤 Assign Non-existent Templates Response Status: {response.status_code}") if response.status_code == 400: print("✅ Correctly rejected assignment of non-existent templates") else: print(f"⚠️ Unexpected response for assigning non-existent templates: {response.status_code}") except Exception as e: print(f"❌ Exception during non-existent template assignment: {str(e)}") def run_all_tests(self): """Run all template message tests""" print("🧪 Starting Template Message Integration Tests...") print("=" * 60) test_results = [] # Run tenant template tests test_results.append(("Create Template", self.test_create_template())) test_results.append(("Get Template", self.test_get_template())) test_results.append(("Update Template", self.test_update_template())) test_results.append(("List Templates", self.test_list_templates())) test_results.append(("Render Template", self.test_render_template())) #test_results.append(("Delete Template", self.test_delete_template())) # Run admin template tests test_results.append(("Admin Create Global", self.test_admin_create_global_template())) test_results.append(("Admin List Global", self.test_admin_list_global_templates())) test_results.append(("Assign Templates", self.test_assign_templates_to_tenant())) test_results.append(("Admin Update Global", self.test_admin_update_global_template())) #test_results.append(("Admin Delete Global", self.test_admin_delete_global_template())) # Run error scenario tests self.test_error_scenarios() # Print test summary print("\n" + "=" * 60) print("📊 TEST SUMMARY") print("=" * 60) passed = 0 failed = 0 for test_name, result in test_results: status = "✅ PASS" if result else "❌ FAIL" print(f"{test_name:<20} {status}") if result: passed += 1 else: failed += 1 print("-" * 60) print(f"Total Tests: {len(test_results)}") print(f"Passed: {passed}") print(f"Failed: {failed}") print(f"Success Rate: {(passed/len(test_results)*100):.1f}%") if failed == 0: print("\n🎉 All tests passed! Template Message API is working correctly.") else: print(f"\n⚠️ {failed} test(s) failed. Please check the implementation.") return failed == 0 def main(): """Main function to run the integration test""" print("🚀 Template Message Real Integration Test") print("=" * 60) try: # Create test instance test = TemplateMessageRealIntegrationTest() # Run all tests success = test.run_all_tests() if success: print("\n✅ Integration test completed successfully!") sys.exit(0) else: print("\n❌ Integration test failed!") sys.exit(1) except KeyboardInterrupt: print("\n⚠️ Test interrupted by user") sys.exit(1) except Exception as e: print(f"\n❌ Unexpected error: {str(e)}") sys.exit(1) if __name__ == "__main__": main()