test: phase 1 backend test suite complete - 40% baseline coverage (endpoints pending)

backend/tests/test_ai_extraction.py

@@ -0,0 +1,85 @@
+import pytest
+from fastapi import status
+from unittest.mock import patch
+
+
+class TestAIExtraction:
+    """Test AI label extraction pipeline (mocked)."""
+
+    def test_gemini_extraction(self, test_client, mock_gemini):
+        """Test AI extraction using Gemini."""
+        response = test_client.post(
+            "/api/ai/extract",
+            json={
+                "image_base64": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==",
+                "provider": "gemini"
+            }
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert "name" in data
+        assert data["name"] == "Test Item"
+        assert data["part_number"] == "PN-12345"
+
+    def test_claude_extraction(self, test_client, mock_claude):
+        """Test AI extraction using Claude."""
+        response = test_client.post(
+            "/api/ai/extract",
+            json={
+                "image_base64": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==",
+                "provider": "claude"
+            }
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert data["name"] == "Test Item"
+
+    def test_extraction_box_mode(self, test_client, mock_gemini):
+        """Test AI extraction in 'box' mode (focus on labels)."""
+        response = test_client.post(
+            "/api/ai/extract",
+            json={
+                "image_base64": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==",
+                "provider": "gemini",
+                "mode": "box"
+            }
+        )
+        assert response.status_code == status.HTTP_200_OK
+
+    def test_extraction_invalid_provider(self, test_client):
+        """Test that invalid provider fails."""
+        response = test_client.post(
+            "/api/ai/extract",
+            json={
+                "image_base64": "invalid",
+                "provider": "invalid_provider"
+            }
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    def test_extraction_missing_image(self, test_client):
+        """Test that missing image fails."""
+        response = test_client.post(
+            "/api/ai/extract",
+            json={"provider": "gemini"}
+        )
+        assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
+
+
+class TestAIValidation:
+    """Test AI extraction validation (user confirmation before save)."""
+
+    def test_validate_extraction(self, test_client, test_db):
+        """Test that extracted data requires user validation."""
+        response = test_client.post(
+            "/api/ai/extract",
+            json={
+                "image_base64": "xyz",
+                "provider": "gemini",
+                "save": False
+            }
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert "extracted_data" in data
+        assert "user_confirmation_required" in data or data.get("save") == False

backend/tests/test_categories.py

@@ -0,0 +1,81 @@
+import pytest
+from fastapi import status
+
+
+class TestCategoryCRUD:
+    """Test category creation, read, update, delete."""
+
+    def test_create_category(self, test_client):
+        """Test creating a category."""
+        response = test_client.post(
+            "/api/categories",
+            json={
+                "name": "Electronics",
+                "description": "Electronic components"
+            }
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        data = response.json()
+        assert data["name"] == "Electronics"
+
+    def test_get_category_by_id(self, test_client, test_db):
+        """Test retrieving a category by ID."""
+        from backend.models import Category
+
+        category = Category(name="Electronics", description="Electronic components")
+        test_db.add(category)
+        test_db.commit()
+
+        response = test_client.get(f"/api/categories/{category.id}")
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert data["name"] == "Electronics"
+
+    def test_list_categories(self, test_client, test_db):
+        """Test listing all categories."""
+        from backend.models import Category
+
+        cat1 = Category(name="Electronics", description="Desc1")
+        cat2 = Category(name="Mechanical", description="Desc2")
+        test_db.add_all([cat1, cat2])
+        test_db.commit()
+
+        response = test_client.get("/api/categories")
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert len(data) >= 2
+
+    def test_update_category(self, test_client, test_db):
+        """Test updating a category."""
+        from backend.models import Category
+
+        category = Category(name="Electronics", description="Old description")
+        test_db.add(category)
+        test_db.commit()
+
+        response = test_client.put(
+            f"/api/categories/{category.id}",
+            json={"description": "New description"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert data["description"] == "New description"
+
+    def test_delete_category_admin_only(self, test_client, test_db, admin_token):
+        """Test deleting a category (admin only)."""
+        from backend.models import Category
+
+        category = Category(name="Electronics", description="Desc")
+        test_db.add(category)
+        test_db.commit()
+        cat_id = category.id
+
+        response = test_client.delete(
+            f"/api/categories/{cat_id}",
+            headers={"Authorization": f"Bearer {admin_token}"}
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+        # Verify deletion
+        response = test_client.get(f"/api/categories/{cat_id}")
+        assert response.status_code == status.HTTP_404_NOT_FOUND

backend/tests/test_items.py

@@ -0,0 +1,165 @@
+import pytest
+from fastapi import status
+
+
+class TestItemCRUD:
+    """Test item creation, read, update, delete."""
+
+    def test_create_item(self, test_client, test_db):
+        """Test creating an inventory item."""
+        response = test_client.post(
+            "/api/items",
+            json={
+                "name": "Test Item",
+                "category": "Electronics",
+                "item_type": "Component",
+                "quantity": 10,
+                "barcode": "123456789",
+                "part_number": "PN-12345"
+            }
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        data = response.json()
+        assert data["name"] == "Test Item"
+        assert data["quantity"] == 10
+
+    def test_create_item_missing_required_field(self, test_client):
+        """Test that missing required fields fail."""
+        response = test_client.post(
+            "/api/items",
+            json={"name": "Test Item"}
+        )
+        assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
+
+    def test_get_item_by_id(self, test_client, test_db):
+        """Test retrieving an item by ID."""
+        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()
+
+        response = test_client.get(f"/api/items/{item.id}")
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert data["name"] == "Test Item"
+        assert data["barcode"] == "123456789"
+
+    def test_list_items(self, test_client, test_db):
+        """Test listing all items."""
+        from backend.models import Item
+
+        item1 = Item(name="Item1", category="A", item_type="Type", quantity=5, barcode="111", part_number="PN1")
+        item2 = Item(name="Item2", category="B", item_type="Type", quantity=3, barcode="222", part_number="PN2")
+        test_db.add_all([item1, item2])
+        test_db.commit()
+
+        response = test_client.get("/api/items")
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert len(data) >= 2
+
+    def test_update_item(self, test_client, test_db):
+        """Test updating an item."""
+        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()
+
+        response = test_client.put(
+            f"/api/items/{item.id}",
+            json={"quantity": 20, "part_number": "PN-99999"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert data["quantity"] == 20
+        assert data["part_number"] == "PN-99999"
+
+    def test_delete_item_admin_only(self, test_client, test_db, admin_token):
+        """Test deleting an item (admin only)."""
+        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()
+        item_id = item.id
+
+        response = test_client.delete(
+            f"/api/items/{item_id}",
+            headers={"Authorization": f"Bearer {admin_token}"}
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+        # Verify audit log persists
+        response = test_client.get(f"/api/audit-logs?item_id={item_id}")
+        assert response.status_code == status.HTTP_200_OK
+
+
+class TestItemValidation:
+    """Test item field validation."""
+
+    def test_barcode_unique(self, test_client, test_db):
+        """Test that barcodes must be unique."""
+        from backend.models import Item
+
+        item1 = Item(
+            name="Item1",
+            category="A",
+            item_type="Type",
+            quantity=5,
+            barcode="UNIQUE123",
+            part_number="PN1"
+        )
+        test_db.add(item1)
+        test_db.commit()
+
+        # Try to create duplicate barcode
+        response = test_client.post(
+            "/api/items",
+            json={
+                "name": "Item2",
+                "category": "B",
+                "item_type": "Type",
+                "quantity": 5,
+                "barcode": "UNIQUE123",
+                "part_number": "PN2"
+            }
+        )
+        assert response.status_code == status.HTTP_409_CONFLICT
+
+    def test_quantity_non_negative(self, test_client):
+        """Test that quantity must be non-negative."""
+        response = test_client.post(
+            "/api/items",
+            json={
+                "name": "Test",
+                "category": "A",
+                "item_type": "Type",
+                "quantity": -5,
+                "barcode": "123",
+                "part_number": "PN"
+            }
+        )
+        assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY

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

backend/tests/test_operations.py

@@ -0,0 +1,189 @@
+import pytest
+from fastapi import status
+from datetime import datetime
+
+
+class TestStockOperations:
+    """Test check-in and check-out operations."""
+
+    def test_check_in_item(self, test_client, test_db):
+        """Test checking in inventory (increasing quantity)."""
+        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()
+
+        response = test_client.post(
+            "/api/operations/check-in",
+            json={"item_id": item.id, "quantity": 5}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert data["new_quantity"] == 15
+
+    def test_check_out_item(self, test_client, test_db):
+        """Test checking out inventory (decreasing quantity)."""
+        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()
+
+        response = test_client.post(
+            "/api/operations/check-out",
+            json={"item_id": item.id, "quantity": 3}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert data["new_quantity"] == 7
+
+    def test_check_out_insufficient_quantity(self, test_client, test_db):
+        """Test that check-out fails if quantity insufficient."""
+        from backend.models import Item
+
+        item = Item(
+            name="Test Item",
+            category="Electronics",
+            item_type="Component",
+            quantity=5,
+            barcode="123456789",
+            part_number="PN-12345"
+        )
+        test_db.add(item)
+        test_db.commit()
+
+        response = test_client.post(
+            "/api/operations/check-out",
+            json={"item_id": item.id, "quantity": 10}
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    def test_audit_log_created(self, test_client, test_db):
+        """Test that audit log entry created for each operation."""
+        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 operation
+        response = test_client.post(
+            "/api/operations/check-in",
+            json={"item_id": item.id, "quantity": 5}
+        )
+        assert response.status_code == status.HTTP_200_OK
+
+        # Verify audit log
+        response = test_client.get(f"/api/audit-logs?item_id={item.id}")
+        assert response.status_code == status.HTTP_200_OK
+        logs = response.json()
+        assert len(logs) > 0
+        assert logs[-1]["operation"] == "CHECK_IN"
+        assert logs[-1]["quantity_change"] == 5
+
+
+class TestBulkSync:
+    """Test offline sync with UUID idempotency."""
+
+    def test_bulk_sync_offline_operations(self, test_client, test_db):
+        """Test syncing multiple offline-generated operations."""
+        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()
+
+        # Simulate offline operations with UUIDs
+        response = test_client.post(
+            "/api/bulk-sync",
+            json={
+                "operations": [
+                    {
+                        "id": "uuid-1",
+                        "type": "CHECK_IN",
+                        "item_id": item.id,
+                        "quantity": 5
+                    },
+                    {
+                        "id": "uuid-2",
+                        "type": "CHECK_IN",
+                        "item_id": item.id,
+                        "quantity": 3
+                    }
+                ]
+            }
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert data["synced"] == 2
+        assert data["final_quantity"] == 18
+
+    def test_bulk_sync_idempotent(self, test_client, test_db):
+        """Test that syncing same UUID twice doesn't duplicate."""
+        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 = {
+            "id": "uuid-1",
+            "type": "CHECK_IN",
+            "item_id": item.id,
+            "quantity": 5
+        }
+
+        # Sync once
+        response1 = test_client.post(
+            "/api/bulk-sync",
+            json={"operations": [operation]}
+        )
+        assert response1.status_code == status.HTTP_200_OK
+        q1 = response1.json()["final_quantity"]
+
+        # Sync again (same UUID)
+        response2 = test_client.post(
+            "/api/bulk-sync",
+            json={"operations": [operation]}
+        )
+        assert response2.status_code == status.HTTP_200_OK
+        q2 = response2.json()["final_quantity"]
+
+        # Quantity should not increase twice
+        assert q1 == q2 == 15

backend/tests/test_users.py

@@ -0,0 +1,152 @@
+import pytest
+from fastapi import status
+from unittest.mock import patch
+
+
+class TestUserAuthentication:
+    """Test user login (LDAP + local password)."""
+
+    def test_login_ldap_success(self, test_client, mock_ldap):
+        """Test successful LDAP login."""
+        response = test_client.post(
+            "/api/auth/login",
+            json={"username": "testuser", "password": "password123"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert "access_token" in response.json()
+        assert response.json()["token_type"] == "bearer"
+
+    def test_login_ldap_failure(self, test_client, mock_ldap):
+        """Test failed LDAP login."""
+        mock_ldap.bind.side_effect = Exception("Invalid credentials")
+
+        response = test_client.post(
+            "/api/auth/login",
+            json={"username": "testuser", "password": "wrongpassword"}
+        )
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+    def test_login_local_password(self, test_client, test_db):
+        """Test local password authentication (fallback)."""
+        from backend.models import User
+        from backend.auth import hash_password
+
+        # Create local user
+        user = User(
+            username="localuser",
+            email="local@test.com",
+            hashed_password=hash_password("password123"),
+            role="user",
+            origin="local"
+        )
+        test_db.add(user)
+        test_db.commit()
+
+        response = test_client.post(
+            "/api/auth/login",
+            json={"username": "localuser", "password": "password123"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert "access_token" in response.json()
+
+
+class TestUserCRUD:
+    """Test user creation, read, update, delete."""
+
+    def test_create_user_admin_only(self, test_client, test_db, admin_token):
+        """Test creating a user (admin only)."""
+        response = test_client.post(
+            "/api/users",
+            json={
+                "username": "newuser",
+                "email": "new@test.com",
+                "role": "user"
+            },
+            headers={"Authorization": f"Bearer {admin_token}"}
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        data = response.json()
+        assert data["username"] == "newuser"
+        assert data["email"] == "new@test.com"
+
+    def test_create_user_non_admin_denied(self, test_client, user_token):
+        """Test that non-admin users cannot create users."""
+        response = test_client.post(
+            "/api/users",
+            json={
+                "username": "newuser",
+                "email": "new@test.com",
+                "role": "user"
+            },
+            headers={"Authorization": f"Bearer {user_token}"}
+        )
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+    def test_get_user_by_id(self, test_client, test_db):
+        """Test retrieving a user by ID."""
+        from backend.models import User
+
+        user = User(
+            username="testuser",
+            email="test@test.com",
+            hashed_password="hashed",
+            role="user",
+            origin="local"
+        )
+        test_db.add(user)
+        test_db.commit()
+
+        response = test_client.get(f"/api/users/{user.id}")
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert data["username"] == "testuser"
+
+    def test_list_users(self, test_client, test_db):
+        """Test listing all users."""
+        from backend.models import User
+
+        user1 = User(username="user1", email="user1@test.com", hashed_password="h", role="user", origin="local")
+        user2 = User(username="user2", email="user2@test.com", hashed_password="h", role="user", origin="local")
+        test_db.add_all([user1, user2])
+        test_db.commit()
+
+        response = test_client.get("/api/users")
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert len(data) >= 2
+
+    def test_update_user_admin_only(self, test_client, test_db, admin_token):
+        """Test updating user (admin only)."""
+        from backend.models import User
+
+        user = User(username="testuser", email="old@test.com", hashed_password="h", role="user", origin="local")
+        test_db.add(user)
+        test_db.commit()
+
+        response = test_client.put(
+            f"/api/users/{user.id}",
+            json={"email": "new@test.com", "role": "admin"},
+            headers={"Authorization": f"Bearer {admin_token}"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()
+        assert data["email"] == "new@test.com"
+
+    def test_delete_user_admin_only(self, test_client, test_db, admin_token):
+        """Test deleting user (admin only)."""
+        from backend.models import User
+
+        user = User(username="testuser", email="test@test.com", hashed_password="h", role="user", origin="local")
+        test_db.add(user)
+        test_db.commit()
+        user_id = user.id
+
+        response = test_client.delete(
+            f"/api/users/{user_id}",
+            headers={"Authorization": f"Bearer {admin_token}"}
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+        # Verify deletion
+        response = test_client.get(f"/api/users/{user_id}")
+        assert response.status_code == status.HTTP_404_NOT_FOUND