Implementation Engineer
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
"""Tests for admin sync endpoints (T04)."""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
from app.services.sync_service import SyncService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_without_token(client):
|
|
"""POST /api/admin/sync without JWT returns 401."""
|
|
resp = await client.post("/api/admin/sync")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_status_without_token(client):
|
|
"""GET /api/admin/sync-status without JWT returns 401."""
|
|
resp = await client.get("/api/admin/sync-status")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_log_without_token(client):
|
|
"""GET /api/admin/sync-log without JWT returns 401."""
|
|
resp = await client.get("/api/admin/sync-log")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_with_valid_token(client, seeded_admin, test_db):
|
|
"""POST /api/admin/sync with valid JWT triggers equipment sync."""
|
|
# Login
|
|
login_resp = await client.post("/api/admin/login", json={
|
|
"username": "admin",
|
|
"password": "testpassword",
|
|
})
|
|
token = login_resp.json()["access_token"]
|
|
|
|
# Mock sync service
|
|
with patch.object(SyncService, "run_sync", new_callable=AsyncMock) as mock_sync:
|
|
mock_sync.return_value = {"sync_id": 42, "items_processed": 10, "items_failed": 0, "status": "completed"}
|
|
resp = await client.post("/api/admin/sync", cookies={"hms_admin_token": token})
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["sync_id"] == 42
|
|
assert data["status"] == "completed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_status_with_valid_token(client, seeded_admin):
|
|
"""GET /api/admin/sync-status with valid JWT returns status."""
|
|
login_resp = await client.post("/api/admin/login", json={
|
|
"username": "admin",
|
|
"password": "testpassword",
|
|
})
|
|
token = login_resp.json()["access_token"]
|
|
|
|
with patch.object(SyncService, "get_last_sync", new_callable=AsyncMock) as mock_status:
|
|
mock_status.return_value = {"last_sync": None, "items_processed": 0, "status": "never"}
|
|
resp = await client.get("/api/admin/sync-status", cookies={"hms_admin_token": token})
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["status"] == "never"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sync_log_with_valid_token(client, seeded_admin):
|
|
"""GET /api/admin/sync-log with valid JWT returns paginated log."""
|
|
login_resp = await client.post("/api/admin/login", json={
|
|
"username": "admin",
|
|
"password": "testpassword",
|
|
})
|
|
token = login_resp.json()["access_token"]
|
|
|
|
mock_log = MagicMock()
|
|
mock_log.id = 1
|
|
mock_log.sync_type = "equipment"
|
|
mock_log.status = "completed"
|
|
mock_log.items_processed = 100
|
|
mock_log.items_failed = 0
|
|
mock_log.error_message = None
|
|
from datetime import datetime
|
|
mock_log.started_at = datetime(2026, 7, 9, 12, 0, 0)
|
|
mock_log.completed_at = datetime(2026, 7, 9, 12, 5, 0)
|
|
|
|
with patch.object(SyncService, "get_sync_log_paginated", new_callable=AsyncMock) as mock_log_fn:
|
|
mock_log_fn.return_value = {
|
|
"items": [mock_log],
|
|
"total": 1,
|
|
"page": 1,
|
|
"page_size": 20,
|
|
"total_pages": 1,
|
|
}
|
|
resp = await client.get("/api/admin/sync-log", cookies={"hms_admin_token": token})
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["total"] == 1
|
|
assert len(data["items"]) == 1
|
|
assert data["items"][0]["status"] == "completed"
|