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

refactor: reorganize tests into admin and vendor subdirectories

Browse Source 
Split integration tests into logical admin/ and vendor/ subdirectories
for better organization. Updated fixture imports and test structure.

🤖 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-13 14:51:05 +01:00
parent 193712aad7
commit 2f770709fd
20 changed files with 1235 additions and 1966 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
tests/integration/api/v1/admin/test_auth.py Normal file
View File

@@ -0,0 +1,169 @@
# tests/integration/api/v1/admin/test_auth.py
"""Integration tests for admin authentication endpoints.
Tests the /api/v1/admin/auth/* endpoints.
"""
from datetime import UTC, datetime, timedelta
import pytest
from jose import jwt
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.auth
class TestAdminAuthAPI:
"""Test admin authentication endpoints at /api/v1/admin/auth/*."""
def test_login_success(self, client, test_admin):
"""Test successful admin login."""
response = client.post(
"/api/v1/admin/auth/login",
json={"email_or_username": test_admin.username, "password": "adminpass123"},
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert data["token_type"] == "bearer"
assert "expires_in" in data
assert data["user"]["username"] == test_admin.username
assert data["user"]["email"] == test_admin.email
def test_login_with_email(self, client, test_admin):
"""Test admin login with email instead of username."""
response = client.post(
"/api/v1/admin/auth/login",
json={"email_or_username": test_admin.email, "password": "adminpass123"},
)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert data["user"]["email"] == test_admin.email
def test_login_wrong_password(self, client, test_admin):
"""Test login with wrong password."""
response = client.post(
"/api/v1/admin/auth/login",
json={"email_or_username": test_admin.username, "password": "wrongpassword"},
)
assert response.status_code == 401
data = response.json()
assert data["error_code"] == "INVALID_CREDENTIALS"
def test_login_nonexistent_user(self, client):
"""Test login with nonexistent user."""
response = client.post(
"/api/v1/admin/auth/login",
json={"email_or_username": "nonexistent", "password": "password123"},
)
assert response.status_code == 401
data = response.json()
assert data["error_code"] == "INVALID_CREDENTIALS"
def test_login_inactive_user(self, client, db, test_admin):
"""Test login with inactive admin account."""
original_status = test_admin.is_active
test_admin.is_active = False
db.commit()
try:
response = client.post(
"/api/v1/admin/auth/login",
json={"email_or_username": test_admin.username, "password": "adminpass123"},
)
assert response.status_code == 403
data = response.json()
assert data["error_code"] == "USER_NOT_ACTIVE"
finally:
test_admin.is_active = original_status
db.commit()
def test_login_non_admin_user_rejected(self, client, test_user):
"""Test that non-admin users cannot use admin login."""
response = client.post(
"/api/v1/admin/auth/login",
json={"email_or_username": test_user.username, "password": "testpass123"},
)
assert response.status_code == 401
data = response.json()
assert data["error_code"] == "INVALID_CREDENTIALS"
def test_login_validation_error(self, client):
"""Test login with invalid request format."""
response = client.post(
"/api/v1/admin/auth/login",
json={
"email_or_username": "", # Empty
},
)
assert response.status_code == 422
data = response.json()
assert data["error_code"] == "VALIDATION_ERROR"
def test_get_current_admin_info(self, client, admin_headers, test_admin):
"""Test getting current admin user info."""
response = client.get("/api/v1/admin/auth/me", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert data["username"] == test_admin.username
assert data["email"] == test_admin.email
assert data["role"] == "admin"
assert data["is_active"] is True
def test_get_current_admin_without_auth(self, client):
"""Test getting current admin without authentication."""
response = client.get("/api/v1/admin/auth/me")
assert response.status_code == 401
data = response.json()
assert data["error_code"] == "INVALID_TOKEN"
def test_get_current_admin_invalid_token(self, client):
"""Test getting current admin with invalid token."""
response = client.get(
"/api/v1/admin/auth/me", headers={"Authorization": "Bearer invalid_token"}
)
assert response.status_code == 401
data = response.json()
assert data["error_code"] == "INVALID_TOKEN"
def test_get_current_admin_expired_token(self, client, test_admin, auth_manager):
"""Test getting current admin with expired token."""
expired_payload = {
"sub": str(test_admin.id),
"username": test_admin.username,
"email": test_admin.email,
"role": test_admin.role,
"exp": datetime.now(UTC) - timedelta(hours=1),
"iat": datetime.now(UTC) - timedelta(hours=2),
}
expired_token = jwt.encode(
expired_payload, auth_manager.secret_key, algorithm=auth_manager.algorithm
)
response = client.get(
"/api/v1/admin/auth/me", headers={"Authorization": f"Bearer {expired_token}"}
)
assert response.status_code == 401
data = response.json()
assert data["error_code"] == "TOKEN_EXPIRED"
def test_logout(self, client, admin_headers):
"""Test admin logout."""
response = client.post("/api/v1/admin/auth/logout", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert data["message"] == "Logged out successfully"

89
tests/integration/api/v1/admin/test_dashboard.py Normal file
View File

@@ -0,0 +1,89 @@
# tests/integration/api/v1/admin/test_dashboard.py
"""
Integration tests for admin dashboard and statistics endpoints.
Tests the /api/v1/admin/dashboard/* endpoints.
"""
import pytest
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
@pytest.mark.stats
class TestAdminDashboardAPI:
"""Tests for admin dashboard endpoints."""
def test_get_dashboard(self, client, admin_headers):
"""Test getting admin dashboard."""
response = client.get("/api/v1/admin/dashboard", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert "platform" in data
assert "users" in data
assert "vendors" in data
assert "recent_vendors" in data
assert "recent_imports" in data
def test_get_dashboard_non_admin(self, client, auth_headers):
"""Test non-admin cannot access dashboard."""
response = client.get("/api/v1/admin/dashboard", headers=auth_headers)
assert response.status_code == 403
data = response.json()
assert data["error_code"] == "ADMIN_REQUIRED"
def test_get_comprehensive_stats(self, client, admin_headers, test_marketplace_product):
"""Test getting comprehensive statistics."""
response = client.get("/api/v1/admin/dashboard/stats", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert "total_products" in data
assert "unique_brands" in data
assert "unique_categories" in data
assert "unique_marketplaces" in data
assert "unique_vendors" in data
assert data["total_products"] >= 0
def test_get_marketplace_stats(self, client, admin_headers, test_marketplace_product):
"""Test getting marketplace statistics."""
response = client.get(
"/api/v1/admin/dashboard/stats/marketplace", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
if len(data) > 0:
assert "marketplace" in data[0]
assert "total_products" in data[0]
assert "unique_vendors" in data[0]
def test_get_platform_stats(self, client, admin_headers):
"""Test getting platform statistics."""
response = client.get(
"/api/v1/admin/dashboard/stats/platform", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert "users" in data
assert "vendors" in data
assert "products" in data
assert "orders" in data
assert "imports" in data
def test_get_stats_without_auth(self, client):
"""Test that stats endpoints require authentication."""
response = client.get("/api/v1/admin/dashboard/stats")
assert response.status_code == 401
def test_get_stats_non_admin(self, client, auth_headers):
"""Test non-admin cannot access stats."""
response = client.get("/api/v1/admin/dashboard/stats", headers=auth_headers)
assert response.status_code == 403
data = response.json()
assert data["error_code"] == "ADMIN_REQUIRED"

62
tests/integration/api/v1/admin/test_marketplace.py Normal file
View File

@@ -0,0 +1,62 @@
# tests/integration/api/v1/admin/test_marketplace.py
"""Integration tests for admin marketplace import job endpoints.
Tests the /api/v1/admin/marketplace-import-jobs/* endpoints.
"""
import pytest
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
class TestAdminMarketplaceAPI:
"""Test admin marketplace import job endpoints at /api/v1/admin/marketplace-import-jobs/*."""
def test_get_marketplace_import_jobs_admin(
self, client, admin_headers, test_marketplace_import_job
):
"""Test admin getting marketplace import jobs."""
response = client.get(
"/api/v1/admin/marketplace-import-jobs", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "total" in data
assert "page" in data
assert "limit" in data
assert len(data["items"]) >= 1
# Check that test_marketplace_import_job is in the response
job_ids = [job["job_id"] for job in data["items"] if "job_id" in job]
assert test_marketplace_import_job.id in job_ids
def test_get_marketplace_import_jobs_with_filters(
self, client, admin_headers, test_marketplace_import_job
):
"""Test admin getting marketplace import jobs with filters."""
response = client.get(
"/api/v1/admin/marketplace-import-jobs",
params={"marketplace": test_marketplace_import_job.marketplace},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert "items" in data
assert len(data["items"]) >= 1
assert all(
job["marketplace"] == test_marketplace_import_job.marketplace
for job in data["items"]
)
def test_get_marketplace_import_jobs_non_admin(self, client, auth_headers):
"""Test non-admin trying to access marketplace import jobs."""
response = client.get(
"/api/v1/admin/marketplace-import-jobs", headers=auth_headers
)
assert response.status_code == 403
data = response.json()
assert data["error_code"] == "ADMIN_REQUIRED"

292
tests/integration/api/v1/admin/test_products.py Normal file
View File

@@ -0,0 +1,292 @@
# tests/integration/api/v1/admin/test_products.py
"""
Integration tests for admin marketplace product catalog endpoints.
Tests the /api/v1/admin/products endpoints.
"""
import pytest
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
@pytest.mark.products
class TestAdminProductsAPI:
"""Tests for admin marketplace products endpoints."""
def test_get_products_admin(self, client, admin_headers, test_marketplace_product):
"""Test admin getting all marketplace products."""
response = client.get("/api/v1/admin/products", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert "products" in data
assert "total" in data
assert "skip" in data
assert "limit" in data
assert data["total"] >= 1
assert len(data["products"]) >= 1
# Check that test_marketplace_product is in the response
product_ids = [p["marketplace_product_id"] for p in data["products"]]
assert test_marketplace_product.marketplace_product_id in product_ids
def test_get_products_non_admin(self, client, auth_headers):
"""Test non-admin trying to access admin products endpoint."""
response = client.get("/api/v1/admin/products", headers=auth_headers)
assert response.status_code == 403
data = response.json()
assert data["error_code"] == "ADMIN_REQUIRED"
def test_get_products_with_search(
self, client, admin_headers, test_marketplace_product
):
"""Test admin searching products by title."""
response = client.get(
"/api/v1/admin/products",
params={"search": "Test MarketplaceProduct"},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
def test_get_products_with_marketplace_filter(
self, client, admin_headers, test_marketplace_product
):
"""Test admin filtering products by marketplace."""
response = client.get(
"/api/v1/admin/products",
params={"marketplace": test_marketplace_product.marketplace},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
# All products should be from the filtered marketplace
for product in data["products"]:
assert product["marketplace"] == test_marketplace_product.marketplace
def test_get_products_with_vendor_filter(
self, client, admin_headers, test_marketplace_product
):
"""Test admin filtering products by vendor name."""
response = client.get(
"/api/v1/admin/products",
params={"vendor_name": test_marketplace_product.vendor_name},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
def test_get_products_pagination(
self, client, admin_headers, multiple_products
):
"""Test admin products pagination."""
# Test first page
response = client.get(
"/api/v1/admin/products",
params={"skip": 0, "limit": 2},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert len(data["products"]) <= 2
assert data["skip"] == 0
assert data["limit"] == 2
# Test second page
response = client.get(
"/api/v1/admin/products",
params={"skip": 2, "limit": 2},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["skip"] == 2
def test_get_product_stats_admin(
self, client, admin_headers, test_marketplace_product
):
"""Test admin getting product statistics."""
response = client.get("/api/v1/admin/products/stats", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert "total" in data
assert "active" in data
assert "inactive" in data
assert "digital" in data
assert "physical" in data
assert "by_marketplace" in data
assert data["total"] >= 1
def test_get_product_stats_non_admin(self, client, auth_headers):
"""Test non-admin trying to access product stats."""
response = client.get("/api/v1/admin/products/stats", headers=auth_headers)
assert response.status_code == 403
def test_get_marketplaces_admin(
self, client, admin_headers, test_marketplace_product
):
"""Test admin getting list of marketplaces."""
response = client.get(
"/api/v1/admin/products/marketplaces", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert "marketplaces" in data
assert isinstance(data["marketplaces"], list)
assert test_marketplace_product.marketplace in data["marketplaces"]
def test_get_vendors_admin(self, client, admin_headers, test_marketplace_product):
"""Test admin getting list of source vendors."""
response = client.get("/api/v1/admin/products/vendors", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert "vendors" in data
assert isinstance(data["vendors"], list)
assert test_marketplace_product.vendor_name in data["vendors"]
def test_get_product_detail_admin(
self, client, admin_headers, test_marketplace_product
):
"""Test admin getting product detail."""
response = client.get(
f"/api/v1/admin/products/{test_marketplace_product.id}",
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["id"] == test_marketplace_product.id
assert (
data["marketplace_product_id"]
== test_marketplace_product.marketplace_product_id
)
assert data["marketplace"] == test_marketplace_product.marketplace
assert data["vendor_name"] == test_marketplace_product.vendor_name
assert "translations" in data
def test_get_product_detail_not_found(self, client, admin_headers):
"""Test admin getting non-existent product detail."""
response = client.get("/api/v1/admin/products/99999", headers=admin_headers)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "PRODUCT_NOT_FOUND"
def test_copy_to_vendor_admin(
self, client, admin_headers, test_marketplace_product, test_vendor
):
"""Test admin copying marketplace product to vendor catalog."""
response = client.post(
"/api/v1/admin/products/copy-to-vendor",
json={
"marketplace_product_ids": [test_marketplace_product.id],
"vendor_id": test_vendor.id,
"skip_existing": True,
},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert "copied" in data
assert "skipped" in data
assert "failed" in data
assert data["copied"] == 1
assert data["skipped"] == 0
assert data["failed"] == 0
def test_copy_to_vendor_skip_existing(
self, client, admin_headers, test_marketplace_product, test_vendor, db
):
"""Test admin copying product that already exists skips it."""
# First copy
response = client.post(
"/api/v1/admin/products/copy-to-vendor",
json={
"marketplace_product_ids": [test_marketplace_product.id],
"vendor_id": test_vendor.id,
"skip_existing": True,
},
headers=admin_headers,
)
assert response.status_code == 200
assert response.json()["copied"] == 1
# Second copy should skip
response = client.post(
"/api/v1/admin/products/copy-to-vendor",
json={
"marketplace_product_ids": [test_marketplace_product.id],
"vendor_id": test_vendor.id,
"skip_existing": True,
},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["copied"] == 0
assert data["skipped"] == 1
def test_copy_to_vendor_not_found(self, client, admin_headers, test_vendor):
"""Test admin copying non-existent product."""
response = client.post(
"/api/v1/admin/products/copy-to-vendor",
json={
"marketplace_product_ids": [99999],
"vendor_id": test_vendor.id,
"skip_existing": True,
},
headers=admin_headers,
)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "PRODUCT_NOT_FOUND"
def test_copy_to_vendor_invalid_vendor(
self, client, admin_headers, test_marketplace_product
):
"""Test admin copying to non-existent vendor."""
response = client.post(
"/api/v1/admin/products/copy-to-vendor",
json={
"marketplace_product_ids": [test_marketplace_product.id],
"vendor_id": 99999,
"skip_existing": True,
},
headers=admin_headers,
)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "VENDOR_NOT_FOUND"
def test_copy_to_vendor_non_admin(
self, client, auth_headers, test_marketplace_product, test_vendor
):
"""Test non-admin trying to copy products."""
response = client.post(
"/api/v1/admin/products/copy-to-vendor",
json={
"marketplace_product_ids": [test_marketplace_product.id],
"vendor_id": test_vendor.id,
"skip_existing": True,
},
headers=auth_headers,
)
assert response.status_code == 403

120
tests/integration/api/v1/admin/test_users.py Normal file
View File

@@ -0,0 +1,120 @@
# tests/integration/api/v1/admin/test_users.py
"""Integration tests for admin user management endpoints.
Tests the /api/v1/admin/users/* endpoints.
"""
import pytest
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
class TestAdminUsersAPI:
"""Test admin user management endpoints at /api/v1/admin/users/*."""
def test_get_all_users_admin(self, client, admin_headers, test_user):
"""Test admin getting all users."""
response = client.get("/api/v1/admin/users", headers=admin_headers)
assert response.status_code == 200
data = response.json()
# Response is paginated with items, total, page, per_page, pages
assert "items" in data
assert "total" in data
assert "page" in data
assert "per_page" in data
assert "pages" in data
assert data["total"] >= 2 # test_user + admin user
# Check that test_user is in the response
user_ids = [user["id"] for user in data["items"] if "id" in user]
assert test_user.id in user_ids
def test_get_all_users_non_admin(self, client, auth_headers):
"""Test non-admin trying to access admin endpoint."""
response = client.get("/api/v1/admin/users", headers=auth_headers)
assert response.status_code == 403
data = response.json()
assert data["error_code"] == "ADMIN_REQUIRED"
assert "Admin privileges required" in data["message"]
def test_toggle_user_status_admin(self, client, admin_headers, test_user):
"""Test admin toggling user status."""
response = client.put(
f"/api/v1/admin/users/{test_user.id}/status", headers=admin_headers
)
assert response.status_code == 200
message = response.json()["message"]
assert "deactivated" in message or "activated" in message
assert test_user.username in message
def test_toggle_user_status_user_not_found(self, client, admin_headers):
"""Test admin toggling status for non-existent user."""
response = client.put("/api/v1/admin/users/99999/status", headers=admin_headers)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "USER_NOT_FOUND"
assert "User with ID '99999' not found" in data["message"]
def test_toggle_user_status_cannot_modify_self(
self, client, admin_headers, test_admin
):
"""Test that admin cannot modify their own account."""
response = client.put(
f"/api/v1/admin/users/{test_admin.id}/status", headers=admin_headers
)
assert response.status_code == 400
data = response.json()
assert data["error_code"] == "CANNOT_MODIFY_SELF"
assert (
"Cannot perform 'deactivate account' on your own account" in data["message"]
)
def test_toggle_user_status_cannot_modify_admin(
self, client, admin_headers, test_admin, another_admin
):
"""Test that admin cannot modify another admin."""
response = client.put(
f"/api/v1/admin/users/{another_admin.id}/status", headers=admin_headers
)
assert response.status_code == 400
data = response.json()
assert data["error_code"] == "USER_STATUS_CHANGE_FAILED"
assert "Cannot modify another admin user" in data["message"]
def test_get_user_statistics(self, client, admin_headers):
"""Test admin getting user statistics."""
response = client.get("/api/v1/admin/users/stats", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert "total_users" in data
assert "active_users" in data
assert "inactive_users" in data
assert "activation_rate" in data
assert isinstance(data["total_users"], int)
def test_admin_pagination_users(self, client, admin_headers, test_user, test_admin):
"""Test user pagination works correctly."""
response = client.get(
"/api/v1/admin/users?page=1&per_page=1", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert "items" in data
assert len(data["items"]) == 1
assert data["per_page"] == 1
# Test second page
response = client.get(
"/api/v1/admin/users?page=2&per_page=1", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert "items" in data
assert len(data["items"]) >= 0

221
tests/integration/api/v1/admin/test_vendor_products.py Normal file
View File

@@ -0,0 +1,221 @@
# tests/integration/api/v1/admin/test_vendor_products.py
"""
Integration tests for admin vendor product catalog endpoints.
Tests the /api/v1/admin/vendor-products endpoints.
"""
import pytest
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
@pytest.mark.products
class TestAdminVendorProductsAPI:
"""Tests for admin vendor products endpoints."""
def test_get_vendor_products_admin(self, client, admin_headers, test_product):
"""Test admin getting all vendor products."""
response = client.get("/api/v1/admin/vendor-products", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert "products" in data
assert "total" in data
assert "skip" in data
assert "limit" in data
assert data["total"] >= 1
assert len(data["products"]) >= 1
# Check that test_product is in the response
product_ids = [p["id"] for p in data["products"]]
assert test_product.id in product_ids
def test_get_vendor_products_non_admin(self, client, auth_headers):
"""Test non-admin trying to access vendor products endpoint."""
response = client.get("/api/v1/admin/vendor-products", headers=auth_headers)
assert response.status_code == 403
data = response.json()
assert data["error_code"] == "ADMIN_REQUIRED"
def test_get_vendor_products_with_vendor_filter(
self, client, admin_headers, test_product, test_vendor
):
"""Test admin filtering products by vendor."""
response = client.get(
"/api/v1/admin/vendor-products",
params={"vendor_id": test_vendor.id},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
# All products should be from the filtered vendor
for product in data["products"]:
assert product["vendor_id"] == test_vendor.id
def test_get_vendor_products_with_active_filter(
self, client, admin_headers, test_product
):
"""Test admin filtering products by active status."""
response = client.get(
"/api/v1/admin/vendor-products",
params={"is_active": True},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
# All products should be active
for product in data["products"]:
assert product["is_active"] is True
def test_get_vendor_products_with_featured_filter(
self, client, admin_headers, test_product
):
"""Test admin filtering products by featured status."""
response = client.get(
"/api/v1/admin/vendor-products",
params={"is_featured": False},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
# All products should not be featured
for product in data["products"]:
assert product["is_featured"] is False
def test_get_vendor_products_pagination(
self, client, admin_headers, test_product
):
"""Test admin vendor products pagination."""
response = client.get(
"/api/v1/admin/vendor-products",
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_vendor_product_stats_admin(
self, client, admin_headers, test_product
):
"""Test admin getting vendor product statistics."""
response = client.get(
"/api/v1/admin/vendor-products/stats", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert "total" in data
assert "active" in data
assert "inactive" in data
assert "featured" in data
assert "digital" in data
assert "physical" in data
assert "by_vendor" in data
assert data["total"] >= 1
def test_get_vendor_product_stats_non_admin(self, client, auth_headers):
"""Test non-admin trying to access vendor product stats."""
response = client.get(
"/api/v1/admin/vendor-products/stats", headers=auth_headers
)
assert response.status_code == 403
def test_get_catalog_vendors_admin(
self, client, admin_headers, test_product, test_vendor
):
"""Test admin getting list of vendors with products."""
response = client.get(
"/api/v1/admin/vendor-products/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_vendor_product_detail_admin(
self, client, admin_headers, test_product
):
"""Test admin getting vendor product detail."""
response = client.get(
f"/api/v1/admin/vendor-products/{test_product.id}",
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["id"] == test_product.id
assert data["vendor_id"] == test_product.vendor_id
assert data["marketplace_product_id"] == test_product.marketplace_product_id
assert "source_marketplace" in data
assert "source_vendor" in data
def test_get_vendor_product_detail_not_found(self, client, admin_headers):
"""Test admin getting non-existent vendor product detail."""
response = client.get(
"/api/v1/admin/vendor-products/99999", headers=admin_headers
)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "PRODUCT_NOT_FOUND"
def test_remove_vendor_product_admin(
self, client, admin_headers, test_product, db
):
"""Test admin removing product from vendor catalog."""
product_id = test_product.id
response = client.delete(
f"/api/v1/admin/vendor-products/{product_id}",
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert "message" in data
assert "removed" in data["message"].lower()
# Verify product is removed
response = client.get(
f"/api/v1/admin/vendor-products/{product_id}",
headers=admin_headers,
)
assert response.status_code == 404
def test_remove_vendor_product_not_found(self, client, admin_headers):
"""Test admin removing non-existent vendor product."""
response = client.delete(
"/api/v1/admin/vendor-products/99999",
headers=admin_headers,
)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "PRODUCT_NOT_FOUND"
def test_remove_vendor_product_non_admin(
self, client, auth_headers, test_product
):
"""Test non-admin trying to remove product."""
response = client.delete(
f"/api/v1/admin/vendor-products/{test_product.id}",
headers=auth_headers,
)
assert response.status_code == 403

116
tests/integration/api/v1/admin/test_vendors.py Normal file
View File

@@ -0,0 +1,116 @@
# tests/integration/api/v1/admin/test_vendors.py
"""Integration tests for admin vendor management endpoints.
Tests the /api/v1/admin/vendors/* endpoints.
"""
import pytest
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
class TestAdminVendorsAPI:
"""Test admin vendor management endpoints at /api/v1/admin/vendors/*."""
def test_get_all_vendors_admin(self, client, admin_headers, test_vendor):
"""Test admin getting all vendors."""
response = client.get("/api/v1/admin/vendors", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert len(data["vendors"]) >= 1
# Check that test_vendor is in the response
vendor_codes = [
vendor["vendor_code"]
for vendor in data["vendors"]
if "vendor_code" in vendor
]
assert test_vendor.vendor_code in vendor_codes
def test_get_all_vendors_non_admin(self, client, auth_headers):
"""Test non-admin trying to access admin vendor endpoint."""
response = client.get("/api/v1/admin/vendors", headers=auth_headers)
assert response.status_code == 403
data = response.json()
assert data["error_code"] == "ADMIN_REQUIRED"
def test_toggle_vendor_verification_admin(self, client, admin_headers, test_vendor):
"""Test admin setting vendor verification status."""
response = client.put(
f"/api/v1/admin/vendors/{test_vendor.id}/verification",
headers=admin_headers,
json={"is_verified": True},
)
assert response.status_code == 200
data = response.json()
assert "id" in data
assert "vendor_code" in data
assert "is_verified" in data
def test_toggle_vendor_verification_not_found(self, client, admin_headers):
"""Test admin verifying non-existent vendor."""
response = client.put(
"/api/v1/admin/vendors/99999/verification",
headers=admin_headers,
json={"is_verified": True},
)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "VENDOR_NOT_FOUND"
assert "99999" in data["message"]
assert "not found" in data["message"]
def test_toggle_vendor_status_admin(self, client, admin_headers, test_vendor):
"""Test admin setting vendor active status."""
response = client.put(
f"/api/v1/admin/vendors/{test_vendor.id}/status",
headers=admin_headers,
json={"is_active": False},
)
assert response.status_code == 200
data = response.json()
assert "id" in data
assert "vendor_code" in data
assert "is_active" in data
def test_toggle_vendor_status_not_found(self, client, admin_headers):
"""Test admin toggling status for non-existent vendor."""
response = client.put(
"/api/v1/admin/vendors/99999/status",
headers=admin_headers,
json={"is_active": True},
)
assert response.status_code == 404
data = response.json()
assert data["error_code"] == "VENDOR_NOT_FOUND"
def test_get_vendor_statistics(self, client, admin_headers):
"""Test admin getting vendor statistics."""
response = client.get("/api/v1/admin/vendors/stats", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert "total" in data
assert "verified" in data
assert "pending" in data
assert "inactive" in data
assert isinstance(data["total"], int)
def test_admin_pagination_vendors(self, client, admin_headers, test_vendor):
"""Test vendor pagination works correctly."""
response = client.get(
"/api/v1/admin/vendors?skip=0&limit=1", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert len(data["vendors"]) >= 0
assert "skip" in data
assert "limit" in data