from datetime import datetime, timezone from decimal import Decimal from unittest.mock import patch from django.test import TestCase, override_settings from orders.models import Order from webhooks.models import ShopifyFulfillmentEvent, ShopifyRefundEvent from webhooks.services.webhook_service import WebhookService class WebhookServiceAmountTests(TestCase): def test_extract_refund_amount_prefers_refund_transactions(self): payload = { "transactions": [ {"kind": "capture", "amount": "99.00"}, {"kind": "refund", "amount": "10.50"}, {"kind": "refund", "amount": "1.25"}, ] } self.assertEqual( WebhookService._extract_refund_amount(payload), Decimal("11.75"), ) def test_extract_refund_amount_falls_back_to_line_items(self): payload = { "refund_line_items": [ {"subtotal": "8.00", "total_tax": "1.52"}, {"subtotal": "2.00", "total_tax": "0.48"}, ] } self.assertEqual( WebhookService._extract_refund_amount(payload), Decimal("12.00"), ) class WebhookServiceDigistoreResultTests(TestCase): def test_is_digistore_refund_successful_handles_refused_details(self): api_result = { "result": "success", "data": { "purchase_id": "ABC", "details": [ { "status": "refused", "code": "refund_right_expired", } ], "refund_amount": 0, }, } ok, reason = WebhookService._is_digistore_refund_successful(api_result) self.assertFalse(ok) self.assertIn("details_refused", reason) def test_is_digistore_refund_successful_accepts_normal_success(self): api_result = { "result": "success", "data": {"status": "success", "refund_amount": 10.0}, } ok, reason = WebhookService._is_digistore_refund_successful(api_result) self.assertTrue(ok) self.assertEqual(reason, "") class HandleRefundCreatedTests(TestCase): def test_handle_refund_created_requires_mapping(self): payload = {"id": "r1", "order_id": "shop-1", "transactions": [{"kind": "refund", "amount": "5"}]} result = WebhookService.handle_refund_created(payload, webhook_id="wh-1") self.assertFalse(result["ok"]) event = ShopifyRefundEvent.objects.get(webhook_id="wh-1") self.assertEqual(event.error, "order_mapping_not_found") self.assertFalse(event.processed) @patch("orders.services.order_service.OrderService.refund_purchase") def test_handle_refund_created_marks_full_refund_on_success(self, refund_purchase_mock): refund_purchase_mock.return_value = { "result": "success", "data": {"status": "success", "refund_amount": 71.97}, } order = Order.objects.create( shopify_order_id="shop-2", digistore_order_id="DS2", status="paid", raw_payload={"amount_brutto": "71.97"}, ) payload = { "id": "r2", "order_id": order.shopify_order_id, "note": "customer request", "transactions": [{"kind": "refund", "amount": "71.97"}], } result = WebhookService.handle_refund_created(payload, webhook_id="wh-2") self.assertEqual(result, {"ok": True}) order.refresh_from_db() self.assertEqual(order.status, "refunded") event = ShopifyRefundEvent.objects.get(webhook_id="wh-2") self.assertTrue(event.processed) self.assertEqual(event.refund_type, "full") @patch("orders.services.order_service.OrderService.refund_purchase") def test_handle_refund_created_marks_failed_for_refused_response(self, refund_purchase_mock): refund_purchase_mock.return_value = { "result": "success", "data": { "purchase_id": "DS3", "details": [{"status": "refused", "code": "refund_right_expired"}], "refund_amount": 0, }, } order = Order.objects.create( shopify_order_id="shop-3", digistore_order_id="DS3", status="paid", raw_payload={"amount_brutto": "37.49"}, ) payload = { "id": "r3", "order_id": order.shopify_order_id, "transactions": [{"kind": "refund", "amount": "37.49"}], } result = WebhookService.handle_refund_created(payload, webhook_id="wh-3") self.assertFalse(result["ok"]) order.refresh_from_db() self.assertEqual(order.status, "paid") event = ShopifyRefundEvent.objects.get(webhook_id="wh-3") self.assertFalse(event.processed) self.assertIn("digistore_failed", event.error) @patch("orders.services.order_service.OrderService.refund_purchase") def test_handle_refund_created_retries_with_force_on_refund_window_expired( self, refund_purchase_mock, ): refund_purchase_mock.side_effect = [ { "result": "success", "data": { "purchase_id": "DS4", "details": [ { "status": "refused", "code": "refund_right_expired", "result_msg": "window expired", } ], "refund_amount": 0, }, }, { "result": "success", "data": {"status": "success", "refund_amount": 37.49}, }, ] order = Order.objects.create( shopify_order_id="shop-4", digistore_order_id="DS4", status="paid", raw_payload={"amount_brutto": "37.49"}, ) payload = { "id": "r4", "order_id": order.shopify_order_id, "transactions": [{"kind": "refund", "amount": "37.49"}], } result = WebhookService.handle_refund_created(payload, webhook_id="wh-4") self.assertEqual(result, {"ok": True}) self.assertEqual(refund_purchase_mock.call_count, 2) first_kwargs = refund_purchase_mock.call_args_list[0].kwargs second_kwargs = refund_purchase_mock.call_args_list[1].kwargs self.assertFalse(first_kwargs.get("force", False)) self.assertTrue(second_kwargs.get("force")) @patch("orders.services.order_service.OrderService.refund_partially") def test_partial_refund_retries_with_force_on_refund_window_expired( self, refund_partially_mock, ): refund_partially_mock.side_effect = [ { "result": "success", "data": { "purchase_id": "DSP", "details": [ { "status": "refused", "code": "refund_right_expired", "result_msg": "window expired", } ], "refund_amount": 0, }, }, { "result": "success", "data": {"status": "success", "refund_amount": 10.0}, }, ] order = Order.objects.create( shopify_order_id="shop-p", digistore_order_id="DSP", status="paid", raw_payload={"amount_brutto": "71.97"}, ) payload = { "id": "rp1", "order_id": order.shopify_order_id, "transactions": [{"kind": "refund", "amount": "10.00"}], } result = WebhookService.handle_refund_created(payload, webhook_id="wh-p") self.assertEqual(result, {"ok": True}) self.assertEqual(refund_partially_mock.call_count, 2) kwargs = refund_partially_mock.call_args_list[1].kwargs self.assertTrue(kwargs.get("force")) order.refresh_from_db() self.assertEqual(order.status, "partially refunded") def _fulfillment_payload( *, order_id: str = "shop-f1", fulfillment_id: str = "ful-1", tracking_number: str = "1Z999AA10123456784", tracking_company: str = "UPS", tracking_url: str = "https://www.ups.com/track?tracknum=1Z999AA10123456784", ) -> dict: payload = { "id": fulfillment_id, "order_id": order_id, "status": "success", "tracking_number": tracking_number, "tracking_company": tracking_company, "tracking_url": tracking_url, "tracking_numbers": [tracking_number] if tracking_number else [], "tracking_urls": [tracking_url] if tracking_url else [], } return payload class HandleFulfillmentCreatedOrUpdatedTests(TestCase): def test_handle_fulfillment_requires_mapping(self): payload = _fulfillment_payload(order_id="unknown-shop") result = WebhookService.handle_fulfillment_created_or_updated( payload, webhook_id="wh-f-1", topic="fulfillments/create", ) self.assertFalse(result["ok"]) self.assertEqual(result["error"], "order_mapping_not_found") event = ShopifyFulfillmentEvent.objects.get(webhook_id="wh-f-1") self.assertEqual(event.error, "order_mapping_not_found") self.assertFalse(event.processed) def test_handle_fulfillment_requires_tracking_number(self): order = Order.objects.create( shopify_order_id="shop-f-notrack", digistore_order_id="DSF0", status="paid", ) payload = _fulfillment_payload( order_id=order.shopify_order_id, tracking_number="", tracking_url="", ) payload["tracking_numbers"] = [] payload["tracking_urls"] = [] result = WebhookService.handle_fulfillment_created_or_updated( payload, webhook_id="wh-f-2", topic="fulfillments/create", ) self.assertFalse(result["ok"]) self.assertEqual(result["error"], "no_tracking_number") event = ShopifyFulfillmentEvent.objects.get(webhook_id="wh-f-2") self.assertEqual(event.error, "no_tracking_number") self.assertFalse(event.processed) @patch("orders.services.order_service.OrderService.sync_tracking_to_digistore") def test_handle_fulfillment_syncs_tracking_on_success(self, sync_mock): sync_mock.return_value = { "ok": True, "delivery_id": "17328355", "tracking": {"parcel_service": "ups"}, } order = Order.objects.create( shopify_order_id="shop-f-ok", digistore_order_id="DSFOK", status="paid", ) payload = _fulfillment_payload(order_id=order.shopify_order_id) result = WebhookService.handle_fulfillment_created_or_updated( payload, webhook_id="wh-f-3", topic="fulfillments/create", ) self.assertEqual(result, {"ok": True, "delivery_id": "17328355"}) sync_mock.assert_called_once() kwargs = sync_mock.call_args.kwargs self.assertEqual(kwargs["shopify_order_id"], order.shopify_order_id) self.assertEqual(kwargs["digistore_order_id"], order.digistore_order_id) self.assertEqual(kwargs["tracking_id"], "1Z999AA10123456784") self.assertEqual(kwargs["parcel_service"], "UPS") self.assertFalse(kwargs["notify_via_email"]) event = ShopifyFulfillmentEvent.objects.get(webhook_id="wh-f-3") self.assertTrue(event.processed) self.assertEqual(event.tracking_number, "1Z999AA10123456784") self.assertEqual(event.parcel_service, "ups") self.assertFalse(event.notify_via_email) self.assertEqual(event.error, "") @patch("orders.services.order_service.OrderService.sync_tracking_to_digistore") def test_handle_fulfillment_returns_duplicate_when_already_processed( self, sync_mock, ): ShopifyFulfillmentEvent.objects.create( webhook_id="wh-f-dup", topic="fulfillments/create", payload={"id": "ful-dup"}, processed=True, ) result = WebhookService.handle_fulfillment_created_or_updated( _fulfillment_payload(), webhook_id="wh-f-dup", topic="fulfillments/update", ) self.assertEqual( result, {"ok": True, "duplicate": True, "webhook_id": "wh-f-dup"}, ) sync_mock.assert_not_called() @override_settings(DIGISTORE_TRACKING_NOTIFY_VIA_EMAIL=True) @patch("orders.services.order_service.OrderService.sync_tracking_to_digistore") def test_handle_fulfillment_notify_flag_respects_setting(self, sync_mock): sync_mock.return_value = { "ok": True, "delivery_id": "99", "tracking": {"parcel_service": "dhl"}, } order = Order.objects.create( shopify_order_id="shop-f-notify", digistore_order_id="DSFN", status="paid", ) payload = _fulfillment_payload( order_id=order.shopify_order_id, tracking_company="DHL", ) result = WebhookService.handle_fulfillment_created_or_updated( payload, webhook_id="wh-f-notify", topic="fulfillments/update", ) self.assertTrue(result["ok"]) kwargs = sync_mock.call_args.kwargs self.assertTrue(kwargs["notify_via_email"]) event = ShopifyFulfillmentEvent.objects.get(webhook_id="wh-f-notify") self.assertTrue(event.notify_via_email) self.assertTrue(event.processed) @patch("orders.services.order_service.OrderService.sync_tracking_to_digistore") def test_handle_fulfillment_stores_error_when_sync_fails(self, sync_mock): sync_mock.return_value = {"ok": False, "error": "API error 3: boom"} order = Order.objects.create( shopify_order_id="shop-f-fail", digistore_order_id="DSFF", status="paid", ) result = WebhookService.handle_fulfillment_created_or_updated( _fulfillment_payload(order_id=order.shopify_order_id), webhook_id="wh-f-fail", topic="fulfillments/create", ) self.assertFalse(result["ok"]) event = ShopifyFulfillmentEvent.objects.get(webhook_id="wh-f-fail") self.assertFalse(event.processed) self.assertEqual(event.error, "API error 3: boom") @override_settings(DIGISTORE_TRACKING_SYNC_AFTER="2026-09-08") @patch("orders.services.order_service.OrderService.sync_tracking_to_digistore") def test_handle_fulfillment_skips_orders_created_before_cutoff(self, sync_mock): order = Order.objects.create( shopify_order_id="shop-f-old", digistore_order_id="YACENW9T", status="paid", ) Order.objects.filter(pk=order.pk).update( created_at=datetime(2026, 9, 5, 16, 40, tzinfo=timezone.utc) ) order.refresh_from_db() result = WebhookService.handle_fulfillment_created_or_updated( _fulfillment_payload(order_id=order.shopify_order_id), webhook_id="wh-f-old", topic="fulfillments/update", ) self.assertEqual( result, {"ok": True, "skipped": True, "reason": "historical_order"}, ) sync_mock.assert_not_called() self.assertFalse( ShopifyFulfillmentEvent.objects.filter(webhook_id="wh-f-old").exists() ) @override_settings(DIGISTORE_TRACKING_SYNC_AFTER="2026-09-08") @patch("orders.services.order_service.OrderService.sync_tracking_to_digistore") def test_handle_fulfillment_deletes_previously_saved_historical_event( self, sync_mock, ): order = Order.objects.create( shopify_order_id="shop-f-old2", digistore_order_id="OLD2", status="paid", ) Order.objects.filter(pk=order.pk).update( created_at=datetime(2026, 9, 5, 16, 40, tzinfo=timezone.utc) ) ShopifyFulfillmentEvent.objects.create( webhook_id="wh-f-old-saved", topic="fulfillments/update", shopify_order_id=order.shopify_order_id, digistore_order_id=order.digistore_order_id, payload={"id": "ful-old"}, processed=True, error="skipped_historical_order cutoff=2026-09-08", ) result = WebhookService.handle_fulfillment_created_or_updated( _fulfillment_payload(order_id=order.shopify_order_id), webhook_id="wh-f-old-saved", topic="fulfillments/update", ) self.assertEqual(result["reason"], "historical_order") sync_mock.assert_not_called() self.assertFalse( ShopifyFulfillmentEvent.objects.filter( webhook_id="wh-f-old-saved" ).exists() )