305 lines
11 KiB
Python
305 lines
11 KiB
Python
"""
|
|
Tests for export service and export endpoints.
|
|
"""
|
|
|
|
import pytest
|
|
import io
|
|
import csv
|
|
from datetime import datetime
|
|
from unittest.mock import MagicMock, patch
|
|
from openpyxl import load_workbook
|
|
|
|
from backend.services.export_service import (
|
|
InventorySnapshotExporter,
|
|
AuditTrailExporter,
|
|
get_export_filename,
|
|
)
|
|
from backend.models import Item, AuditLog, User
|
|
|
|
|
|
class TestInventorySnapshotExporter:
|
|
"""Tests for inventory snapshot export functionality."""
|
|
|
|
@pytest.fixture
|
|
def sample_items(self):
|
|
"""Create sample items for testing."""
|
|
items = []
|
|
for i in range(3):
|
|
item = MagicMock(spec=Item)
|
|
item.id = i + 1
|
|
item.name = f"Item {i + 1}"
|
|
item.part_number = f"PN{i + 1}"
|
|
item.barcode = f"BC{i + 1}"
|
|
item.category = "Electronics"
|
|
item.type = "Component"
|
|
item.quantity = float(10 + i)
|
|
item.min_quantity = 1.0
|
|
item.description = f"Description {i + 1}"
|
|
item.color = "Blue"
|
|
item.size = "Medium"
|
|
item.connector = "USB"
|
|
item.box_label = f"BOX{i + 1}"
|
|
item.created_at = datetime(2026, 4, 22, 10, 0, 0)
|
|
item.updated_at = datetime(2026, 4, 22, 15, 0, 0)
|
|
items.append(item)
|
|
return items
|
|
|
|
def test_csv_export_basic(self, sample_items):
|
|
"""Test CSV export with sample items."""
|
|
csv_content = InventorySnapshotExporter.to_csv(sample_items, "2026-04-22")
|
|
|
|
# Verify CSV is not empty
|
|
assert csv_content
|
|
assert "Item 1" in csv_content
|
|
assert "PN1" in csv_content
|
|
|
|
# Parse CSV and verify structure
|
|
reader = csv.DictReader(io.StringIO(csv_content))
|
|
rows = list(reader)
|
|
assert len(rows) == 3
|
|
assert rows[0]["Name"] == "Item 1"
|
|
assert rows[0]["Part Number"] == "PN1"
|
|
|
|
def test_csv_export_headers(self, sample_items):
|
|
"""Test CSV export contains all expected headers."""
|
|
csv_content = InventorySnapshotExporter.to_csv(sample_items, "2026-04-22")
|
|
|
|
expected_headers = [
|
|
"ID", "Name", "Part Number", "Barcode", "Category",
|
|
"Type", "Quantity", "Min Quantity", "Description",
|
|
"Color", "Size", "Connector", "Box Label", "Created", "Modified"
|
|
]
|
|
|
|
reader = csv.DictReader(io.StringIO(csv_content))
|
|
assert list(reader.fieldnames) == expected_headers
|
|
|
|
def test_csv_export_empty(self):
|
|
"""Test CSV export with empty item list."""
|
|
csv_content = InventorySnapshotExporter.to_csv([], "2026-04-22")
|
|
|
|
# Should still have headers
|
|
assert "ID" in csv_content
|
|
assert "Name" in csv_content
|
|
|
|
def test_excel_export_basic(self, sample_items):
|
|
"""Test Excel export with sample items."""
|
|
excel_content = InventorySnapshotExporter.to_excel(sample_items, "2026-04-22")
|
|
|
|
# Verify it's bytes
|
|
assert isinstance(excel_content, bytes)
|
|
assert len(excel_content) > 0
|
|
|
|
# Load and verify Excel structure
|
|
wb = load_workbook(io.BytesIO(excel_content))
|
|
ws = wb.active
|
|
assert ws.title == "Inventory Snapshot"
|
|
assert ws["A1"].value == "Inventory Snapshot - 2026-04-22"
|
|
|
|
def test_excel_export_headers(self, sample_items):
|
|
"""Test Excel export contains all headers."""
|
|
excel_content = InventorySnapshotExporter.to_excel(sample_items, "2026-04-22")
|
|
wb = load_workbook(io.BytesIO(excel_content))
|
|
ws = wb.active
|
|
|
|
# Headers are in row 2 (row 1 is title)
|
|
headers = [cell.value for cell in ws[2]]
|
|
expected_headers = [
|
|
"ID", "Name", "Part Number", "Barcode", "Category",
|
|
"Type", "Quantity", "Min Quantity", "Description",
|
|
"Color", "Size", "Connector", "Box Label", "Created", "Modified"
|
|
]
|
|
assert headers == expected_headers
|
|
|
|
def test_excel_export_data(self, sample_items):
|
|
"""Test Excel export contains correct data."""
|
|
excel_content = InventorySnapshotExporter.to_excel(sample_items, "2026-04-22")
|
|
wb = load_workbook(io.BytesIO(excel_content))
|
|
ws = wb.active
|
|
|
|
# Data starts at row 3 (row 1 = title, row 2 = headers)
|
|
first_data_row = list(ws[3])
|
|
assert first_data_row[0].value == 1 # ID
|
|
assert first_data_row[1].value == "Item 1" # Name
|
|
|
|
def test_excel_export_empty(self):
|
|
"""Test Excel export with empty item list."""
|
|
excel_content = InventorySnapshotExporter.to_excel([], "2026-04-22")
|
|
wb = load_workbook(io.BytesIO(excel_content))
|
|
ws = wb.active
|
|
|
|
# Should have title and headers
|
|
assert ws["A1"].value == "Inventory Snapshot - 2026-04-22"
|
|
assert ws["A2"].value == "ID"
|
|
|
|
|
|
class TestAuditTrailExporter:
|
|
"""Tests for audit trail export functionality."""
|
|
|
|
@pytest.fixture
|
|
def sample_logs(self):
|
|
"""Create sample audit logs for testing."""
|
|
logs = []
|
|
for i in range(3):
|
|
user = MagicMock(spec=User)
|
|
user.username = f"user{i + 1}"
|
|
|
|
log = MagicMock(spec=AuditLog)
|
|
log.id = i + 1
|
|
log.timestamp = datetime(2026, 4, 22, 10 + i, 0, 0)
|
|
log.user = user
|
|
log.action = "CHECK_IN" if i % 2 == 0 else "CHECK_OUT"
|
|
log.target_item_id = i + 100
|
|
log.target_item_name = f"Item {i + 1}"
|
|
log.target_item_pn = f"PN{i + 1}"
|
|
log.target_item_barcode = f"BC{i + 1}"
|
|
log.quantity_change = float(i + 1)
|
|
log.details = f"Action details {i + 1}"
|
|
logs.append(log)
|
|
return logs
|
|
|
|
def test_csv_export_basic(self, sample_logs):
|
|
"""Test audit trail CSV export with sample logs."""
|
|
csv_content = AuditTrailExporter.to_csv(sample_logs, "2026-04-22")
|
|
|
|
# Verify CSV is not empty
|
|
assert csv_content
|
|
assert "CHECK_IN" in csv_content or "CHECK_OUT" in csv_content
|
|
|
|
# Parse CSV and verify structure
|
|
reader = csv.DictReader(io.StringIO(csv_content))
|
|
rows = list(reader)
|
|
assert len(rows) == 3
|
|
|
|
def test_csv_export_headers(self, sample_logs):
|
|
"""Test audit trail CSV export headers."""
|
|
csv_content = AuditTrailExporter.to_csv(sample_logs, "2026-04-22")
|
|
|
|
expected_headers = [
|
|
"ID", "Timestamp", "User", "Action", "Item ID",
|
|
"Item Name", "Item Part Number", "Item Barcode",
|
|
"Quantity Change", "Details"
|
|
]
|
|
|
|
reader = csv.DictReader(io.StringIO(csv_content))
|
|
assert list(reader.fieldnames) == expected_headers
|
|
|
|
def test_excel_export_basic(self, sample_logs):
|
|
"""Test audit trail Excel export."""
|
|
excel_content = AuditTrailExporter.to_excel(sample_logs, "2026-04-22")
|
|
|
|
assert isinstance(excel_content, bytes)
|
|
assert len(excel_content) > 0
|
|
|
|
wb = load_workbook(io.BytesIO(excel_content))
|
|
ws = wb.active
|
|
assert ws.title == "Audit Trail"
|
|
assert "Audit Trail - 2026-04-22" in ws["A1"].value
|
|
|
|
def test_excel_export_data(self, sample_logs):
|
|
"""Test audit trail Excel export contains correct data."""
|
|
excel_content = AuditTrailExporter.to_excel(sample_logs, "2026-04-22")
|
|
wb = load_workbook(io.BytesIO(excel_content))
|
|
ws = wb.active
|
|
|
|
# Data starts at row 3
|
|
first_data_row = list(ws[3])
|
|
assert first_data_row[0].value == 1 # ID
|
|
assert "user1" in str(first_data_row[2].value) or first_data_row[2].value == "user1" # User
|
|
|
|
|
|
class TestFilenameGeneration:
|
|
"""Tests for export filename generation."""
|
|
|
|
def test_csv_filename(self):
|
|
"""Test CSV filename generation."""
|
|
filename = get_export_filename("inventory_snapshot", "csv", "2026-04-22")
|
|
assert filename == "inventory_snapshot_2026-04-22.csv"
|
|
|
|
def test_xlsx_filename(self):
|
|
"""Test Excel filename generation."""
|
|
filename = get_export_filename("audit_trail", "xlsx", "2026-04-22")
|
|
assert filename == "audit_trail_2026-04-22.xlsx"
|
|
|
|
def test_filename_with_different_dates(self):
|
|
"""Test filename with different date formats."""
|
|
filename = get_export_filename("inventory_snapshot", "csv", "2026-01-15")
|
|
assert "2026-01-15" in filename
|
|
|
|
|
|
class TestExportEndpoints:
|
|
"""Tests for export API endpoints."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_inventory_snapshot_csv(self, client, admin_user, db):
|
|
"""Test inventory snapshot CSV export endpoint."""
|
|
response = client.post(
|
|
"/api/admin/exports/inventory-snapshot?format=csv",
|
|
headers={"Authorization": f"Bearer {admin_user.token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "text/csv" in response.headers.get("content-type", "")
|
|
assert "inventory_snapshot" in response.headers.get("content-disposition", "")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_inventory_snapshot_xlsx(self, client, admin_user, db):
|
|
"""Test inventory snapshot Excel export endpoint."""
|
|
response = client.post(
|
|
"/api/admin/exports/inventory-snapshot?format=xlsx",
|
|
headers={"Authorization": f"Bearer {admin_user.token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "spreadsheetml" in response.headers.get("content-type", "")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_audit_trail_csv(self, client, admin_user, db):
|
|
"""Test audit trail CSV export endpoint."""
|
|
response = client.post(
|
|
"/api/admin/exports/audit-trail?format=csv",
|
|
headers={"Authorization": f"Bearer {admin_user.token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "text/csv" in response.headers.get("content-type", "")
|
|
assert "audit_trail" in response.headers.get("content-disposition", "")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_audit_trail_xlsx(self, client, admin_user, db):
|
|
"""Test audit trail Excel export endpoint."""
|
|
response = client.post(
|
|
"/api/admin/exports/audit-trail?format=xlsx",
|
|
headers={"Authorization": f"Bearer {admin_user.token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "spreadsheetml" in response.headers.get("content-type", "")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_invalid_format(self, client, admin_user, db):
|
|
"""Test export with invalid format parameter."""
|
|
response = client.post(
|
|
"/api/admin/exports/inventory-snapshot?format=pdf",
|
|
headers={"Authorization": f"Bearer {admin_user.token}"}
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "Invalid format" in response.json().get("detail", "")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_unauthorized(self, client, db):
|
|
"""Test export endpoint without authorization."""
|
|
response = client.post("/api/admin/exports/inventory-snapshot?format=csv")
|
|
|
|
assert response.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_export_non_admin(self, client, regular_user, db):
|
|
"""Test export endpoint with non-admin user."""
|
|
response = client.post(
|
|
"/api/admin/exports/inventory-snapshot?format=csv",
|
|
headers={"Authorization": f"Bearer {regular_user.token}"}
|
|
)
|
|
|
|
assert response.status_code == 403
|