from decimal import Decimal import calendar from django.utils import timezone from django.test import override_settings from django.urls import reverse from rest_framework.test import APITestCase from rest_framework import status from unittest.mock import patch, MagicMock from users.models import User from restaurant.models import ( Restaurant, MenuItem, Booking, BookingItem, LoyaltyProgram, LoyaltyRule, CustomerWallet, CoinTransaction, LoyaltySubscription, WalletTopup, KhataAccount, KhataTransaction, LoyaltyGroup, LoyaltyGroupMember, DeadHourDrop ) from restaurant.loyalty_services import ( WalletService, SubscriptionService, WalletTopupService, KhataService, VaultService, RulesEngine, IronLawsEnforcer, IronLawsError, get_or_create_wallet ) from restaurant.tasks import ( expire_loyalty_coins, expire_old_subscriptions, reset_khata_lifelines, check_overdue_khata, send_coin_expiry_warnings, process_dead_hour_drop ) @override_settings(CACHES={ 'default': { 'BACKEND': 'django.core.cache.backends.locmem.LocMemCache', 'LOCATION': 'unique-snowflake', } }) class SyankoLoyaltyTests(APITestCase): def setUp(self): # Create users self.owner = User.objects.create_user( username="test_owner", email="owner@test.com", password="password123", role="RESTAURANT_OWNER" ) self.customer = User.objects.create_user( username="test_customer", email="customer@test.com", password="password123", role="CUSTOMER", phone="919876543210" ) self.customer2 = User.objects.create_user( username="test_customer2", email="customer2@test.com", password="password123", role="CUSTOMER", phone="919876543211" ) self.admin = User.objects.create_user( username="test_admin", email="admin@test.com", password="password123", role="ADMIN", is_superuser=True ) # Create restaurant self.restaurant = Restaurant.objects.create( name="Test Restaurant", location="Dehradun", cuisine="North Indian", owner=self.owner, is_active=True ) # Create menu items self.food_item = MenuItem.objects.create( restaurant=self.restaurant, name="Paneer Tikka", price=Decimal("150.00"), category="MAIN_COURSE", is_available=True ) self.beverage_item = MenuItem.objects.create( restaurant=self.restaurant, name="Masala Chai", price=Decimal("50.00"), category="BEVERAGE", is_available=True ) # Create booking helper self.booking = Booking.objects.create( restaurant=self.restaurant, customer=self.customer, booking_type='TAKEAWAY', customer_name="Test Customer", contact_number="919876543210", status='CONFIRMED', payment_status='PAID', total_amount=Decimal("600.00") ) BookingItem.objects.create( booking=self.booking, menu_item=self.food_item, quantity=3, unit_price=Decimal("150.00") ) BookingItem.objects.create( booking=self.booking, menu_item=self.beverage_item, quantity=3, unit_price=Decimal("50.00") ) def test_phase1_foundation_welcome_referral_summit(self): # 1. Setup loyalty program via API self.client.force_authenticate(user=self.owner) setup_url = reverse('loyalty-setup', args=[self.restaurant.id]) response = self.client.post(setup_url, { 'program_name': 'Syanko <3U Loyalty', 'is_active': True, 'coin_value_in_rupees': 1.00, 'redemption_cap_percent': 50.00 }, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.restaurant.refresh_from_db() # 2. Setup rules (WELCOME_HOOK, REFERRAL, SUMMIT) rules_url = reverse('loyalty-rules', args=[self.restaurant.id]) # Welcome Hook: min spend 500, reward 100 response = self.client.post(rules_url, { 'rule_type': 'WELCOME_HOOK', 'title': 'Welcome Hook Rule', 'is_active': True, 'trigger_config': {'min_spend': 500, 'first_order_only': True}, 'reward_coins': 100, 'reward_expiry_hours': 168 }, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) # Referral: reward 50 response = self.client.post(rules_url, { 'rule_type': 'REFERRAL', 'title': 'Sherpa Referral Link', 'is_active': True, 'trigger_config': {'coins_per_referral': 50}, 'reward_coins': 50 }, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) # Summit VIP: target spend 7000, reward 500 response = self.client.post(rules_url, { 'rule_type': 'SUMMIT', 'title': 'Ultimate Summit VIP', 'is_active': True, 'trigger_config': {'target_spend': 7000, 'max_vips': 10}, 'reward_coins': 500 }, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) summit_rule = LoyaltyRule.objects.get(pk=response.data['id']) # 3. Customer B places order over 500 -> status goes to COMPLETED self.booking.refresh_from_db() self.booking.status = 'COMPLETED' self.booking.save() wallet = get_or_create_wallet(self.customer, self.restaurant) # Verify customer wallet credited self.assertEqual(wallet.coin_balance, Decimal('100.00')) # Verify transaction created txs = CoinTransaction.objects.filter(wallet=wallet, transaction_type='EARNED') self.assertTrue(txs.exists()) self.assertEqual(txs.first().coins, Decimal('100.00')) # 4. Customer B tries to redeem 200 coins on a ₹100 bill -> rejected (Iron Law 1: max 50% cap on ₹100 bill) self.client.force_authenticate(user=self.customer) booking_100 = Booking.objects.create( restaurant=self.restaurant, customer=self.customer, booking_type='TAKEAWAY', customer_name="Test Customer", contact_number="919876543210", status='PENDING', payment_status='UNPAID', total_amount=Decimal("100.00") ) redeem_url = reverse('loyalty-redeem', args=[self.restaurant.id]) response = self.client.post(redeem_url, { 'booking_id': booking_100.id, 'coins_to_redeem': 200 }, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) # Try to redeem 60 coins on ₹100 bill -> rejected (Iron Law 1) response = self.client.post(redeem_url, { 'booking_id': booking_100.id, 'coins_to_redeem': 60 }, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) # 5. Customer redeems 50 coins on ₹100 bill -> approved response = self.client.post(redeem_url, { 'booking_id': booking_100.id, 'coins_to_redeem': 50 }, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) wallet.refresh_from_db() self.assertEqual(wallet.coin_balance, Decimal('50.00')) # Verify transaction created tx_red = CoinTransaction.objects.filter(wallet=wallet, transaction_type='REDEEMED').first() self.assertIsNotNone(tx_red) self.assertEqual(tx_red.coins, Decimal('-50.00')) # 6. Referral code flow self.client.force_authenticate(user=self.customer2) ref_code_url = reverse('loyalty-referral-code') response = self.client.get(ref_code_url, {'restaurant_id': self.restaurant.id}) self.assertEqual(response.status_code, status.HTTP_200_OK) referral_code = response.data['referral_code'] # Apply code to referred user referred_user = User.objects.create_user( username="referred_user", email="ref@test.com", password="password123", role="CUSTOMER", phone="919876543219" ) self.client.force_authenticate(user=referred_user) apply_url = reverse('loyalty-referral-apply') response = self.client.post(apply_url, { 'referral_code': referral_code, 'restaurant_id': self.restaurant.id }, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) referred_wallet = CustomerWallet.objects.get(customer=referred_user, restaurant=self.restaurant) self.assertEqual(referred_wallet.referred_by, self.customer2) # Place first order for referred customer -> both get 50 coins ref_booking = Booking.objects.create( restaurant=self.restaurant, customer=referred_user, booking_type='TAKEAWAY', customer_name="Referred", contact_number="919876543219", status='CONFIRMED', payment_status='UNPAID', total_amount=Decimal("200.00") ) ref_booking.status = 'COMPLETED' ref_booking.payment_status = 'PAID' ref_booking.save() # Check referrer (customer2) wallet referrer_wallet = CustomerWallet.objects.get(customer=self.customer2, restaurant=self.restaurant) self.assertEqual(referrer_wallet.coin_balance, Decimal('50.00')) # Check referred customer wallet referred_wallet.refresh_from_db() self.assertEqual(referred_wallet.coin_balance, Decimal('50.00')) # 7. Summit VIP flow referred_wallet.lifetime_cash_spent = Decimal('7200.00') referred_wallet.save() RulesEngine._check_summit(summit_rule, referred_wallet) referred_wallet.refresh_from_db() self.assertTrue(referred_wallet.is_summit_vip) self.assertEqual(referred_wallet.coin_balance, Decimal('550.00')) # 50 + 500 summit reward # Verify SummitVIPList view includes referred customer self.client.force_authenticate(user=self.owner) vips_url = reverse('loyalty-summit-vips', args=[self.restaurant.id]) response = self.client.get(vips_url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(len(response.data), 1) self.assertEqual(response.data[0]['username'], referred_user.username) def test_phase2_subscription_and_topup(self): program = LoyaltyProgram.objects.create( restaurant=self.restaurant, program_name="Test Program", is_active=True, coin_value_in_rupees=1.00 ) # 1. Create subscription rule sub_rule = LoyaltyRule.objects.create( program=program, rule_type='SUBSCRIPTION', title='Liquid Gold Pass', is_active=True, trigger_config={'monthly_price': 399, 'min_food_pair': 99, 'daily_item_category': 'BEVERAGE'} ) # 2. Customer subscribes self.client.force_authenticate(user=self.customer) sub_url = reverse('loyalty-subscribe', args=[self.restaurant.id]) response = self.client.post(sub_url, { 'rule_id': sub_rule.id, 'payment_reference': 'pay_12345' }, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) self.assertEqual(response.data['status'], 'ACTIVE') # Verify subscription in database wallet = get_or_create_wallet(self.customer, self.restaurant) sub = LoyaltySubscription.objects.filter(wallet=wallet).first() self.assertIsNotNone(sub) self.assertEqual(sub.status, 'ACTIVE') self.assertEqual(sub.amount_paid, Decimal('399.00')) # 3. Customer places order with food + beverage on the same day -> check entitlement booking = Booking.objects.create( restaurant=self.restaurant, customer=self.customer, booking_type='TAKEAWAY', customer_name="Cust", contact_number="919876543210", status='CONFIRMED', payment_status='PAID', total_amount=Decimal("200.00") ) BookingItem.objects.create(booking=booking, menu_item=self.food_item, quantity=1, unit_price=Decimal("150.00")) BookingItem.objects.create(booking=booking, menu_item=self.beverage_item, quantity=1, unit_price=Decimal("50.00")) entitled = SubscriptionService.check_entitlement(wallet, booking) self.assertTrue(entitled) sub.refresh_from_db() self.assertEqual(sub.last_used_date, timezone.now().date()) self.assertEqual(sub.total_uses, 1) # Try to use again on same day entitled_again = SubscriptionService.check_entitlement(wallet, booking) self.assertFalse(entitled_again) # 4. Topup bonus rule topup_rule = LoyaltyRule.objects.create( program=program, rule_type='WALLET_TOPUP', title='Topup Bonus', is_active=True, trigger_config={'min_topup': 1000, 'bonus_percent': 50} ) # Top up 1000 -> get 1500 coins topup_url = reverse('loyalty-topup', args=[self.restaurant.id]) response = self.client.post(topup_url, { 'rule_id': topup_rule.id, 'cash_amount': 1000, 'payment_reference': 'top_12345' }, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) wallet.refresh_from_db() self.assertEqual(wallet.coin_balance, Decimal('1500.00')) self.assertEqual(response.data['new_balance'], '1500.00') # Top up 500 (below minimum) -> rejected response = self.client.post(topup_url, { 'rule_id': topup_rule.id, 'cash_amount': 500 }, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_phase3_khata_and_groups(self): program = LoyaltyProgram.objects.create( restaurant=self.restaurant, program_name="Test Program", is_active=True, coin_value_in_rupees=1.00 ) wallet = get_or_create_wallet(self.customer, self.restaurant) # 1. Customer gets Khata status -> active, 5 lifelines self.client.force_authenticate(user=self.customer) khata_url = reverse('loyalty-khata-status', args=[self.restaurant.id]) response = self.client.get(khata_url) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['lifelines_remaining'], 5) self.assertEqual(response.data['status'], 'ACTIVE') # 2. Customer uses lifeline on a booking booking = Booking.objects.create( restaurant=self.restaurant, customer=self.customer, booking_type='TAKEAWAY', customer_name="Cust", contact_number="919876543210", status='CONFIRMED', payment_status='UNPAID', total_amount=Decimal("200.00") ) use_khata_url = reverse('loyalty-khata-use', args=[self.restaurant.id]) response = self.client.post(use_khata_url, { 'booking_id': booking.id }, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) # Verify KhataTransaction created khata = KhataAccount.objects.get(wallet=wallet) tx = KhataTransaction.objects.filter(khata=khata, booking=booking).first() self.assertIsNotNone(tx) self.assertEqual(khata.lifelines_used, 1) self.assertEqual(khata.lifelines_remaining, 4) # Verify booking status booking.refresh_from_db() self.assertEqual(booking.payment_status, 'UNPAID') # Verify tab due date (1st of next month) today = timezone.now().date() if today.month == 12: expected_due = today.replace(year=today.year + 1, month=1, day=1) else: expected_due = today.replace(month=today.month + 1, day=1) self.assertEqual(khata.tab_due_date, expected_due) # 3. Customer uses remaining lifelines -> 6th attempt rejected for i in range(4): bk = Booking.objects.create( restaurant=self.restaurant, customer=self.customer, booking_type='TAKEAWAY', total_amount=Decimal("10.00") ) KhataService.use_lifeline(wallet, bk) khata.refresh_from_db() self.assertEqual(khata.lifelines_remaining, 0) bk_6 = Booking.objects.create( restaurant=self.restaurant, customer=self.customer, booking_type='TAKEAWAY', total_amount=Decimal("10.00") ) with self.assertRaises(ValueError): KhataService.use_lifeline(wallet, bk_6) # 4. Check overdue Celery task # Manually set tab_due_date to past date khata.tab_due_date = today - timezone.timedelta(days=2) khata.save() check_overdue_khata() khata.refresh_from_db() self.assertEqual(khata.status, 'OVERDUE') khata.tab_due_date = today - timezone.timedelta(days=20) khata.save() check_overdue_khata() khata.refresh_from_db() self.assertEqual(khata.status, 'BLOCKED') # 5. Clear tab clear_url = reverse('loyalty-khata-clear', args=[self.restaurant.id]) response = self.client.post(clear_url, {}, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) khata.refresh_from_db() self.assertEqual(khata.current_tab_amount, Decimal('0.00')) self.assertEqual(khata.status, 'ACTIVE') self.assertTrue(all(tx.is_cleared for tx in KhataTransaction.objects.filter(khata=khata))) # 6. Reset Celery task khata.lifelines_used = 3 khata.save() reset_khata_lifelines() khata.refresh_from_db() self.assertEqual(khata.lifelines_used, 0) # 7. Group Savings Flow # Owner creates LoyaltyGroup self.client.force_authenticate(user=self.owner) group_url = reverse('loyalty-groups', args=[self.restaurant.id]) response = self.client.post(group_url, { 'group_name': 'Tech Guild', 'vault_percentage': 5.00 }, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) group_id = response.data['id'] group = LoyaltyGroup.objects.get(pk=group_id) # Customer joins group self.client.force_authenticate(user=self.customer) join_url = reverse('loyalty-group-join') response = self.client.post(join_url, { 'group_id': group.id, 'restaurant_id': self.restaurant.id }, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertTrue(LoyaltyGroupMember.objects.filter(group=group, wallet=wallet).exists()) # Customer places order -> booking completes -> adds to group vault booking_grp = Booking.objects.create( restaurant=self.restaurant, customer=self.customer, booking_type='TAKEAWAY', total_amount=Decimal("1000.00"), payment_status='UNPAID' ) booking_grp.status = 'COMPLETED' booking_grp.payment_status = 'PAID' booking_grp.save() group.refresh_from_db() self.assertEqual(group.vault_balance, Decimal('50.00')) # Owner redeems from vault self.client.force_authenticate(user=self.owner) redeem_vault_url = reverse('loyalty-vault-redeem', args=[self.restaurant.id, group.id]) response = self.client.post(redeem_vault_url, { 'amount': 30.00, 'notes': 'Catering discount' }, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) group.refresh_from_db() self.assertEqual(group.vault_balance, Decimal('20.00')) # Try to redeem more than vault balance -> rejected response = self.client.post(redeem_vault_url, { 'amount': 50.00, 'notes': 'Overdraft' }, format='json') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) def test_phase4_drops_and_analytics(self): program = LoyaltyProgram.objects.create( restaurant=self.restaurant, program_name="Test Program", is_active=True, coin_value_in_rupees=1.00 ) wallet = get_or_create_wallet(self.customer, self.restaurant) # 1. Owner sends Dead-Hour Drop self.client.force_authenticate(user=self.owner) drop_url = reverse('loyalty-drop-create', args=[self.restaurant.id]) with patch('restaurant.whatsapp_service._send') as mock_send: mock_send.return_value = True response = self.client.post(drop_url, { 'message': 'Noon special: 30% off on mains!', 'discount_code': 'NOON30', 'valid_until': (timezone.now() + timezone.timedelta(hours=2)).isoformat() }, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) # Verify drop record drop = DeadHourDrop.objects.filter(restaurant=self.restaurant).first() self.assertIsNotNone(drop) self.assertEqual(drop.total_recipients, 1) # 2. GET Full Analytics endpoint CoinTransaction.objects.create(wallet=wallet, transaction_type='EARNED', coins=Decimal('300.00')) booking_red = Booking.objects.create( restaurant=self.restaurant, customer=self.customer, booking_type='TAKEAWAY', total_amount=Decimal("400.00") ) CoinTransaction.objects.create(wallet=wallet, transaction_type='REDEEMED', coins=Decimal('-100.00'), source_booking=booking_red) analytics_url = reverse('loyalty-full-analytics', args=[self.restaurant.id]) response = self.client.get(analytics_url) self.assertEqual(response.status_code, status.HTTP_200_OK) fields = [ 'total_wallets', 'active_wallets_30d', 'coins_issued_total', 'coins_redeemed_total', 'redemption_rate_percent', 'revenue_influenced', 'summit_vips', 'coins_expiring_24h', 'avg_coins_per_customer', 'top_10_customers' ] for f in fields: self.assertIn(f, response.data) self.assertEqual(response.data['total_wallets'], 1) self.assertEqual(float(response.data['revenue_influenced']), 400.00) @patch('restaurant.whatsapp_service.send_reply') def test_whatsapp_webhook_state_machine(self, mock_send_reply): # Configure loyalty program and items program = LoyaltyProgram.objects.create( restaurant=self.restaurant, program_name="Syanko Program", is_active=True, coin_value_in_rupees=1.00, redemption_cap_percent=50.00 ) wallet = get_or_create_wallet(self.customer, self.restaurant) WalletService.credit_coins(wallet, Decimal('100.00'), 'EARNED') # Webhook URL webhook_url = reverse('whatsapp-webhook') phone = "919876543210" def send_msg(text): response = self.client.post(webhook_url, { 'waId': phone, 'type': 'text', 'text': text }, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.json(), {'status': 'ok'}) args, kwargs = mock_send_reply.call_args reply = args[1] mock_send_reply.reset_mock() return reply # 1. State: IDLE -> AWAITING_RESTAURANT reply = send_msg("Hi") self.assertIn("Which restaurant", reply) # 2. State: AWAITING_RESTAURANT -> MAIN_MENU reply = send_msg(str(self.restaurant.id)) self.assertIn("Test Restaurant", reply) self.assertIn("View Menu", reply) # 3. State: MAIN_MENU -> MENU_CATEGORIES reply = send_msg("1") self.assertIn("Menu", reply) self.assertIn("Main Course", reply) # 4. State: MENU_CATEGORIES -> MENU_ITEMS reply = send_msg("2") self.assertIn("Paneer Tikka", reply) # 5. State: MENU_ITEMS -> MENU_ITEMS reply = send_msg("1 2") self.assertIn("Added", reply) self.assertIn("Cart total: ₹300", reply) # 6. State: MENU_ITEMS -> CART reply = send_msg("done") self.assertIn("Your Cart", reply) self.assertIn("Checkout", reply) self.assertIn("100 coins", reply) # 7. State: CART -> AWAITING_ORDER_TYPE reply = send_msg("1") self.assertIn("How would you like your order?", reply) # 7b. State: AWAITING_ORDER_TYPE -> CHECKOUT_COINS reply = send_msg("1") # Takeaway self.assertIn("Use Loyalty Coins?", reply) self.assertIn("use up to *100 coins*", reply) # 8. State: CHECKOUT_COINS -> CHECKOUT_PAYMENT reply = send_msg("YES") self.assertIn("Payment", reply) self.assertIn("Coins discount: -₹100", reply) self.assertIn("You pay: ₹200", reply) # 9. State: CHECKOUT_PAYMENT -> COMPLETED reply = send_msg("1") self.assertIn("Confirmed", reply) self.assertIn("You pay: ₹200", reply) # Verify Booking created bookings = Booking.objects.filter(customer=self.customer, restaurant=self.restaurant).order_by('-created_at') booking = bookings.first() self.assertIsNotNone(booking) self.assertEqual(booking.status, 'CONFIRMED') # Verify coins deducted wallet.refresh_from_db() self.assertEqual(wallet.coin_balance, Decimal('0.00')) def test_iron_laws(self): program = LoyaltyProgram.objects.create( restaurant=self.restaurant, program_name="Test Program", is_active=True, coin_value_in_rupees=1.00, redemption_cap_percent=50.00 ) wallet = get_or_create_wallet(self.customer, self.restaurant) # Iron Law 1: Try to redeem coins worth 60% of a ₹1000 bill -> rejected WalletService.credit_coins(wallet, Decimal('600.00'), 'EARNED') booking_1000 = Booking.objects.create( restaurant=self.restaurant, customer=self.customer, booking_type='TAKEAWAY', total_amount=Decimal("1000.00") ) with self.assertRaises(IronLawsError): IronLawsEnforcer.validate_redemption(wallet, booking_1000.total_amount, Decimal('600.00')) # Iron Law 2: Place order matching both WELCOME_HOOK and MILESTONE_SPEND -> Only higher-value rule fires self.restaurant.loyalty_program.rules.all().delete() welcome_rule = LoyaltyRule.objects.create( program=program, rule_type='WELCOME_HOOK', title='Welcome Drop', is_active=True, trigger_config={'min_spend': 500, 'first_order_only': True}, reward_coins=100 ) milestone_rule = LoyaltyRule.objects.create( program=program, rule_type='MILESTONE_SPEND', title='Milestone Spend', is_active=True, trigger_config={'order_amount': 500}, reward_coins=200 ) wallet.total_orders = 0 wallet.coin_balance = 0 wallet.save() booking_stack = Booking.objects.create( restaurant=self.restaurant, customer=self.customer, booking_type='TAKEAWAY', total_amount=Decimal("600.00") ) RulesEngine.evaluate_booking(booking_stack) wallet.refresh_from_db() self.assertEqual(wallet.coin_balance, Decimal('200.00')) # Iron Law 3: Create a CoinTransaction with expires_at = 1 hour ago -> balance does not include those coins wallet.coin_balance = Decimal('100.00') wallet.save() CoinTransaction.objects.create( wallet=wallet, transaction_type='EARNED', coins=Decimal('50.00'), expires_at=timezone.now() - timezone.timedelta(hours=1) ) wallet.coin_balance = Decimal('150.00') wallet.save() live_balance = WalletService.get_live_balance(wallet) self.assertEqual(live_balance, Decimal('100.00')) # Iron Law 4: Try to call VaultRedeemView as a customer (not owner) -> 403 self.client.force_authenticate(user=self.customer) group = LoyaltyGroup.objects.create( restaurant=self.restaurant, admin=self.owner, group_name="G1", vault_balance=Decimal("100.00") ) vault_url = reverse('loyalty-vault-redeem', args=[self.restaurant.id, group.id]) response = self.client.post(vault_url, { 'amount': 30.00, 'notes': 'customer tries to redeem' }, format='json') self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) @patch('requests.post') def test_whatsapp_notifications_and_delivery(self, mock_post): mock_response = MagicMock() mock_response.status_code = 200 mock_post.return_value = mock_response with patch('restaurant.whatsapp_service.WHATSAPP_PROVIDER', 'wati'), \ patch('restaurant.whatsapp_service.WAPI_URL', 'http://wati.api'), \ patch('restaurant.whatsapp_service.WAPI_KEY', 'key123'): # 1. send_reply from restaurant.whatsapp_service import send_reply success = send_reply("919876543210", "Hello there") self.assertTrue(success) mock_post.assert_called_with( 'http://wati.api/api/v1/sendSessionMessage?whatsappNumber=919876543210', json={'messageText': 'Hello there'}, headers={'Authorization': 'Bearer key123'}, timeout=10 ) mock_post.reset_mock() # 2. welcome_coins template program = LoyaltyProgram.objects.create( restaurant=self.restaurant, program_name="Welcome Program", is_active=True ) wallet = get_or_create_wallet(self.customer, self.restaurant) from restaurant.whatsapp_service import send_welcome_message success = send_welcome_message(wallet) self.assertTrue(success) mock_post.assert_called() called_json = mock_post.call_args[1]['json'] self.assertEqual(called_json['template_name'], 'welcome_coins') mock_post.reset_mock() # 3. coin_expiry_warning template CoinTransaction.objects.create( wallet=wallet, transaction_type='EARNED', coins=Decimal('50.00'), expires_at=timezone.now() + timezone.timedelta(hours=10) ) from restaurant.whatsapp_service import send_coin_expiry_warning success = send_coin_expiry_warning(wallet) self.assertTrue(success) mock_post.assert_called() called_json = mock_post.call_args[1]['json'] self.assertEqual(called_json['template_name'], 'coin_expiry_warning') mock_post.reset_mock() # 4. dead_hour_drop template from restaurant.whatsapp_service import send_dead_hour_drop sent = send_dead_hour_drop(self.restaurant, "Noon Drop", "CODE", 2) self.assertEqual(sent, 1) mock_post.assert_called() called_json = mock_post.call_args[1]['json'] self.assertEqual(called_json['template_name'], 'dead_hour_drop') @patch('requests.post') def test_multi_provider_whatsapp_outbound(self, mock_post): mock_response = MagicMock() mock_response.status_code = 200 mock_post.return_value = mock_response from restaurant import whatsapp_service as wa # Test Meta Provider with patch('restaurant.whatsapp_service.META_ACCESS_TOKEN', 'token123'), \ patch('restaurant.whatsapp_service.META_PHONE_NUMBER_ID', 'id123'), \ patch('restaurant.whatsapp_service.META_LANGUAGE_CODE', 'en'), \ patch('restaurant.whatsapp_service.WHATSAPP_PROVIDER', 'meta'): success = wa._send("919876543210", "welcome_coins", ["User", "Rest", "100"]) self.assertTrue(success) mock_post.assert_called_with( 'https://graph.facebook.com/v17.0/id123/messages', json={ 'messaging_product': 'whatsapp', 'recipient_type': 'individual', 'to': '919876543210', 'type': 'template', 'template': { 'name': 'welcome_coins', 'language': {'code': 'en'}, 'components': [{'type': 'body', 'parameters': [{'type': 'text', 'text': 'User'}, {'type': 'text', 'text': 'Rest'}, {'type': 'text', 'text': '100'}]}] } }, headers={'Authorization': 'Bearer token123', 'Content-Type': 'application/json'}, timeout=10 ) mock_post.reset_mock() success = wa.send_reply("919876543210", "Hello there") self.assertTrue(success) mock_post.assert_called_with( 'https://graph.facebook.com/v17.0/id123/messages', json={ 'messaging_product': 'whatsapp', 'recipient_type': 'individual', 'to': '919876543210', 'type': 'text', 'text': {'body': 'Hello there'} }, headers={'Authorization': 'Bearer token123', 'Content-Type': 'application/json'}, timeout=10 ) mock_post.reset_mock() # Test Twilio Provider with patch('restaurant.whatsapp_service.TWILIO_ACCOUNT_SID', 'sid123'), \ patch('restaurant.whatsapp_service.TWILIO_AUTH_TOKEN', 'auth123'), \ patch('restaurant.whatsapp_service.TWILIO_FROM_NUMBER', '+14155238886'), \ patch('restaurant.whatsapp_service.WHATSAPP_PROVIDER', 'twilio'): success = wa.send_reply("919876543210", "Hello there") self.assertTrue(success) mock_post.assert_called_with( 'https://api.twilio.com/2010-04-01/Accounts/sid123/Messages.json', data={ 'From': 'whatsapp:+14155238886', 'To': 'whatsapp:+919876543210', 'Body': 'Hello there' }, auth=('sid123', 'auth123'), headers={'Content-Type': 'application/x-www-form-urlencoded'}, timeout=10 ) mock_post.reset_mock() success = wa._send("919876543210", "welcome_coins", ["User", "Rest", "100"]) self.assertTrue(success) mock_post.assert_called_with( 'https://api.twilio.com/2010-04-01/Accounts/sid123/Messages.json', data={ 'From': 'whatsapp:+14155238886', 'To': 'whatsapp:+919876543210', 'Body': 'Hello User, welcome to Rest! You have earned 100 welcome coins. 🪙' }, auth=('sid123', 'auth123'), headers={'Content-Type': 'application/x-www-form-urlencoded'}, timeout=10 ) mock_post.reset_mock() # Test Gupshup Provider with patch('restaurant.whatsapp_service.GUPSHUP_API_KEY', 'gskey123'), \ patch('restaurant.whatsapp_service.GUPSHUP_FROM_NUMBER', '919876543211'), \ patch('restaurant.whatsapp_service.GUPSHUP_APP_NAME', 'myapp'), \ patch('restaurant.whatsapp_service.WHATSAPP_PROVIDER', 'gupshup'): success = wa._send("919876543210", "welcome_coins", ["User", "Rest", "100"]) self.assertTrue(success) mock_post.assert_called_with( 'https://api.gupshup.io/sm/api/v1/template/msg', data={ 'source': '919876543211', 'destination': '919876543210', 'template': '{"id": "welcome_coins", "params": ["User", "Rest", "100"]}' }, headers={'apikey': 'gskey123', 'Content-Type': 'application/x-www-form-urlencoded'}, timeout=10 ) mock_post.reset_mock() success = wa.send_reply("919876543210", "Hello there") self.assertTrue(success) mock_post.assert_called_with( 'https://api.gupshup.io/sm/api/v1/msg', data={ 'channel': 'whatsapp', 'source': '919876543211', 'destination': '919876543210', 'message': '{"isHSM": "false", "type": "text", "text": "Hello there"}', 'src.name': 'myapp' }, headers={'apikey': 'gskey123', 'Content-Type': 'application/x-www-form-urlencoded'}, timeout=10 ) @patch('restaurant.whatsapp_service.send_reply') def test_multi_provider_whatsapp_webhook(self, mock_send_reply): from unittest.mock import ANY webhook_url = reverse('whatsapp-webhook') # 1. Test Meta Webhook Verification (GET) with patch('os.environ.get', return_value='my_secret_verify_token'): response = self.client.get(webhook_url, { 'hub.mode': 'subscribe', 'hub.verify_token': 'my_secret_verify_token', 'hub.challenge': 'my_challenge_123' }) self.assertEqual(response.status_code, 200) self.assertEqual(response.content.decode(), 'my_challenge_123') response = self.client.get(webhook_url, { 'hub.mode': 'subscribe', 'hub.verify_token': 'wrong_token', 'hub.challenge': 'my_challenge_123' }) self.assertEqual(response.status_code, 403) # 2. Test Meta Webhook Inbound Message (POST) meta_payload = { "object": "whatsapp_business_account", "entry": [ { "id": "123456", "changes": [ { "value": { "messaging_product": "whatsapp", "messages": [ { "from": "919876543210", "type": "text", "text": {"body": "Hi"} } ] }, "field": "messages" } ] } ] } response = self.client.post(webhook_url, meta_payload, format='json') self.assertEqual(response.status_code, 200) self.assertEqual(response.json(), {'status': 'ok'}) mock_send_reply.assert_called_with('919876543210', ANY) mock_send_reply.reset_mock() # 3. Test Twilio Webbound Inbound Message (POST - x-www-form-urlencoded) import urllib.parse twilio_payload = { 'From': 'whatsapp:+919876543210', 'Body': 'Hi' } encoded_data = urllib.parse.urlencode(twilio_payload) response = self.client.post(webhook_url, encoded_data, content_type='application/x-www-form-urlencoded') self.assertEqual(response.status_code, 200) self.assertEqual(response.content.decode(), '') mock_send_reply.assert_called_with('+919876543210', ANY) mock_send_reply.reset_mock() # 4. Test Gupshup Webhook Inbound Message (POST) gupshup_payload = { "app": "DemoApp", "type": "message", "payload": { "source": "919876543210", "type": "text", "payload": {"text": "Hi"} } } response = self.client.post(webhook_url, gupshup_payload, format='json') self.assertEqual(response.status_code, 200) self.assertEqual(response.json(), {'status': 'ok'}) mock_send_reply.assert_called_with('919876543210', ANY)