Template
westfarn
80a5f5dadf
Clients can run in-house retail without Shopify while keeping the same Docker/Django/Postgres stack and data-ownership promise. Closes #5.
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")
|
|
client = Client()
|
|
client.login(username="clerk", password="test-pass-123")
|
|
self.assertEqual(client.get(reverse("pos_sync:event_list")).status_code, 200)
|