1271 lines
38 KiB
Markdown
1271 lines
38 KiB
Markdown
# Phase 1: Backend Tests Implementation Plan
|
|
|
|
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
|
|
|
**Goal:** Build comprehensive Pytest test suite for backend (85%+ coverage) before any code refactoring begins.
|
|
|
|
**Architecture:** Create 7 test modules covering routers (users, items, operations, categories), models, auth, AI extraction, and offline sync. Use shared fixtures (conftest.py) for mocked auth, in-memory SQLite, and AI responses. Test types: unit (functions) + integration (full workflows).
|
|
|
|
**Tech Stack:** Pytest, Pytest-cov, SQLAlchemy (in-memory), Pydantic, Mocking (unittest.mock)
|
|
|
|
---
|
|
|
|
## File Structure
|
|
|
|
**Test files to create:**
|
|
- `backend/tests/conftest.py` — Shared fixtures (mocked auth, test DB, test client)
|
|
- `backend/tests/test_users.py` — Auth, user CRUD, LDAP simulation
|
|
- `backend/tests/test_items.py` — Item CRUD, barcode validation
|
|
- `backend/tests/test_operations.py` — Check-in/out workflows
|
|
- `backend/tests/test_categories.py` — Category CRUD
|
|
- `backend/tests/test_ai_extraction.py` — AI extraction pipeline (mocked)
|
|
- `backend/tests/test_offline_sync.py` — UUID idempotency, bulk sync
|
|
|
|
**Existing files to verify/update:**
|
|
- `backend/tests/test_admin.py` — Verify structure, expand if needed
|
|
- `backend/requirements.txt` — Verify pytest, pytest-cov, pytest-asyncio present
|
|
|
|
---
|
|
|
|
## Tasks
|
|
|
|
### Task 1: Setup pytest configuration and conftest.py
|
|
|
|
**Files:**
|
|
- Create: `backend/tests/conftest.py`
|
|
- Create: `backend/tests/pytest.ini` (optional)
|
|
- Modify: `backend/requirements.txt`
|
|
|
|
- [ ] **Step 1: Verify pytest packages in requirements.txt**
|
|
|
|
Run: `grep -E "pytest|pytest-cov|pytest-asyncio" backend/requirements.txt`
|
|
|
|
Expected output:
|
|
```
|
|
pytest==7.4.0
|
|
pytest-cov==4.1.0
|
|
pytest-asyncio==0.21.0
|
|
```
|
|
|
|
If missing, add them:
|
|
```bash
|
|
echo "pytest==7.4.0" >> backend/requirements.txt
|
|
echo "pytest-cov==4.1.0" >> backend/requirements.txt
|
|
echo "pytest-asyncio==0.21.0" >> backend/requirements.txt
|
|
```
|
|
|
|
- [ ] **Step 2: Create conftest.py with shared fixtures**
|
|
|
|
Create `backend/tests/conftest.py`:
|
|
|
|
```python
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from fastapi.testclient import TestClient
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from backend.main import app
|
|
from backend.models import Base
|
|
from backend.config_manager import ConfigManager
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def test_db():
|
|
"""Create in-memory SQLite database for tests."""
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
SessionLocal = sessionmaker(bind=engine)
|
|
session = SessionLocal()
|
|
yield session
|
|
session.close()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def test_client(test_db):
|
|
"""Create FastAPI test client with mocked database."""
|
|
|
|
def override_get_db():
|
|
return test_db
|
|
|
|
from backend.database import get_db
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
|
|
client = TestClient(app)
|
|
yield client
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_ldap():
|
|
"""Mock LDAP authentication."""
|
|
with patch("backend.auth.ldap.initialize") as mock_ldap_init:
|
|
mock_conn = MagicMock()
|
|
mock_ldap_init.return_value = mock_conn
|
|
mock_conn.simple_bind_s = MagicMock(return_value=True)
|
|
mock_conn.search_s = MagicMock(return_value=[
|
|
("uid=testuser,ou=people,dc=example,dc=com", {"uid": [b"testuser"]})
|
|
])
|
|
yield mock_ldap_init
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_gemini():
|
|
"""Mock Google Gemini AI extraction."""
|
|
with patch("backend.ai.gemini_extractor.generate_content") as mock_gen:
|
|
mock_gen.return_value = MagicMock(text='''{
|
|
"name": "Test Item",
|
|
"part_number": "PN-12345",
|
|
"category": "Electronics",
|
|
"quantity": 5
|
|
}''')
|
|
yield mock_gen
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_claude():
|
|
"""Mock Anthropic Claude AI extraction."""
|
|
with patch("backend.ai.claude_extractor.generate_content") as mock_gen:
|
|
mock_gen.return_value = MagicMock(content=[MagicMock(text='''{
|
|
"name": "Test Item",
|
|
"part_number": "PN-12345",
|
|
"category": "Electronics",
|
|
"quantity": 5
|
|
}''')])
|
|
yield mock_gen
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def mock_config():
|
|
"""Mock ConfigManager."""
|
|
with patch.object(ConfigManager, "get_config") as mock_get:
|
|
mock_get.return_value = {
|
|
"ai_provider": "gemini",
|
|
"api_key": "test-key"
|
|
}
|
|
yield mock_get
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def admin_token(test_client, test_db):
|
|
"""Create a test admin user and return auth token."""
|
|
from backend.models import User
|
|
|
|
admin = User(
|
|
username="admin",
|
|
email="admin@test.com",
|
|
is_admin=True,
|
|
password_hash="hashed_password"
|
|
)
|
|
test_db.add(admin)
|
|
test_db.commit()
|
|
|
|
# Return token (mocked)
|
|
return "test-admin-token"
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
def user_token(test_client, test_db):
|
|
"""Create a test regular user and return auth token."""
|
|
from backend.models import User
|
|
|
|
user = User(
|
|
username="user",
|
|
email="user@test.com",
|
|
is_admin=False,
|
|
password_hash="hashed_password"
|
|
)
|
|
test_db.add(user)
|
|
test_db.commit()
|
|
|
|
return "test-user-token"
|
|
```
|
|
|
|
- [ ] **Step 3: Run conftest.py syntax check**
|
|
|
|
Run: `python -m py_compile backend/tests/conftest.py`
|
|
|
|
Expected: No output (success)
|
|
|
|
- [ ] **Step 4: Commit**
|
|
|
|
```bash
|
|
git add backend/tests/conftest.py backend/requirements.txt
|
|
git commit -m "test: create pytest conftest with shared fixtures for backend tests"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 2: Create test_users.py (auth, CRUD, LDAP)
|
|
|
|
**Files:**
|
|
- Create: `backend/tests/test_users.py`
|
|
|
|
- [ ] **Step 1: Create test_users.py with auth tests**
|
|
|
|
Create `backend/tests/test_users.py`:
|
|
|
|
```python
|
|
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.return_value.simple_bind_s.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",
|
|
password_hash=hash_password("password123"),
|
|
is_admin=False
|
|
)
|
|
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",
|
|
"is_admin": False
|
|
},
|
|
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",
|
|
"is_admin": False
|
|
},
|
|
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",
|
|
is_admin=False,
|
|
password_hash="hashed"
|
|
)
|
|
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", password_hash="h", is_admin=False)
|
|
user2 = User(username="user2", email="user2@test.com", password_hash="h", is_admin=False)
|
|
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", password_hash="h", is_admin=False)
|
|
test_db.add(user)
|
|
test_db.commit()
|
|
|
|
response = test_client.put(
|
|
f"/api/users/{user.id}",
|
|
json={"email": "new@test.com", "is_admin": True},
|
|
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", password_hash="h", is_admin=False)
|
|
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
|
|
```
|
|
|
|
- [ ] **Step 2: Run test_users.py to verify structure (will fail, that's expected)**
|
|
|
|
Run: `pytest backend/tests/test_users.py -v`
|
|
|
|
Expected: Tests fail with "endpoint not found" or similar (fixtures work, but routes not mocked yet).
|
|
|
|
- [ ] **Step 3: Commit**
|
|
|
|
```bash
|
|
git add backend/tests/test_users.py
|
|
git commit -m "test: add user authentication and CRUD tests"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 3: Create test_items.py (item CRUD, validation)
|
|
|
|
**Files:**
|
|
- Create: `backend/tests/test_items.py`
|
|
|
|
- [ ] **Step 1: Create test_items.py**
|
|
|
|
Create `backend/tests/test_items.py`:
|
|
|
|
```python
|
|
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"} # Missing category, quantity, etc.
|
|
)
|
|
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
|
|
# Audit log should still exist (immutable)
|
|
|
|
|
|
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
|
|
```
|
|
|
|
- [ ] **Step 2: Run test_items.py**
|
|
|
|
Run: `pytest backend/tests/test_items.py -v`
|
|
|
|
Expected: Tests fail (endpoints not implemented yet, that's OK).
|
|
|
|
- [ ] **Step 3: Commit**
|
|
|
|
```bash
|
|
git add backend/tests/test_items.py
|
|
git commit -m "test: add item CRUD and validation tests"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 4: Create test_operations.py (check-in/out workflows)
|
|
|
|
**Files:**
|
|
- Create: `backend/tests/test_operations.py`
|
|
|
|
- [ ] **Step 1: Create test_operations.py**
|
|
|
|
Create `backend/tests/test_operations.py`:
|
|
|
|
```python
|
|
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
|
|
```
|
|
|
|
- [ ] **Step 2: Run test_operations.py**
|
|
|
|
Run: `pytest backend/tests/test_operations.py -v`
|
|
|
|
Expected: Tests fail (endpoints not yet mocked/implemented).
|
|
|
|
- [ ] **Step 3: Commit**
|
|
|
|
```bash
|
|
git add backend/tests/test_operations.py
|
|
git commit -m "test: add stock operations and offline sync tests"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 5: Create test_categories.py
|
|
|
|
**Files:**
|
|
- Create: `backend/tests/test_categories.py`
|
|
|
|
- [ ] **Step 1: Create test_categories.py**
|
|
|
|
Create `backend/tests/test_categories.py`:
|
|
|
|
```python
|
|
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
|
|
```
|
|
|
|
- [ ] **Step 2: Run test_categories.py**
|
|
|
|
Run: `pytest backend/tests/test_categories.py -v`
|
|
|
|
- [ ] **Step 3: Commit**
|
|
|
|
```bash
|
|
git add backend/tests/test_categories.py
|
|
git commit -m "test: add category CRUD tests"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 6: Create test_ai_extraction.py (mocked AI pipeline)
|
|
|
|
**Files:**
|
|
- Create: `backend/tests/test_ai_extraction.py`
|
|
|
|
- [ ] **Step 1: Create test_ai_extraction.py**
|
|
|
|
Create `backend/tests/test_ai_extraction.py`:
|
|
|
|
```python
|
|
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."""
|
|
# AI extraction should NOT save directly
|
|
# Instead, it should return for user confirmation
|
|
response = test_client.post(
|
|
"/api/ai/extract",
|
|
json={
|
|
"image_base64": "xyz",
|
|
"provider": "gemini",
|
|
"save": False # Extraction only, don't save
|
|
}
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
# Response contains extracted_data for user review
|
|
data = response.json()
|
|
assert "extracted_data" in data
|
|
assert "user_confirmation_required" in data or data.get("save") == False
|
|
```
|
|
|
|
- [ ] **Step 2: Run test_ai_extraction.py**
|
|
|
|
Run: `pytest backend/tests/test_ai_extraction.py -v`
|
|
|
|
- [ ] **Step 3: Commit**
|
|
|
|
```bash
|
|
git add backend/tests/test_ai_extraction.py
|
|
git commit -m "test: add AI extraction pipeline tests (mocked)"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 7: Create test_offline_sync.py
|
|
|
|
**Files:**
|
|
- Create: `backend/tests/test_offline_sync.py`
|
|
|
|
- [ ] **Step 1: Create test_offline_sync.py**
|
|
|
|
Create `backend/tests/test_offline_sync.py`:
|
|
|
|
```python
|
|
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
|
|
```
|
|
|
|
- [ ] **Step 2: Run test_offline_sync.py**
|
|
|
|
Run: `pytest backend/tests/test_offline_sync.py -v`
|
|
|
|
- [ ] **Step 3: Commit**
|
|
|
|
```bash
|
|
git add backend/tests/test_offline_sync.py
|
|
git commit -m "test: add offline sync and UUID idempotency tests"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 8: Run full pytest suite with coverage
|
|
|
|
**Files:**
|
|
- No new files (coverage analysis)
|
|
|
|
- [ ] **Step 1: Install dependencies**
|
|
|
|
Run:
|
|
```bash
|
|
cd backend
|
|
pip install -r requirements.txt
|
|
cd ..
|
|
```
|
|
|
|
Expected: All packages installed successfully (pytest, pytest-cov, etc.)
|
|
|
|
- [ ] **Step 2: Run pytest with coverage**
|
|
|
|
Run:
|
|
```bash
|
|
pytest backend/tests/ --cov=backend --cov-report=html --cov-report=term
|
|
```
|
|
|
|
Expected output (approximate):
|
|
```
|
|
====== test session starts ======
|
|
backend/tests/test_users.py .......... PASSED
|
|
backend/tests/test_items.py .......... PASSED
|
|
backend/tests/test_operations.py ..... PASSED
|
|
backend/tests/test_categories.py ..... PASSED
|
|
backend/tests/test_ai_extraction.py .. PASSED
|
|
backend/tests/test_offline_sync.py ... PASSED
|
|
|
|
------ Coverage report ------
|
|
Name Stmts Miss Cover
|
|
backend/routers/users.py 150 10 93%
|
|
backend/routers/items.py 120 15 87%
|
|
backend/routers/operations.py 95 8 91%
|
|
backend/routers/categories.py 80 5 93%
|
|
backend/models.py 200 10 95%
|
|
backend/auth.py 120 8 93%
|
|
backend/ai/gemini_extractor.py 60 5 91%
|
|
backend/ai/claude_extractor.py 55 4 92%
|
|
------ Coverage: 85%+ ------
|
|
```
|
|
|
|
- [ ] **Step 3: Verify coverage report**
|
|
|
|
Run: `ls -lh backend/coverage/`
|
|
|
|
Expected: HTML coverage report exists at `htmlcov/index.html`
|
|
|
|
- [ ] **Step 4: Check for test failures**
|
|
|
|
Run: `pytest backend/tests/ -v --tb=short | grep -E "FAILED|PASSED|ERROR"`
|
|
|
|
Expected: All tests PASS (or minimal failures due to unimplemented routes, which is OK for baseline)
|
|
|
|
- [ ] **Step 5: Commit test results**
|
|
|
|
```bash
|
|
git add backend/tests/ .coverage
|
|
git commit -m "test: phase 1 backend test suite complete - 85%+ coverage achieved"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 9: Create git tag for Phase 1 completion
|
|
|
|
**Files:**
|
|
- No new files (git tag)
|
|
|
|
- [ ] **Step 1: Create git tag**
|
|
|
|
Run: `git tag phase-1-complete`
|
|
|
|
Expected: Tag created
|
|
|
|
- [ ] **Step 2: Verify tag**
|
|
|
|
Run: `git tag -l | grep phase-1`
|
|
|
|
Expected output: `phase-1-complete`
|
|
|
|
- [ ] **Step 3: Update REFACTORING_PROGRESS.md**
|
|
|
|
Run:
|
|
```bash
|
|
# View current status
|
|
git log --oneline -5
|
|
git tag -l
|
|
```
|
|
|
|
Then update `REFACTORING_PROGRESS.md`:
|
|
- Change Phase 1 status to ✅ COMPLETE
|
|
- Update completion % to 100%
|
|
- Update last updated date
|
|
- Update commits count
|
|
|
|
- [ ] **Step 4: Final commit**
|
|
|
|
```bash
|
|
git add REFACTORING_PROGRESS.md
|
|
git commit -m "docs: mark phase 1 complete - backend tests suite ready for refactoring"
|
|
```
|
|
|
|
---
|
|
|
|
## Phase 1 Completion Checklist
|
|
|
|
- [ ] All 7 test files created (conftest, test_users, test_items, test_operations, test_categories, test_ai_extraction, test_offline_sync)
|
|
- [ ] `pytest backend/tests/ --cov=backend` shows **85%+ coverage**
|
|
- [ ] All tests passing (or acceptable failures for unimplemented routes)
|
|
- [ ] Coverage report generated (`htmlcov/index.html`)
|
|
- [ ] AGENTS.md updated with testing guidelines
|
|
- [ ] REFACTORING_PROGRESS.md created and updated
|
|
- [ ] Git tag `phase-1-complete` created
|
|
- [ ] All commits pushed to `refactor/ai-friendly` branch
|
|
|
|
---
|
|
|
|
## Execution Options
|
|
|
|
**Plan complete and saved to `docs/superpowers/plans/2026-04-18-phase-1-backend-tests.md`.**
|
|
|
|
Two execution options:
|
|
|
|
**1. Subagent-Driven (recommended)** — I dispatch a fresh subagent per task, review between tasks, fast iteration
|
|
|
|
**2. Inline Execution** — Execute tasks in this session using executing-plans, batch execution with checkpoints
|
|
|
|
Which approach?
|