sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add admin inventory management (Phase 1)

Browse Source 
- Add admin API endpoints for inventory management
- Add inventory page with vendor selector and filtering
- Add admin schemas for cross-vendor inventory operations
- Support digital products with unlimited inventory
- Add integration tests for admin inventory API
- Add inventory management guide documentation

Mirrors vendor inventory functionality with admin-level access.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2025-12-18 21:05:12 +01:00
parent 0ab10128ae
commit 8d8d41808b
12 changed files with 2880 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

497
tests/integration/api/v1/admin/test_inventory.py Normal file
View File

@@ -0,0 +1,497 @@
# tests/integration/api/v1/admin/test_inventory.py
"""
Integration tests for admin inventory management endpoints.
Tests the /api/v1/admin/inventory/* endpoints.
All endpoints require admin JWT authentication.
"""
import pytest
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
@pytest.mark.inventory
class TestAdminInventoryAPI:
"""Tests for admin inventory management endpoints."""
# ========================================================================
# List & Statistics Tests
# ========================================================================
def test_get_all_inventory_admin(
self, client, admin_headers, test_inventory, test_vendor
):
"""Test admin getting all inventory."""
response = client.get("/api/v1/admin/inventory", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert "inventories" in data
assert "total" in data
assert "skip" in data
assert "limit" in data
assert data["total"] >= 1
assert len(data["inventories"]) >= 1
# Check that test inventory is in the response
inventory_ids = [i["id"] for i in data["inventories"]]
assert test_inventory.id in inventory_ids
def test_get_all_inventory_non_admin(self, client, auth_headers):
"""Test non-admin trying to access inventory endpoint."""
response = client.get("/api/v1/admin/inventory", headers=auth_headers)
assert response.status_code == 403
data = response.json()
assert data["error_code"] == "ADMIN_REQUIRED"
def test_get_all_inventory_with_vendor_filter(
self, client, admin_headers, test_inventory, test_vendor
):
"""Test admin filtering inventory by vendor."""
response = client.get(
"/api/v1/admin/inventory",
params={"vendor_id": test_vendor.id},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert data["vendor_filter"] == test_vendor.id
# All inventory should be from the filtered vendor
for item in data["inventories"]:
assert item["vendor_id"] == test_vendor.id
def test_get_all_inventory_with_location_filter(
self, client, admin_headers, test_inventory
):
"""Test admin filtering inventory by location."""
location = test_inventory.location[:5] # Partial match
response = client.get(
"/api/v1/admin/inventory",
params={"location": location},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
# All items should have matching location
for item in data["inventories"]:
assert location.upper() in item["location"].upper()
def test_get_all_inventory_with_low_stock_filter(
self, client, admin_headers, test_inventory, db
):
"""Test admin filtering inventory by low stock threshold."""
# Set test inventory to low stock
test_inventory.quantity = 5
db.commit()
response = client.get(
"/api/v1/admin/inventory",
params={"low_stock": 10},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
# All items should have quantity <= threshold
for item in data["inventories"]:
assert item["quantity"] <= 10
def test_get_all_inventory_pagination(
self, client, admin_headers, test_inventory
):
"""Test admin inventory pagination."""
response = client.get(
"/api/v1/admin/inventory",
params={"skip": 0, "limit": 10},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["skip"] == 0
assert data["limit"] == 10
def test_get_inventory_stats_admin(
self, client, admin_headers, test_inventory
):
"""Test admin getting inventory statistics."""
response = client.get(
"/api/v1/admin/inventory/stats", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert "total_entries" in data
assert "total_quantity" in data
assert "total_reserved" in data
assert "total_available" in data
assert "low_stock_count" in data
assert "vendors_with_inventory" in data
assert "unique_locations" in data
assert data["total_entries"] >= 1
def test_get_inventory_stats_non_admin(self, client, auth_headers):
"""Test non-admin trying to access inventory stats."""
response = client.get(
"/api/v1/admin/inventory/stats", headers=auth_headers
)
assert response.status_code == 403
def test_get_low_stock_items_admin(
self, client, admin_headers, test_inventory, db
):
"""Test admin getting low stock items."""
# Set test inventory to low stock
test_inventory.quantity = 3
db.commit()
response = client.get(
"/api/v1/admin/inventory/low-stock",
params={"threshold": 10},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
# All items should have quantity <= threshold
for item in data:
assert item["quantity"] <= 10
def test_get_vendors_with_inventory_admin(
self, client, admin_headers, test_inventory, test_vendor
):
"""Test admin getting vendors with inventory."""
response = client.get(
"/api/v1/admin/inventory/vendors", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert "vendors" in data
assert isinstance(data["vendors"], list)
assert len(data["vendors"]) >= 1
# Check that test_vendor is in the list
vendor_ids = [v["id"] for v in data["vendors"]]
assert test_vendor.id in vendor_ids
def test_get_inventory_locations_admin(
self, client, admin_headers, test_inventory
):
"""Test admin getting inventory locations."""
response = client.get(
"/api/v1/admin/inventory/locations", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert "locations" in data
assert isinstance(data["locations"], list)
assert len(data["locations"]) >= 1
assert test_inventory.location in data["locations"]
# ========================================================================
# Vendor-Specific Tests
# ========================================================================
def test_get_vendor_inventory_admin(
self, client, admin_headers, test_inventory, test_vendor
):
"""Test admin getting vendor-specific inventory."""
response = client.get(
f"/api/v1/admin/inventory/vendors/{test_vendor.id}",
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert "inventories" in data
assert "total" in data
assert data["vendor_filter"] == test_vendor.id
assert data["total"] >= 1
# All inventory should be from this vendor
for item in data["inventories"]:
assert item["vendor_id"] == test_vendor.id
def test_get_vendor_inventory_not_found(self, client, admin_headers):
"""Test admin getting inventory for non-existent vendor."""
response = client.get(
"/api/v1/admin/inventory/vendors/99999",
headers=admin_headers,
)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "VENDOR_NOT_FOUND"
def test_get_product_inventory_admin(
self, client, admin_headers, test_inventory, test_product
):
"""Test admin getting product inventory summary."""
response = client.get(
f"/api/v1/admin/inventory/products/{test_product.id}",
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["product_id"] == test_product.id
assert "total_quantity" in data
assert "total_reserved" in data
assert "total_available" in data
assert "locations" in data
def test_get_product_inventory_not_found(self, client, admin_headers):
"""Test admin getting inventory for non-existent product."""
response = client.get(
"/api/v1/admin/inventory/products/99999",
headers=admin_headers,
)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "PRODUCT_NOT_FOUND"
# ========================================================================
# Inventory Modification Tests
# ========================================================================
def test_set_inventory_admin(
self, client, admin_headers, test_product, test_vendor
):
"""Test admin setting inventory for a product."""
inventory_data = {
"vendor_id": test_vendor.id,
"product_id": test_product.id,
"location": "ADMIN_TEST_WAREHOUSE",
"quantity": 150,
}
response = client.post(
"/api/v1/admin/inventory/set",
headers=admin_headers,
json=inventory_data,
)
assert response.status_code == 200, f"Failed: {response.json()}"
data = response.json()
assert data["product_id"] == test_product.id
assert data["vendor_id"] == test_vendor.id
assert data["quantity"] == 150
assert data["location"] == "ADMIN_TEST_WAREHOUSE"
def test_set_inventory_non_admin(
self, client, auth_headers, test_product, test_vendor
):
"""Test non-admin trying to set inventory."""
inventory_data = {
"vendor_id": test_vendor.id,
"product_id": test_product.id,
"location": "WAREHOUSE_A",
"quantity": 100,
}
response = client.post(
"/api/v1/admin/inventory/set",
headers=auth_headers,
json=inventory_data,
)
assert response.status_code == 403
def test_set_inventory_vendor_not_found(
self, client, admin_headers, test_product
):
"""Test admin setting inventory for non-existent vendor."""
inventory_data = {
"vendor_id": 99999,
"product_id": test_product.id,
"location": "WAREHOUSE_A",
"quantity": 100,
}
response = client.post(
"/api/v1/admin/inventory/set",
headers=admin_headers,
json=inventory_data,
)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "VENDOR_NOT_FOUND"
def test_adjust_inventory_add_admin(
self, client, admin_headers, test_inventory, test_vendor, test_product
):
"""Test admin adding to inventory."""
original_qty = test_inventory.quantity
adjust_data = {
"vendor_id": test_vendor.id,
"product_id": test_product.id,
"location": test_inventory.location,
"quantity": 25,
"reason": "Admin restocking",
}
response = client.post(
"/api/v1/admin/inventory/adjust",
headers=admin_headers,
json=adjust_data,
)
assert response.status_code == 200
data = response.json()
assert data["quantity"] == original_qty + 25
def test_adjust_inventory_remove_admin(
self, client, admin_headers, test_inventory, test_vendor, test_product, db
):
"""Test admin removing from inventory."""
# Ensure we have enough inventory
test_inventory.quantity = 100
db.commit()
adjust_data = {
"vendor_id": test_vendor.id,
"product_id": test_product.id,
"location": test_inventory.location,
"quantity": -10,
"reason": "Admin adjustment",
}
response = client.post(
"/api/v1/admin/inventory/adjust",
headers=admin_headers,
json=adjust_data,
)
assert response.status_code == 200
data = response.json()
assert data["quantity"] == 90
def test_adjust_inventory_insufficient(
self, client, admin_headers, test_inventory, test_vendor, test_product, db
):
"""Test admin trying to remove more than available."""
# Set low inventory
test_inventory.quantity = 5
db.commit()
adjust_data = {
"vendor_id": test_vendor.id,
"product_id": test_product.id,
"location": test_inventory.location,
"quantity": -100,
}
response = client.post(
"/api/v1/admin/inventory/adjust",
headers=admin_headers,
json=adjust_data,
)
# Service wraps InsufficientInventoryException in ValidationException
assert response.status_code == 422
data = response.json()
# Error is wrapped - check message contains relevant info
assert "error_code" in data
assert "insufficient" in data.get("message", "").lower() or data["error_code"] in [
"INSUFFICIENT_INVENTORY",
"VALIDATION_ERROR",
]
def test_update_inventory_admin(
self, client, admin_headers, test_inventory
):
"""Test admin updating inventory entry."""
update_data = {
"quantity": 200,
}
response = client.put(
f"/api/v1/admin/inventory/{test_inventory.id}",
headers=admin_headers,
json=update_data,
)
assert response.status_code == 200
data = response.json()
assert data["quantity"] == 200
def test_update_inventory_not_found(self, client, admin_headers):
"""Test admin updating non-existent inventory."""
update_data = {"quantity": 100}
response = client.put(
"/api/v1/admin/inventory/99999",
headers=admin_headers,
json=update_data,
)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "INVENTORY_NOT_FOUND"
def test_delete_inventory_admin(
self, client, admin_headers, test_product, test_vendor, db
):
"""Test admin deleting inventory entry."""
# Create a new inventory entry to delete
from models.database.inventory import Inventory
new_inventory = Inventory(
product_id=test_product.id,
vendor_id=test_vendor.id,
location="TO_DELETE_WAREHOUSE",
quantity=50,
)
db.add(new_inventory)
db.commit()
db.refresh(new_inventory)
inventory_id = new_inventory.id
response = client.delete(
f"/api/v1/admin/inventory/{inventory_id}",
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert "message" in data
assert "deleted" in data["message"].lower()
# Verify it's deleted
deleted = db.query(Inventory).filter(Inventory.id == inventory_id).first()
assert deleted is None
def test_delete_inventory_not_found(self, client, admin_headers):
"""Test admin deleting non-existent inventory."""
response = client.delete(
"/api/v1/admin/inventory/99999",
headers=admin_headers,
)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "INVENTORY_NOT_FOUND"
def test_delete_inventory_non_admin(
self, client, auth_headers, test_inventory
):
"""Test non-admin trying to delete inventory."""
response = client.delete(
f"/api/v1/admin/inventory/{test_inventory.id}",
headers=auth_headers,
)
assert response.status_code == 403