generated from westfarn/web_django_template
westfarn
c9b81ceed7
CI / test (pull_request) Successful in 35s
Shoppers can register, save shipping details, and view order history while cards stay on Stripe. EasyPost tracker updates (including numbers from Pirate Ship) and 1–5 star reviews are limited to buyers. The contact form now only asks for email and a message.
116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
from decimal import Decimal
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import Client, TestCase, override_settings
|
|
from django.urls import reverse
|
|
|
|
from pos_sync.models import POSConnection, SyncEvent
|
|
from pos_sync.services import apply_inbound_sale, dispatch_pending_outbound, enqueue_online_sale
|
|
from shop.models import Product
|
|
from shop.services import add_to_cart, create_order_from_cart, mark_paid
|
|
|
|
|
|
class POSInboundTests(TestCase):
|
|
def setUp(self):
|
|
self.product = Product.objects.create(
|
|
name="Booster pack",
|
|
sku="TCG-PACK",
|
|
price=Decimal("4.99"),
|
|
stock_qty=10,
|
|
is_published=True,
|
|
)
|
|
|
|
def test_inbound_sale_decrements_stock(self):
|
|
event = apply_inbound_sale(sku="tcg-pack", quantity=3, external_id="pos-1")
|
|
self.product.refresh_from_db()
|
|
self.assertEqual(self.product.stock_qty, 7)
|
|
self.assertEqual(event.status, SyncEvent.Status.DONE)
|
|
|
|
@override_settings(POS_WEBHOOK_SECRET="pos-secret")
|
|
def test_webhook_requires_secret_and_updates_stock(self):
|
|
url = reverse("pos_sync:inventory_webhook")
|
|
anon = Client().post(
|
|
url,
|
|
data='{"sku":"TCG-PACK","quantity":1}',
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(anon.status_code, 401)
|
|
ok = Client().post(
|
|
url,
|
|
data='{"sku":"TCG-PACK","quantity":2,"external_id":"reg-9"}',
|
|
content_type="application/json",
|
|
HTTP_AUTHORIZATION="Bearer pos-secret",
|
|
)
|
|
self.assertEqual(ok.status_code, 200)
|
|
self.product.refresh_from_db()
|
|
self.assertEqual(self.product.stock_qty, 8)
|
|
|
|
|
|
class POSOutboundTests(TestCase):
|
|
def setUp(self):
|
|
self.product = Product.objects.create(
|
|
name="Figure",
|
|
sku="FIG-1",
|
|
price=Decimal("12.00"),
|
|
stock_qty=5,
|
|
is_published=True,
|
|
)
|
|
self.connection = POSConnection.objects.create(
|
|
name="Counter",
|
|
api_base_url="https://pos.example.com",
|
|
api_token="tok",
|
|
is_active=True,
|
|
)
|
|
|
|
def test_paid_order_enqueues_outbound(self):
|
|
session = self.client.session
|
|
add_to_cart(session, self.product, 1)
|
|
session.save()
|
|
order = create_order_from_cart(self.client.session, email="a@example.com")
|
|
mark_paid(order)
|
|
event = SyncEvent.objects.get(direction=SyncEvent.Direction.OUTBOUND)
|
|
self.assertEqual(event.sku, "FIG-1")
|
|
self.assertEqual(event.status, SyncEvent.Status.PENDING)
|
|
|
|
def test_dispatch_posts_reserve(self):
|
|
enqueue_online_sale(
|
|
type(
|
|
"O",
|
|
(),
|
|
{
|
|
"number": "ORD-1",
|
|
"pk": "x",
|
|
"items": type(
|
|
"M",
|
|
(),
|
|
{
|
|
"all": lambda self: [
|
|
type("I", (), {"sku": "FIG-1", "quantity": 1})()
|
|
]
|
|
},
|
|
)(),
|
|
},
|
|
)()
|
|
)
|
|
with patch("pos_sync.services.requests.post") as post:
|
|
post.return_value.raise_for_status = lambda: None
|
|
sent = dispatch_pending_outbound()
|
|
self.assertEqual(sent, 1)
|
|
post.assert_called_once()
|
|
self.assertIn("/inventory/reserve", post.call_args.args[0])
|
|
event = SyncEvent.objects.get()
|
|
self.assertEqual(event.status, SyncEvent.Status.DONE)
|
|
|
|
|
|
class POSPortalTests(TestCase):
|
|
def test_list_requires_login(self):
|
|
self.assertEqual(Client().get(reverse("pos_sync:event_list")).status_code, 302)
|
|
|
|
def test_list_ok_when_logged_in(self):
|
|
User = get_user_model()
|
|
User.objects.create_user("clerk", password="test-pass-123", is_staff=True)
|
|
client = Client()
|
|
client.login(username="clerk", password="test-pass-123")
|
|
self.assertEqual(client.get(reverse("pos_sync:event_list")).status_code, 200)
|