#!/usr/bin/env python3 """ 🧪 Business Logic Tests für Aegilock Fraud Detection API Erweiterte Python-basierte Tests für komplexe Geschäftslogik """ import requests import json import time import sys from datetime import datetime, timedelta from typing import Dict, List, Optional, Any import threading from concurrent.futures import ThreadPoolExecutor import statistics class FraudDetectionTester: def __init__(self, api_url: str = "https://api.aegilock.de/woocommerce/fraud-check", api_key: str = "YOUR_API_KEY"): self.api_url = api_url self.api_key = api_key self.headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } self.test_results = [] self.session = requests.Session() def log(self, message: str, level: str = "INFO"): """Log messages with timestamp""" timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] {level}: {message}") def validate_response(self, response: Dict[str, Any]) -> bool: """Validate API response structure""" required_fields = ['decision', 'confidence', 'risk_score'] valid_decisions = ['APPROVE', 'MANUAL_REVIEW', 'BLOCK'] try: # Check required fields for field in required_fields: if field not in response: raise ValueError(f"Missing field: {field}") # Validate confidence range (no longer checking risk_score) confidence = response.get('confidence') if confidence is not None and not (0 <= confidence <= 100): raise ValueError('confidence must be between 0 and 100') if response.get('decision') not in valid_decisions: raise ValueError('Invalid decision value') # Already checked above return True except (ValueError, TypeError) as e: self.log(f"Response validation failed: {e}", "ERROR") return False def make_request(self, test_data: Dict[str, Any], test_name: str) -> Optional[Dict[str, Any]]: """Make API request with error handling""" try: # Wrap in proper API format if needed if 'features' not in test_data: request_data = { 'features': test_data, 'timestamp': datetime.now().isoformat(), 'version': '1.0.0', 'site_url': 'https://test-shop.com' } else: request_data = test_data start_time = time.time() response = self.session.post( self.api_url, headers=self.headers, json=request_data, timeout=10 ) end_time = time.time() response_time = end_time - start_time if response.status_code == 200: result = response.json() if self.validate_response(result): result['_response_time'] = response_time result['_test_name'] = test_name self.test_results.append(result) return result else: self.log(f"Invalid response format for {test_name}", "ERROR") return None else: self.log(f"HTTP {response.status_code} for {test_name}: {response.text}", "ERROR") return None except requests.exceptions.RequestException as e: self.log(f"Request failed for {test_name}: {e}", "ERROR") return None def test_business_logic_scenarios(self): """Test verschiedene Business Logic Szenarien""" self.log("🧠 Starting Business Logic Tests") test_cases = [ { "name": "Low-risk: PayPal repeat customer", "data": { "order_id": "BL-001", "order_amount": 150, "is_new_customer": False, "customer_order_count": 5, "customer_email": "loyal@customer.de", "billing_country": "DE", "shipping_country": "DE", "currency": "EUR", "payment_method": "paypal" }, "expected_decision": ["APPROVE"] }, { "name": "Low-risk: Small SEPA order", "data": { "order_id": "BL-002", "order_amount": 45, "customer_email": "regular@shopper.de", "is_new_customer": False, "customer_order_count": 2, "billing_country": "DE", "shipping_country": "DE", "currency": "EUR", "payment_method": "sepa" }, "expected_decision": ["APPROVE"] }, { "name": "Medium-risk: High-value new customer", "data": { "order_id": "BL-003", "order_amount": 1200, "customer_email": "newbuyer@example.com", "is_new_customer": True, "customer_order_count": 0, "billing_country": "DE", "shipping_country": "DE", "currency": "EUR", "payment_method": "credit_card" }, "expected_decision": ["MANUAL_REVIEW"] }, { "name": "High-risk: Packstation + COD", "data": { "order_id": "BL-004", "order_amount": 800, "customer_email": "anonymous@tempmail.org", "is_new_customer": True, "billing_country": "DE", "shipping_country": "DE", "currency": "EUR", "payment_method": "cash_on_delivery", "is_packstation": True, "shipping_address": "Packstation 123" }, "expected_decision": ["MANUAL_REVIEW", "BLOCK"] }, { "name": "High-risk: Country mismatch + VPN", "data": { "order_id": "BL-005", "order_amount": 600, "customer_email": "suspicious@buyer.com", "is_new_customer": True, "billing_country": "DE", "shipping_country": "RO", "currency": "EUR", "payment_method": "credit_card", "is_vpn": True, "customer_ip": "104.248.57.123" }, "expected_decision": ["MANUAL_REVIEW", "BLOCK"] }, { "name": "Auto-approve: Klarna payment", "data": { "order_id": "BL-006", "order_amount": 300, "customer_email": "klarna@customer.de", "billing_country": "DE", "shipping_country": "DE", "currency": "EUR", "payment_method": "klarna", "is_new_customer": True }, "expected_decision": ["APPROVE"] } ] for test_case in test_cases: self.log(f"Testing: {test_case['name']}") result = self.make_request(test_case['data'], test_case['name']) if result: decision = result.get('decision', 'UNKNOWN') confidence = result.get('confidence', 0) response_time = result.get('_response_time', 0) explanation = result.get('explanation', 'No explanation provided') # Evaluate results decision_match = decision in test_case['expected_decision'] status = "✅ PASS" if decision_match else "❌ FAIL" print(f" {status} Decision: {decision} (Expected: {test_case['expected_decision']})") print(f" Confidence: {confidence}%") print(f" Response Time: {response_time:.3f}s") print(f" Explanation: {explanation[:100]}...") print() else: print(f" ❌ FAIL - No valid response") print() time.sleep(0.5) # Rate limiting def categorize_decision(self, decision: str) -> str: """Categorize decision into risk level""" if decision == "APPROVE": return "low-risk" elif decision == "MANUAL_REVIEW": return "medium-risk" else: return "high-risk" def test_velocity_patterns(self): """Test velocity and pattern detection""" self.log("🚀 Testing Velocity Patterns") # Test 1: Rapid successive orders from same IP base_ip = "192.168.1.100" for i in range(5): test_data = { "order_id": f"VEL-IP-{i+1}", "order_amount": 299.99, "customer_ip": base_ip, "customer_email": f"customer{i+1}@example.com", "timestamp": datetime.now().isoformat() } result = self.make_request(test_data, f"Velocity Test IP #{i+1}") if result: decision = result.get('decision', 'UNKNOWN') confidence = result.get('confidence', 0) print(f" Request {i+1}: Decision {decision}, Confidence: {confidence}%") time.sleep(0.2) # Very short delay to trigger velocity detection print() # Test 2: Same email, different IPs (account takeover pattern) base_email = "victim@example.com" ips = ["10.0.0.1", "192.168.2.50", "172.16.0.100"] for i, ip in enumerate(ips): test_data = { "order_id": f"VEL-EMAIL-{i+1}", "order_amount": 150.00, "customer_ip": ip, "customer_email": base_email, "timestamp": datetime.now().isoformat() } result = self.make_request(test_data, f"Email Velocity #{i+1}") if result: decision = result.get('decision', 'UNKNOWN') confidence = result.get('confidence', 0) print(f" IP {ip}: Decision {decision}, Confidence: {confidence}%") time.sleep(0.3) def test_geographic_patterns(self): """Test geographic risk patterns""" self.log("🌍 Testing Geographic Patterns") geo_tests = [ { "name": "DACH region (low risk)", "billing_country": "DE", "shipping_country": "DE", "customer_ip": "89.15.123.45", # German IP "expected_risk": "low" }, { "name": "EU but different countries", "billing_country": "DE", "shipping_country": "FR", "customer_ip": "89.15.123.45", "expected_risk": "medium" }, { "name": "High-risk country combination", "billing_country": "DE", "shipping_country": "NG", # Nigeria "customer_ip": "89.15.123.45", "expected_risk": "high" }, { "name": "VPN/Proxy detection", "billing_country": "DE", "shipping_country": "DE", "customer_ip": "104.248.57.123", # Known VPN IP "is_vpn": True, "expected_risk": "medium-high" } ] for test in geo_tests: test_data = { "order_id": f"GEO-{test['name'].replace(' ', '-').upper()}", "order_amount": 200.00, "billing_country": test["billing_country"], "shipping_country": test["shipping_country"], "customer_ip": test["customer_ip"], "customer_email": "test@example.com" } if "is_vpn" in test: test_data["is_vpn"] = test["is_vpn"] result = self.make_request(test_data, test['name']) if result: decision = result.get('decision', 'UNKNOWN') confidence = result.get('confidence', 0) risk_level = self.categorize_decision(decision) match = test['expected_risk'] in risk_level status = "✅" if match else "❌" print(f" {status} {test['name']}: {decision} ({risk_level}, expected {test['expected_risk']})") print() def test_payment_method_risks(self): """Test different payment method risk levels""" self.log("💳 Testing Payment Method Risks") payment_tests = [ {"method": "sepa", "expected_risk": "low", "amount": 500}, {"method": "paypal", "expected_risk": "low", "amount": 500}, {"method": "credit_card", "expected_risk": "medium", "amount": 500}, {"method": "cod", "expected_risk": "high", "amount": 500}, # Cash on Delivery {"method": "prepayment", "expected_risk": "low", "amount": 500}, {"method": "cod", "expected_risk": "very_high", "amount": 2000, "is_packstation": True} # COD + Packstation ] for test in payment_tests: test_data = { "order_id": f"PAY-{test['method'].upper()}-{test['amount']}", "order_amount": test['amount'], "payment_method": test['method'], "customer_email": "payment-test@example.com", "billing_country": "DE", "shipping_country": "DE" } if test.get('is_packstation'): test_data['is_packstation'] = True test_data['shipping_address'] = "Packstation 123" result = self.make_request(test_data, f"{test['method']} payment test") if result: decision = result.get('decision', 'UNKNOWN') confidence = result.get('confidence', 0) risk_level = self.categorize_decision(decision) print(f" {test['method']}: Decision {decision} ({risk_level}), Confidence: {confidence}%") print() def test_time_based_patterns(self): """Test time-based risk patterns""" self.log("⏰ Testing Time-based Patterns") time_tests = [ {"time": "09:30:00", "day": "monday", "expected_risk": "low"}, {"time": "14:00:00", "day": "wednesday", "expected_risk": "low"}, {"time": "23:45:00", "day": "friday", "expected_risk": "medium"}, {"time": "03:15:00", "day": "sunday", "expected_risk": "high"}, {"time": "02:30:00", "day": "tuesday", "expected_risk": "high"} ] for test in time_tests: test_data = { "order_id": f"TIME-{test['time'].replace(':', '')}-{test['day'].upper()}", "order_amount": 300.00, "order_time": test['time'], "order_day": test['day'], "customer_email": "timetest@example.com" } result = self.make_request(test_data, f"Time test {test['time']} {test['day']}") if result: decision = result.get('decision', 'UNKNOWN') confidence = result.get('confidence', 0) risk_level = self.categorize_decision(decision) print(f" {test['time']} {test['day']}: Decision {decision} ({risk_level}), Confidence: {confidence}%") print() def test_concurrent_load(self, num_threads: int = 10, requests_per_thread: int = 5): """Test concurrent load and performance""" self.log(f"⚡ Testing Concurrent Load ({num_threads} threads, {requests_per_thread} requests each)") response_times = [] success_count = 0 total_requests = num_threads * requests_per_thread def worker_thread(thread_id: int): nonlocal success_count, response_times for i in range(requests_per_thread): test_data = { "order_id": f"LOAD-T{thread_id}-R{i}", "order_amount": 100.00, "customer_email": f"load-test-{thread_id}-{i}@example.com" } start_time = time.time() try: response = self.session.post( self.api_url, headers=self.headers, json=test_data, timeout=10 ) end_time = time.time() if response.status_code == 200: success_count += 1 response_times.append(end_time - start_time) except Exception as e: pass # Count as failure # Start all threads start_total = time.time() with ThreadPoolExecutor(max_workers=num_threads) as executor: futures = [executor.submit(worker_thread, i) for i in range(num_threads)] for future in futures: future.result() end_total = time.time() # Calculate statistics total_time = end_total - start_total success_rate = (success_count / total_requests) * 100 if response_times: avg_response = statistics.mean(response_times) min_response = min(response_times) max_response = max(response_times) p95_response = sorted(response_times)[int(len(response_times) * 0.95)] else: avg_response = min_response = max_response = p95_response = 0 print(f" Total Requests: {total_requests}") print(f" Successful: {success_count} ({success_rate:.1f}%)") print(f" Total Time: {total_time:.2f}s") print(f" Requests/sec: {total_requests/total_time:.2f}") print(f" Response Times:") print(f" Average: {avg_response*1000:.1f}ms") print(f" Min: {min_response*1000:.1f}ms") print(f" Max: {max_response*1000:.1f}ms") print(f" 95th percentile: {p95_response*1000:.1f}ms") print() def generate_summary_report(self): """Generate summary report of all tests""" self.log("📊 Generating Summary Report") if not self.test_results: print("No test results to analyze") return # Basic statistics total_tests = len(self.test_results) confidences = [r.get('confidence', 0) for r in self.test_results] response_times = [r.get('_response_time', 0) for r in self.test_results if r.get('_response_time')] decisions = {} for result in self.test_results: decision = result.get('decision', 'UNKNOWN') decisions[decision] = decisions.get(decision, 0) + 1 print(f" Total Tests: {total_tests}") print(f" Confidence Distribution:") if confidences: print(f" Average: {statistics.mean(confidences):.1f}%") print(f" Min: {min(confidences):.1f}%") print(f" Max: {max(confidences):.1f}%") print(f" Standard Deviation: {statistics.stdev(confidences) if len(confidences) > 1 else 0:.1f}%") print(f" Decision Distribution:") for decision, count in decisions.items(): percentage = (count / total_tests) * 100 print(f" {decision}: {count} ({percentage:.1f}%)") if response_times: print(f" Performance:") print(f" Average Response Time: {statistics.mean(response_times)*1000:.1f}ms") if len(response_times) > 1: print(f" 95th Percentile: {sorted(response_times)[int(len(response_times)*0.95)]*1000:.1f}ms") # Save detailed results to JSON timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"fraud_test_results_{timestamp}.json" with open(filename, 'w') as f: json.dump(self.test_results, f, indent=2, default=str) print(f" Detailed results saved to: {filename}") def run_all_tests(self): """Run complete test suite""" self.log("🚀 Starting Complete Fraud Detection Test Suite") print("=" * 60) try: self.test_business_logic_scenarios() self.test_velocity_patterns() self.test_geographic_patterns() self.test_payment_method_risks() self.test_time_based_patterns() self.test_concurrent_load() except KeyboardInterrupt: self.log("Tests interrupted by user", "WARNING") except Exception as e: self.log(f"Unexpected error during testing: {e}", "ERROR") finally: print("=" * 60) self.generate_summary_report() self.log("Test Suite Complete") def main(): """Main entry point""" if len(sys.argv) > 1 and sys.argv[1] == "--api-key": if len(sys.argv) < 3: print("Usage: python3 business-logic-tests.py --api-key YOUR_API_KEY") sys.exit(1) api_key = sys.argv[2] else: api_key = "YOUR_API_KEY" # Default placeholder tester = FraudDetectionTester(api_key=api_key) tester.run_all_tests() if __name__ == "__main__": main()