111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
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
|