#!/usr/bin/env python3 """ Test script for Action Mapping API endpoints Tests CRUD operations and validates the implementation """ import requests import json from typing import Optional import sys class ActionMappingAPITester: """Test harness for Action Mapping API""" def __init__(self, base_url: str = "http://localhost:8000", token: Optional[str] = None): self.base_url = base_url self.token = token self.session = requests.Session() if token: self.session.headers.update({ "Authorization": f"Bearer {token}" }) def login(self, username: str = "admin", password: str = "admin123"): """Login and get access token""" print(f"\n=== Logging in as {username} ===") response = self.session.post( f"{self.base_url}/api/v1/auth/login", json={"username": username, "password": password} ) if response.status_code == 200: data = response.json() self.token = data["access_token"] self.session.headers.update({ "Authorization": f"Bearer {self.token}" }) print(f"✓ Login successful") return True else: print(f"✗ Login failed: {response.status_code} - {response.text}") return False def list_action_mappings(self, enabled_only: bool = False): """Test: List all action mappings""" print(f"\n=== Test: List Action Mappings (enabled_only={enabled_only}) ===") params = {"enabled_only": enabled_only} response = self.session.get( f"{self.base_url}/api/v1/action-mappings", params=params ) print(f"Status: {response.status_code}") if response.status_code == 200: data = response.json() print(f"✓ Found {data['total_count']} action mappings") print(f" Enabled: {data['enabled_count']}") print(f" Disabled: {data['disabled_count']}") for mapping in data['mappings']: print(f"\n {mapping['name']} ({mapping['id']})") print(f" Input: {mapping['input_action']}") print(f" Outputs: {mapping['output_actions']}") print(f" Enabled: {mapping['enabled']}") print(f" Executions: {mapping['execution_count']}") return data else: print(f"✗ Failed: {response.text}") return None def create_action_mapping(self, name: str, input_action: str, output_actions: list): """Test: Create new action mapping""" print(f"\n=== Test: Create Action Mapping '{name}' ===") mapping_data = { "name": name, "description": f"Test mapping: {name}", "input_action": input_action, "output_actions": output_actions, "enabled": True } print(f"Creating:") print(f" Input: {input_action}") print(f" Outputs: {output_actions}") response = self.session.post( f"{self.base_url}/api/v1/action-mappings", json=mapping_data ) print(f"Status: {response.status_code}") if response.status_code == 201: data = response.json() print(f"✓ Created action mapping") print(f" ID: {data['id']}") print(f" Name: {data['name']}") return data else: print(f"✗ Failed: {response.text}") return None def get_action_mapping(self, mapping_id: str): """Test: Get specific action mapping""" print(f"\n=== Test: Get Action Mapping {mapping_id} ===") response = self.session.get( f"{self.base_url}/api/v1/action-mappings/{mapping_id}" ) print(f"Status: {response.status_code}") if response.status_code == 200: data = response.json() print(f"✓ Retrieved action mapping") print(f" Name: {data['name']}") print(f" Input: {data['input_action']}") print(f" Outputs: {data['output_actions']}") return data else: print(f"✗ Failed: {response.text}") return None def update_action_mapping(self, mapping_id: str, updates: dict): """Test: Update action mapping""" print(f"\n=== Test: Update Action Mapping {mapping_id} ===") print(f"Updates: {json.dumps(updates, indent=2)}") response = self.session.put( f"{self.base_url}/api/v1/action-mappings/{mapping_id}", json=updates ) print(f"Status: {response.status_code}") if response.status_code == 200: data = response.json() print(f"✓ Updated action mapping") print(f" Name: {data['name']}") print(f" Enabled: {data['enabled']}") return data else: print(f"✗ Failed: {response.text}") return None def delete_action_mapping(self, mapping_id: str): """Test: Delete action mapping""" print(f"\n=== Test: Delete Action Mapping {mapping_id} ===") response = self.session.delete( f"{self.base_url}/api/v1/action-mappings/{mapping_id}" ) print(f"Status: {response.status_code}") if response.status_code == 204: print(f"✓ Deleted action mapping") return True else: print(f"✗ Failed: {response.text}") return False def run_full_test_suite(self): """Run complete test suite""" print("=" * 60) print("Action Mapping API Test Suite") print("=" * 60) # Login if not self.login(): print("\n✗ Cannot proceed without authentication") return False # Test 1: List initial state initial_list = self.list_action_mappings() # Test 2: Create action mapping created = self.create_action_mapping( name="Test Motion Detection Alert", input_action="VMD_Start(101038)", output_actions=[ "CrossSwitch(101038, 1, 0)", "SendMail(security@example.com, Motion Detected)" ] ) if not created: print("\n✗ Cannot proceed - failed to create action mapping") return False mapping_id = created['id'] # Test 3: Get the created mapping self.get_action_mapping(mapping_id) # Test 4: Update the mapping self.update_action_mapping( mapping_id, { "enabled": False, "description": "Updated test description" } ) # Test 5: List again to see the update self.list_action_mappings() # Test 6: Delete the mapping self.delete_action_mapping(mapping_id) # Test 7: Verify deletion print(f"\n=== Test: Verify Deletion ===") response = self.session.get(f"{self.base_url}/api/v1/action-mappings/{mapping_id}") if response.status_code == 404: print("✓ Mapping successfully deleted (404 Not Found)") else: print(f"✗ Unexpected response: {response.status_code}") print("\n" + "=" * 60) print("Test Suite Complete") print("=" * 60) return True def main(): """Main entry point""" import argparse parser = argparse.ArgumentParser(description="Test Action Mapping API") parser.add_argument("--url", default="http://localhost:8000", help="API base URL") parser.add_argument("--username", default="admin", help="Username for login") parser.add_argument("--password", default="admin123", help="Password for login") args = parser.parse_args() tester = ActionMappingAPITester(base_url=args.url) if not tester.login(args.username, args.password): print("Login failed - check credentials and API availability") sys.exit(1) success = tester.run_full_test_suite() sys.exit(0 if success else 1) if __name__ == "__main__": main()