import hashlib import hmac import json import requests from unittest.mock import patch from django.test import TransactionTestCase, Client, RequestFactory, override_settings from django.urls import resolve, reverse from unittest.mock import MagicMock from restaurant.models import Restaurant, AaharConfig from restaurant import whatsapp_service from restaurant.whatsapp_session import get_session, set_session from restaurant.whatsapp_webhook import ConversationHandler, WhatsAppWebhookView, _extract_message_meta from users.models import User LOCAL_MEMORY_CACHE = { "default": { "BACKEND": "django.core.cache.backends.locmem.LocMemCache", "LOCATION": "interakt-webhook-tests", } } @override_settings(CACHES=LOCAL_MEMORY_CACHE) class InteraktWhatsAppWebhookTests(TransactionTestCase): def setUp(self): self.factory = RequestFactory() def test_extracts_interakt_text_message(self): payload = { "type": "message_received", "data": { "customer": {"phoneNumber": "919876543210"}, "message": {"id": "wamid.1", "type": "text", "text": {"body": "Hi"}}, }, } request = self.factory.post( "/api/restaurants/whatsapp/webhook/", data=json.dumps(payload), content_type="application/json", ) phone, text, message_id = _extract_message_meta(request) self.assertEqual(phone, "919876543210") self.assertEqual(text, "Hi") self.assertEqual(message_id, "wamid.1") @override_settings(INTERAKT_WEBHOOK_SECRET="whsec") def test_rejects_interakt_webhook_with_bad_signature(self): payload = {"type": "message_received", "data": {"phoneNumber": "919876543210", "message": "Hi"}} body = json.dumps(payload).encode("utf-8") request = self.factory.generic( "POST", "/api/restaurants/whatsapp/webhook/", data=body, content_type="application/json", HTTP_X_INTERAKT_SIGNATURE="sha256=bad", ) response = WhatsAppWebhookView.as_view()(request) self.assertEqual(response.status_code, 401) @override_settings(INTERAKT_WEBHOOK_SECRET="whsec") @patch("restaurant.whatsapp_webhook.wa.send_reply") @patch("restaurant.whatsapp_webhook._handler.handle", return_value="ok") def test_accepts_interakt_webhook_with_valid_signature(self, mock_handle, mock_send_reply): payload = { "type": "message_received", "data": { "customer": {"phoneNumber": "919876543210"}, "message": {"id": "wamid.2", "message": "Hi"}, }, } body = json.dumps(payload).encode("utf-8") signature = hmac.new(b"whsec", body, hashlib.sha256).hexdigest() request = self.factory.generic( "POST", "/api/restaurants/whatsapp/webhook/", data=body, content_type="application/json", HTTP_X_INTERAKT_SIGNATURE=f"sha256={signature}", ) response = WhatsAppWebhookView.as_view()(request) self.assertEqual(response.status_code, 200) mock_handle.assert_called_once_with("919876543210", "Hi") mock_send_reply.assert_called_once_with("919876543210", "ok") @patch("restaurant.whatsapp_service._send_reply_interakt", return_value=True) @patch("restaurant.whatsapp_service.requests.post", side_effect=requests.RequestException("bad interactive")) def test_interakt_list_falls_back_to_text(self, mock_post, mock_reply): with patch("restaurant.whatsapp_service.WHATSAPP_PROVIDER", "interakt"), \ patch("restaurant.whatsapp_service.INTERAKT_SECRET_KEY", "secret"): sent = whatsapp_service.send_interactive_list( "919876543210", "Pick a restaurant", "Select", [{"title": "Available", "rows": [{"id": "rest_1", "title": "Cafe", "description": "MG Road"}]}], ) self.assertTrue(sent) mock_reply.assert_called_once() self.assertIn("Cafe", mock_reply.call_args.args[1]) @patch("restaurant.whatsapp_webhook.wa.send_interactive_buttons") def test_restaurant_number_reply_uses_presented_options(self, mock_buttons): user = User.objects.create_user(username="wa_919876543210", phone="919876543210", role="CUSTOMER") restaurant = Restaurant.objects.create(name="Late Id Cafe", location="MG Road", cuisine="Indian") set_session("919876543210", { "state": "AWAITING_RESTAURANT", "user_id": user.id, "phone": "919876543210", "restaurant_options": [restaurant.id], }) handler = ConversationHandler() reply = handler._s_awaiting_restaurant( "919876543210", "1", get_session("919876543210"), {"intent": "UNKNOWN"}, ) self.assertEqual(reply, "") self.assertEqual(get_session("919876543210")["restaurant_id"], restaurant.id) mock_buttons.assert_called_once() @override_settings(INTERAKT_WEBHOOK_SECRET="whsec") def test_webhook_get_health_response(self): request = self.factory.get("/api/restaurants/whatsapp/webhook/") response = WhatsAppWebhookView.as_view()(request) self.assertEqual(response.status_code, 200) self.assertIn(b'"status": "ok"', response.content) self.assertIn(b'"webhook_secret_required": true', response.content) @override_settings( WHATSAPP_PROVIDER="interakt", INTERAKT_SECRET_KEY="interakt-secret", TWILIO_ACCOUNT_SID="twilio-sid", TWILIO_AUTH_TOKEN="twilio-token", ) def test_health_prefers_configured_interakt_over_twilio(self): request = self.factory.get("/api/restaurants/whatsapp/webhook/") response = WhatsAppWebhookView.as_view()(request) self.assertEqual(response.status_code, 200) self.assertIn(b'"provider": "interakt"', response.content) def test_root_webhook_aliases_resolve(self): for path in ( "/api/restaurants/whatsapp/webhook/", "/api/whatsapp/webhook/", "/whatsapp/webhook/", "/webhook/whatsapp/", "/interakt/webhook/", ): self.assertEqual(resolve(path).func.view_class, WhatsAppWebhookView) class DirectAaharIntegrationTests(TransactionTestCase): def setUp(self): self.client = Client() # Create a user to own the restaurant if needed, though owner can be null. # Clean up existing data first AaharConfig.objects.all().delete() Restaurant.objects.all().delete() self.restaurant = Restaurant.objects.create( name="Test Restaurant", location="Test City", cuisine="Indian" ) def test_direct_aahar_restaurant_list_not_configured(self): # Ensure no AaharConfig is active url = reverse('direct-aahar-restaurants') response = self.client.get(url) self.assertEqual(response.status_code, 404) self.assertEqual(response.json(), {"error": "Aahar not configured"}) def test_direct_aahar_restaurant_list_single_tenant(self): # Create active config in single tenant mode (is_multitenant=False) config = AaharConfig.objects.create( restaurant=self.restaurant, aahar_restaurant_service_id="17", aahar_domain="http://localhost:8001", is_multitenant=False, is_active=True ) url = reverse('direct-aahar-restaurants') response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertEqual( response.json(), {"data": [{"restaurant_service_id": "17", "name": "Test Restaurant"}]} ) @patch('requests.get') def test_direct_aahar_restaurant_list_multi_tenant(self, mock_get): # Create active config in multi tenant mode (is_multitenant=True) config = AaharConfig.objects.create( restaurant=self.restaurant, aahar_restaurant_service_id="17", aahar_domain="http://localhost:8001", aahar_api_token="testtoken123", is_multitenant=True, is_active=True ) mock_response = MagicMock() mock_response.json.return_value = { "status": "success", "data": [ {"restaurant_service_id": "17", "name": "Aahar POS"} ] } mock_get.return_value = mock_response url = reverse('direct-aahar-restaurants') response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["status"], "success") mock_get.assert_called_once_with( "http://localhost:8001/apis/backend/web/get-all-restaurants", headers={"Access-Token": "testtoken123"}, timeout=5 ) def test_direct_aahar_menu_not_configured(self): # Ensure no AaharConfig is active url = reverse('direct-aahar-menu', args=[17]) response = self.client.get(url) self.assertEqual(response.status_code, 404) self.assertEqual(response.json(), {"error": "Aahar not configured"}) @patch('requests.get') def test_direct_aahar_menu_success(self, mock_get): config = AaharConfig.objects.create( restaurant=self.restaurant, aahar_restaurant_service_id="17", aahar_domain="http://localhost:8001", aahar_api_token="testtoken123", is_multitenant=True, is_active=True ) mock_response = MagicMock() mock_response.json.return_value = { "status": "success", "menu": [{"id": 1, "name": "Paneer Tikka"}] } mock_get.return_value = mock_response url = reverse('direct-aahar-menu', args=[17]) response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["status"], "success") mock_get.assert_called_once_with( "http://localhost:8001/apis/backend/web/export-menu?restaurant_id=17", headers={"Access-Token": "testtoken123"}, timeout=5 ) def test_direct_aahar_order_not_configured(self): url = reverse('direct-aahar-order', args=[17]) response = self.client.post(url, {}, content_type='application/json') self.assertEqual(response.status_code, 404) @patch('requests.post') def test_direct_aahar_order_success(self, mock_post): mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = { "status": "success", "forwarded_payload": {"restaurant_id": 17} } mock_post.return_value = mock_response config = AaharConfig.objects.create( restaurant=self.restaurant, aahar_restaurant_service_id="17", aahar_domain="http://localhost:8001", is_multitenant=False, is_active=True ) url = reverse('direct-aahar-order', args=[17]) payload = {"items": [{"item_id": 1, "qty": 2}]} response = self.client.post(url, payload, content_type='application/json') self.assertEqual(response.status_code, 200) res_data = response.json() self.assertEqual(res_data["status"], "success") self.assertEqual(res_data["forwarded_payload"]["restaurant_id"], 17) def test_direct_aahar_order_status_webhook_invalid(self): url = reverse('direct-aahar-order-status-webhook') response = self.client.post(url, {}, content_type='application/json') self.assertEqual(response.status_code, 400) self.assertEqual(response.json(), {"error": "order_id and status required"}) def test_direct_aahar_order_status_webhook_success(self): url = reverse('direct-aahar-order-status-webhook') payload = {"order_id": 9988, "status": "Preparing"} response = self.client.post(url, payload, content_type='application/json') self.assertEqual(response.status_code, 200) self.assertEqual(response.json(), {"status": "acknowledged"}) # Verify database event_outbox entry from django.db import connection with connection.cursor() as cursor: cursor.execute("SELECT topic, payload FROM event_outbox WHERE topic='food_order_status_updated'") rows = cursor.fetchall() self.assertEqual(len(rows), 1) import json payload_db = json.loads(rows[0][1]) self.assertEqual(payload_db["order_id"], 9988) self.assertEqual(payload_db["status"], "Preparing") # Cleanup outbox table with connection.cursor() as cursor: cursor.execute("DROP TABLE event_outbox")