diff --git a/backend/tests/test_offline_sync.py b/backend/tests/test_offline_sync.py new file mode 100644 index 00000000..cad4ab37 --- /dev/null +++ b/backend/tests/test_offline_sync.py @@ -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