"""Integration tests for hook system with WalletService.""" from decimal import Decimal from django.test import TestCase from wallet_utils import ( WalletService, WalletRepository, HookType, clear_hooks, register_hook, ) from wallet_utils.hooks import HookContext from wallet_utils.exceptions import HookRejectionError from tests.models import TestUser class TestWalletServiceWithHooks(TestCase): """Test WalletService operations with hooks enabled.""" def setUp(self): """Set up test data.""" self.user1 = TestUser.objects.create(username="user1", email="user1@test.com") self.user2 = TestUser.objects.create(username="user2", email="user2@test.com") self.point_types = { "cash": 2, # Using cash instead of cash_balance for simplicity "credit_balance": 2, "reward_points": 0, } self.repository = WalletRepository( user_model=TestUser, point_types=self.point_types, ) self.service = WalletService(self.repository, enable_hooks=True) # Clear any hooks from previous tests clear_hooks() def tearDown(self): """Clean up after tests.""" clear_hooks() # ============================================================================ # add_point() Tests # ============================================================================ def test_add_point_with_pre_hook_success(self): """Test add_point with PRE hook that allows transaction.""" executed = [] def check_amount(context: HookContext) -> bool: executed.append("pre_hook") # Allow transactions under 1000 return context.amount < Decimal("1000") register_hook( name="check_amount", hook_type=HookType.PRE, callback=check_amount, operation="add", priority=100, ) result = self.service.add_point( user_id=self.user1.id, point_type="cash", amount=Decimal("100"), remarks="Test add", ) self.assertTrue(result.success) self.assertIsNotNone(result.transaction_id) self.assertIsNone(result.error_code) self.assertEqual(executed, ["pre_hook"]) def test_add_point_with_pre_hook_rejection_false(self): """Test add_point with PRE hook that rejects by returning False.""" def reject_large_amounts(context: HookContext) -> bool: # Reject amounts over 500 if context.amount > Decimal("500"): return False return True register_hook( name="reject_large", hook_type=HookType.PRE, callback=reject_large_amounts, operation="add", ) result = self.service.add_point( user_id=self.user1.id, point_type="cash", amount=Decimal("600"), remarks="Large amount", ) self.assertFalse(result.success) self.assertIsNone(result.transaction_id) self.assertIn("HOOK_REJECTED_REJECT_LARGE", result.error_code) # Balance should not change balance = self.repository.get_user_balance(self.user1.id, "cash") self.assertEqual(balance, Decimal("0")) def test_add_point_with_pre_hook_rejection_exception(self): """Test add_point with PRE hook that rejects by raising HookRejectionError.""" def check_daily_limit(context: HookContext) -> bool: if context.amount > Decimal("500"): raise HookRejectionError( error_code="DAILY_LIMIT_EXCEEDED", message="Daily transaction limit exceeded", details={"limit": 500, "requested": float(context.amount)}, ) return True register_hook( name="daily_limit", hook_type=HookType.PRE, callback=check_daily_limit, operation="add", ) result = self.service.add_point( user_id=self.user1.id, point_type="cash", amount=Decimal("600"), remarks="Over limit", ) self.assertFalse(result.success) self.assertEqual(result.error_code, "DAILY_LIMIT_EXCEEDED") self.assertEqual(result.error_message, "Daily transaction limit exceeded") self.assertEqual(result.error_details["limit"], 500) self.assertEqual(result.error_details["requested"], 600.0) def test_add_point_with_post_hook_success(self): """Test add_point with POST hook that executes successfully.""" executed = [] def log_transaction(context: HookContext) -> None: executed.append("post_hook") # Access transaction_id from metadata tx_id = context.metadata.get("transaction_id") self.assertIsNotNone(tx_id) register_hook( name="log_tx", hook_type=HookType.POST, callback=log_transaction, operation="add", ) result = self.service.add_point( user_id=self.user1.id, point_type="cash", amount=Decimal("100"), remarks="Test add", ) self.assertTrue(result.success) self.assertIsNotNone(result.transaction_id) self.assertIsNone(result.post_hook_errors) self.assertEqual(executed, ["post_hook"]) def test_add_point_with_post_hook_error(self): """Test add_point with POST hook that fails - transaction still succeeds.""" def failing_post_hook(context: HookContext) -> None: raise ValueError("POST hook failed") register_hook( name="failing_post", hook_type=HookType.POST, callback=failing_post_hook, operation="add", ) result = self.service.add_point( user_id=self.user1.id, point_type="cash", amount=Decimal("100"), remarks="Test add", ) # Transaction should succeed self.assertTrue(result.success) self.assertIsNotNone(result.transaction_id) # But POST hook error should be captured self.assertIsNotNone(result.post_hook_errors) self.assertEqual(len(result.post_hook_errors), 1) self.assertEqual(result.post_hook_errors[0].hook_name, "failing_post") self.assertEqual(result.post_hook_errors[0].error_code, "HOOK_EXECUTION_ERROR") def test_add_point_with_multiple_hooks_priority(self): """Test add_point with multiple hooks executed in priority order.""" executed = [] def hook_low_priority(context: HookContext) -> bool: executed.append("low_priority") return True def hook_high_priority(context: HookContext) -> bool: executed.append("high_priority") return True def hook_medium_priority(context: HookContext) -> bool: executed.append("medium_priority") return True # Register in random order register_hook("low", HookType.PRE, hook_low_priority, "add", priority=500) register_hook("high", HookType.PRE, hook_high_priority, "add", priority=10) register_hook("medium", HookType.PRE, hook_medium_priority, "add", priority=100) result = self.service.add_point( user_id=self.user1.id, point_type="cash", amount=Decimal("100"), remarks="Test priority", ) self.assertTrue(result.success) # Should execute in priority order self.assertEqual(executed, ["high_priority", "medium_priority", "low_priority"]) # ============================================================================ # deduct_point() Tests # ============================================================================ def test_deduct_point_with_pre_hook_success(self): """Test deduct_point with PRE hook that allows transaction.""" # Add balance first self.service.add_point( self.user1.id, "cash", Decimal("500"), "Initial balance" ) executed = [] def check_balance(context: HookContext) -> bool: executed.append("pre_hook") # Allow deduction if balance is sufficient return context.current_balance >= context.amount register_hook( name="check_balance", hook_type=HookType.PRE, callback=check_balance, operation="deduct", ) result = self.service.deduct_point( user_id=self.user1.id, point_type="cash", amount=Decimal("100"), remarks="Test deduct", ) self.assertTrue(result.success) self.assertEqual(executed, ["pre_hook"]) def test_deduct_point_with_pre_hook_rejection(self): """Test deduct_point with PRE hook that rejects transaction.""" # Add balance first self.service.add_point( self.user1.id, "cash", Decimal("500"), "Initial balance" ) def reject_large_deductions(context: HookContext) -> bool: if context.amount > Decimal("200"): raise HookRejectionError( error_code="AMOUNT_TOO_LARGE", message="Deduction amount too large", details={"max": 200, "requested": float(context.amount)}, ) return True register_hook( name="limit_deduction", hook_type=HookType.PRE, callback=reject_large_deductions, operation="deduct", ) result = self.service.deduct_point( user_id=self.user1.id, point_type="cash", amount=Decimal("300"), remarks="Large deduction", ) self.assertFalse(result.success) self.assertEqual(result.error_code, "AMOUNT_TOO_LARGE") # Balance should remain at 500 balance = self.repository.get_user_balance(self.user1.id, "cash") self.assertEqual(balance, Decimal("500")) def test_deduct_point_insufficient_balance_returns_result(self): """Test deduct_point with insufficient balance returns TransactionResult.""" # No balance added result = self.service.deduct_point( user_id=self.user1.id, point_type="cash", amount=Decimal("100"), remarks="Insufficient", ) self.assertFalse(result.success) self.assertEqual(result.error_code, "INSUFFICIENT_BALANCE") self.assertIsNotNone(result.error_message) def test_deduct_point_with_post_hook(self): """Test deduct_point with POST hook.""" # Add balance first self.service.add_point( self.user1.id, "cash", Decimal("500"), "Initial balance" ) executed = [] def send_notification(context: HookContext) -> None: executed.append("notification_sent") # In real scenario, this would send email/SMS register_hook( name="notify", hook_type=HookType.POST, callback=send_notification, operation="deduct", ) result = self.service.deduct_point( user_id=self.user1.id, point_type="cash", amount=Decimal("100"), remarks="Test deduct", ) self.assertTrue(result.success) self.assertEqual(executed, ["notification_sent"]) # ============================================================================ # transfer_points() Tests # ============================================================================ def test_transfer_with_pre_hook_success(self): """Test transfer_points with PRE hook that allows transaction.""" # Add balance to user1 self.service.add_point( self.user1.id, "cash", Decimal("500"), "Initial balance" ) executed = [] def check_transfer_limit(context: HookContext) -> bool: executed.append("pre_hook") # Allow transfers under 300 return context.amount < Decimal("300") register_hook( name="transfer_limit", hook_type=HookType.PRE, callback=check_transfer_limit, operation="transfer", ) result = self.service.transfer_points( from_user_id=self.user1.id, from_point_type="cash", to_user_id=self.user2.id, to_point_type="cash", amount=Decimal("200"), remarks="Test transfer", ) self.assertTrue(result["success"]) self.assertIsNone(result.get("error")) self.assertEqual(executed, ["pre_hook"]) def test_transfer_with_pre_hook_rejection(self): """Test transfer_points with PRE hook that rejects transaction.""" # Add balance to user1 self.service.add_point( self.user1.id, "cash", Decimal("500"), "Initial balance" ) def block_suspicious_transfer(context: HookContext) -> bool: # Block transfers over 400 if context.amount > Decimal("400"): raise HookRejectionError( error_code="SUSPICIOUS_TRANSFER", message="Transfer amount flagged as suspicious", details={"max_safe": 400}, ) return True register_hook( name="fraud_check", hook_type=HookType.PRE, callback=block_suspicious_transfer, operation="transfer", ) result = self.service.transfer_points( from_user_id=self.user1.id, from_point_type="cash", to_user_id=self.user2.id, to_point_type="cash", amount=Decimal("450"), remarks="Suspicious transfer", ) self.assertFalse(result["success"]) self.assertEqual(result["error"]["code"], "SUSPICIOUS_TRANSFER") # Balances should not change balance1 = self.repository.get_user_balance(self.user1.id, "cash") balance2 = self.repository.get_user_balance(self.user2.id, "cash") self.assertEqual(balance1, Decimal("500")) self.assertEqual(balance2, Decimal("0")) def test_transfer_with_post_hook(self): """Test transfer_points with POST hook.""" # Add balance to user1 self.service.add_point( self.user1.id, "cash", Decimal("500"), "Initial balance" ) executed = [] def log_transfer(context: HookContext) -> None: executed.append("logged") # Log transfer details self.assertEqual(context.operation, "transfer") self.assertEqual(context.to_user_id, self.user2.id) register_hook( name="log_transfer", hook_type=HookType.POST, callback=log_transfer, operation="transfer", ) result = self.service.transfer_points( from_user_id=self.user1.id, from_point_type="cash", to_user_id=self.user2.id, to_point_type="cash", amount=Decimal("200"), remarks="Test transfer", ) self.assertTrue(result["success"]) self.assertEqual(executed, ["logged"]) def test_transfer_with_post_hook_error(self): """Test transfer_points with POST hook error - transaction still succeeds.""" # Add balance to user1 self.service.add_point( self.user1.id, "cash", Decimal("500"), "Initial balance" ) def failing_post_hook(context: HookContext) -> None: raise ValueError("Failed to send notification") register_hook( name="notify_failed", hook_type=HookType.POST, callback=failing_post_hook, operation="transfer", ) result = self.service.transfer_points( from_user_id=self.user1.id, from_point_type="cash", to_user_id=self.user2.id, to_point_type="cash", amount=Decimal("200"), remarks="Test transfer", ) # Transfer should succeed self.assertTrue(result["success"]) # But warnings should be present self.assertIn("warnings", result) self.assertEqual(len(result["warnings"]), 1) self.assertEqual(result["warnings"][0]["hook"], "notify_failed") # ============================================================================ # Metadata and Inter-Hook Communication Tests # ============================================================================ def test_hooks_can_communicate_via_metadata(self): """Test that hooks can share data via metadata.""" executed = [] def hook1(context: HookContext) -> bool: context.metadata["hook1_data"] = "processed_by_hook1" executed.append("hook1") return True def hook2(context: HookContext) -> bool: # Access data from hook1 data = context.metadata.get("hook1_data") self.assertEqual(data, "processed_by_hook1") executed.append("hook2") return True register_hook("hook1", HookType.PRE, hook1, "add", priority=10) register_hook("hook2", HookType.PRE, hook2, "add", priority=20) result = self.service.add_point( self.user1.id, "cash", Decimal("100"), "Test metadata" ) self.assertTrue(result.success) self.assertEqual(executed, ["hook1", "hook2"]) # ============================================================================ # Global Hooks Tests # ============================================================================ def test_global_hook_applies_to_all_operations(self): """Test that global hooks ('*') apply to all operations.""" executed = [] def global_hook(context: HookContext) -> bool: executed.append(context.operation) return True register_hook( name="global", hook_type=HookType.PRE, callback=global_hook, operation="*", # Global hook ) # Test add self.service.add_point( self.user1.id, "cash", Decimal("100"), "Test add" ) # Test deduct self.service.deduct_point( self.user1.id, "cash", Decimal("50"), "Test deduct" ) # Test transfer (need balance) result = self.service.transfer_points( from_user_id=self.user1.id, to_user_id=self.user2.id, from_point_type="cash", to_point_type="cash", amount=Decimal("20"), remarks="Test transfer" ) self.assertEqual(executed, ["add", "deduct", "transfer"]) # ============================================================================ # Backwards Compatibility Tests # ============================================================================ def test_service_with_hooks_disabled_returns_transaction_id(self): """Test that with hooks disabled, service returns int (backwards compatible).""" # Create a new repository for this service repo_no_hooks = WalletRepository( user_model=TestUser, point_types=self.point_types, ) service_no_hooks = WalletService(repo_no_hooks, enable_hooks=False) # Register a hook (should not be executed) def should_not_run(context: HookContext) -> bool: raise Exception("This should not be called") register_hook("unused", HookType.PRE, should_not_run, "add") result = service_no_hooks.add_point( self.user1.id, "cash", Decimal("100"), "Test" ) # Should return int (transaction_id), not TransactionResult self.assertIsInstance(result, int) self.assertGreater(result, 0) def test_to_dict_method_for_api_responses(self): """Test TransactionResult.to_dict() for API responses.""" result = self.service.add_point( self.user1.id, "cash", Decimal("100"), "Test" ) result_dict = result.to_dict() self.assertTrue(result_dict["success"]) self.assertIsNotNone(result_dict["transaction_id"]) self.assertNotIn("error", result_dict) self.assertNotIn("warnings", result_dict) def test_to_dict_with_rejection_error(self): """Test TransactionResult.to_dict() with rejection error.""" def reject_hook(context: HookContext) -> bool: raise HookRejectionError( error_code="TEST_ERROR", message="Test rejection", details={"key": "value"}, ) register_hook("reject", HookType.PRE, reject_hook, "add") result = self.service.add_point( self.user1.id, "cash", Decimal("100"), "Test" ) result_dict = result.to_dict() self.assertFalse(result_dict["success"]) self.assertIsNone(result_dict["transaction_id"]) self.assertEqual(result_dict["error"]["code"], "TEST_ERROR") self.assertEqual(result_dict["error"]["message"], "Test rejection") self.assertEqual(result_dict["error"]["details"]["key"], "value") def test_to_dict_with_post_hook_warnings(self): """Test TransactionResult.to_dict() with POST hook warnings.""" def failing_post(context: HookContext) -> None: raise ValueError("POST failed") register_hook("fail", HookType.POST, failing_post, "add") result = self.service.add_point( self.user1.id, "cash", Decimal("100"), "Test" ) result_dict = result.to_dict() self.assertTrue(result_dict["success"]) self.assertIsNotNone(result_dict["transaction_id"]) self.assertIn("warnings", result_dict) self.assertEqual(len(result_dict["warnings"]), 1) self.assertEqual(result_dict["warnings"][0]["hook"], "fail") self.assertEqual(result_dict["warnings"][0]["code"], "HOOK_EXECUTION_ERROR")