test: add offline sync and UUID idempotency tests
This commit is contained in:
110
backend/tests/test_offline_sync.py
Normal file
110
backend/tests/test_offline_sync.py
Normal file
@@ -0,0 +1,110 @@
|
||||
import pytest
|
||||
from fastapi import status
|
||||
from uuid import uuid4
|
||||
|
||||
|
||||
class TestOfflineSync:
|
||||
"""Test offline sync functionality."""
|
||||
|
||||
def test_sync_operations_with_uuid(self, test_client, test_db):
|
||||
"""Test that offline operations with UUIDs are tracked."""
|
||||
from backend.models import Item
|
||||
|
||||
item = Item(
|
||||
name="Test Item",
|
||||
category="Electronics",
|
||||
item_type="Component",
|
||||
quantity=10,
|
||||
barcode="123456789",
|
||||
part_number="PN-12345"
|
||||
)
|
||||
test_db.add(item)
|
||||
test_db.commit()
|
||||
|
||||
operation_uuid = str(uuid4())
|
||||
response = test_client.post(
|
||||
"/api/bulk-sync",
|
||||
json={
|
||||
"operations": [
|
||||
{
|
||||
"id": operation_uuid,
|
||||
"type": "CHECK_IN",
|
||||
"item_id": item.id,
|
||||
"quantity": 5,
|
||||
"timestamp": "2026-04-18T10:00:00Z"
|
||||
}
|
||||
]
|
||||
}
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["synced"] == 1
|
||||
|
||||
def test_sync_duplicate_uuid_ignored(self, test_client, test_db):
|
||||
"""Test that duplicate UUIDs don't create duplicate entries."""
|
||||
from backend.models import Item, AuditLog
|
||||
|
||||
item = Item(
|
||||
name="Test Item",
|
||||
category="Electronics",
|
||||
item_type="Component",
|
||||
quantity=10,
|
||||
barcode="123456789",
|
||||
part_number="PN-12345"
|
||||
)
|
||||
test_db.add(item)
|
||||
test_db.commit()
|
||||
|
||||
operation_uuid = str(uuid4())
|
||||
operation = {
|
||||
"id": operation_uuid,
|
||||
"type": "CHECK_IN",
|
||||
"item_id": item.id,
|
||||
"quantity": 5
|
||||
}
|
||||
|
||||
# First sync
|
||||
response1 = test_client.post("/api/bulk-sync", json={"operations": [operation]})
|
||||
assert response1.status_code == status.HTTP_200_OK
|
||||
|
||||
# Count logs
|
||||
logs_before = test_db.query(AuditLog).filter_by(item_id=item.id).count()
|
||||
|
||||
# Second sync (same UUID)
|
||||
response2 = test_client.post("/api/bulk-sync", json={"operations": [operation]})
|
||||
assert response2.status_code == status.HTTP_200_OK
|
||||
|
||||
# Logs count should be same (no duplicate)
|
||||
logs_after = test_db.query(AuditLog).filter_by(item_id=item.id).count()
|
||||
assert logs_before == logs_after
|
||||
|
||||
def test_sync_preserves_audit_trail(self, test_client, test_db):
|
||||
"""Test that audit logs are preserved during sync."""
|
||||
from backend.models import Item
|
||||
|
||||
item = Item(
|
||||
name="Test Item",
|
||||
category="Electronics",
|
||||
item_type="Component",
|
||||
quantity=10,
|
||||
barcode="123456789",
|
||||
part_number="PN-12345"
|
||||
)
|
||||
test_db.add(item)
|
||||
test_db.commit()
|
||||
|
||||
# Perform multiple operations
|
||||
operations = [
|
||||
{"id": str(uuid4()), "type": "CHECK_IN", "item_id": item.id, "quantity": 5},
|
||||
{"id": str(uuid4()), "type": "CHECK_IN", "item_id": item.id, "quantity": 3},
|
||||
{"id": str(uuid4()), "type": "CHECK_OUT", "item_id": item.id, "quantity": 2}
|
||||
]
|
||||
|
||||
response = test_client.post("/api/bulk-sync", json={"operations": operations})
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
|
||||
# Verify audit logs
|
||||
response = test_client.get(f"/api/audit-logs?item_id={item.id}")
|
||||
logs = response.json()
|
||||
assert len(logs) == 3
|
||||
assert logs[0]["operation"] == "CHECK_IN"
|
||||
assert logs[0]["quantity_change"] == 5
|
||||
Reference in New Issue
Block a user