118 lines
4.5 KiB
Python
118 lines
4.5 KiB
Python
import pytest
|
|
from fastapi import status
|
|
from uuid import uuid4
|
|
from datetime import datetime
|
|
|
|
|
|
class TestOfflineSync:
|
|
"""Test offline sync functionality."""
|
|
|
|
def test_sync_operations_with_uuid(self, test_client, test_db, user_token):
|
|
"""Test that offline operations with UUIDs are tracked."""
|
|
from backend.models import Item, User
|
|
|
|
item = Item(name="Test Item", category="Electronics", type="Component",
|
|
quantity=10, barcode="BC-UUID-1", part_number="PN-UUID")
|
|
test_db.add(item)
|
|
test_db.commit()
|
|
|
|
user = test_db.query(User).filter(User.username == "user").first()
|
|
operation_uuid = str(uuid4())
|
|
response = test_client.post(
|
|
"/sync/bulk-sync",
|
|
json={
|
|
"user_id": user.id,
|
|
"operations": [
|
|
{
|
|
"type": "CHECK_IN",
|
|
"barcode": "BC-UUID-1",
|
|
"quantity": 5,
|
|
"uuid": operation_uuid,
|
|
"timestamp": datetime.utcnow().isoformat()
|
|
}
|
|
]
|
|
},
|
|
headers={"Authorization": f"Bearer {user_token}"}
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
data = response.json()
|
|
assert "success" in data
|
|
assert len(data["success"]) == 1
|
|
|
|
def test_sync_duplicate_uuid_ignored(self, test_client, test_db, user_token):
|
|
"""Test that duplicate UUIDs don't create duplicate entries."""
|
|
from backend.models import Item, AuditLog, User
|
|
|
|
item = Item(name="Test Item", category="Electronics", type="Component",
|
|
quantity=10, barcode="BC-UUID-DUP", part_number="PN-DUP")
|
|
test_db.add(item)
|
|
test_db.commit()
|
|
|
|
user = test_db.query(User).filter(User.username == "user").first()
|
|
operation_uuid = str(uuid4())
|
|
payload = {
|
|
"user_id": user.id,
|
|
"operations": [
|
|
{
|
|
"type": "CHECK_IN",
|
|
"barcode": "BC-UUID-DUP",
|
|
"quantity": 5,
|
|
"uuid": operation_uuid,
|
|
"timestamp": datetime.utcnow().isoformat()
|
|
}
|
|
]
|
|
}
|
|
|
|
# First sync
|
|
response1 = test_client.post(
|
|
"/sync/bulk-sync", json=payload,
|
|
headers={"Authorization": f"Bearer {user_token}"}
|
|
)
|
|
assert response1.status_code == status.HTTP_200_OK
|
|
assert len(response1.json()["success"]) == 1
|
|
|
|
# Second sync (same UUID) — returns "Already synced" note
|
|
response2 = test_client.post(
|
|
"/sync/bulk-sync", json=payload,
|
|
headers={"Authorization": f"Bearer {user_token}"}
|
|
)
|
|
assert response2.status_code == status.HTTP_200_OK
|
|
assert response2.json()["success"][0].get("note") == "Already synced"
|
|
|
|
def test_sync_preserves_audit_trail(self, test_client, test_db, user_token):
|
|
"""Test that audit logs are preserved during sync."""
|
|
from backend.models import Item, AuditLog, User
|
|
|
|
item = Item(name="Test Item", category="Electronics", type="Component",
|
|
quantity=10, barcode="BC-AUDIT-TRAIL", part_number="PN-AT")
|
|
test_db.add(item)
|
|
test_db.commit()
|
|
|
|
user = test_db.query(User).filter(User.username == "user").first()
|
|
response = test_client.post(
|
|
"/sync/bulk-sync",
|
|
json={
|
|
"user_id": user.id,
|
|
"operations": [
|
|
{"type": "CHECK_IN", "barcode": "BC-AUDIT-TRAIL", "quantity": 5,
|
|
"uuid": str(uuid4()), "timestamp": datetime.utcnow().isoformat()},
|
|
{"type": "CHECK_IN", "barcode": "BC-AUDIT-TRAIL", "quantity": 3,
|
|
"uuid": str(uuid4()), "timestamp": datetime.utcnow().isoformat()},
|
|
{"type": "CHECK_OUT", "barcode": "BC-AUDIT-TRAIL", "quantity": 2,
|
|
"uuid": str(uuid4()), "timestamp": datetime.utcnow().isoformat()}
|
|
]
|
|
},
|
|
headers={"Authorization": f"Bearer {user_token}"}
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert len(response.json()["success"]) == 3
|
|
|
|
# Verify audit logs via API
|
|
logs_response = test_client.get(
|
|
"/operations/logs",
|
|
headers={"Authorization": f"Bearer {user_token}"}
|
|
)
|
|
assert logs_response.status_code == status.HTTP_200_OK
|
|
logs = logs_response.json()
|
|
assert len(logs) >= 3
|