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

Tests restructuring

Browse Source
This commit is contained in:
Samir Boulahtit
2025-09-19 21:23:57 +02:00
parent d0924f90c4
commit 366093bbc6
70 changed files with 212 additions and 1957 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
tests/integration/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
# tests/integration/__init__.py
"""Integration tests - multiple components working together."""

3
tests/integration/api/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
# tests/integration/api/__init__.py
"""API integration tests."""

3
tests/integration/api/v1/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
# tests/integration/api/v1/__init__.py
"""API v1 endpoint integration tests."""

195
tests/integration/api/v1/test_admin_endpoints.py Normal file
View File

@@ -0,0 +1,195 @@
# tests/integration/api/v1/test_admin_endpoints.py
import pytest
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
class TestAdminAPI:
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()
assert len(data) >= 2 # test_user + admin user
# Check that test_user is in the response
user_ids = [user["id"] for user in data 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
assert (
"Access denied" in response.json()["detail"]
or "admin" in response.json()["detail"].lower()
)
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
# Verify the username is in the 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
assert "User not found" in response.json()["detail"]
def test_toggle_user_status_cannot_deactivate_self(
self, client, admin_headers, test_admin
):
"""Test that admin cannot deactivate their own account"""
response = client.put(
f"/api/v1/admin/users/{test_admin.id}/status", headers=admin_headers
)
assert response.status_code == 400
assert "Cannot deactivate your own account" in response.json()["detail"]
def test_get_all_shops_admin(self, client, admin_headers, test_shop):
"""Test admin getting all shops"""
response = client.get("/api/v1/admin/shops", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert len(data["shops"]) >= 1
# Check that test_shop is in the response
shop_codes = [
shop["shop_code"] for shop in data["shops"] if "shop_code" in shop
]
assert test_shop.shop_code in shop_codes
def test_get_all_shops_non_admin(self, client, auth_headers):
"""Test non-admin trying to access admin shop endpoint"""
response = client.get("/api/v1/admin/shops", headers=auth_headers)
assert response.status_code == 403
assert (
"Access denied" in response.json()["detail"]
or "admin" in response.json()["detail"].lower()
)
def test_verify_shop_admin(self, client, admin_headers, test_shop):
"""Test admin verifying/unverifying shop"""
response = client.put(
f"/api/v1/admin/shops/{test_shop.id}/verify", headers=admin_headers
)
assert response.status_code == 200
message = response.json()["message"]
assert "verified" in message or "unverified" in message
assert test_shop.shop_code in message
def test_verify_shop_not_found(self, client, admin_headers):
"""Test admin verifying non-existent shop"""
response = client.put("/api/v1/admin/shops/99999/verify", headers=admin_headers)
assert response.status_code == 404
assert "Shop not found" in response.json()["detail"]
def test_toggle_shop_status_admin(self, client, admin_headers, test_shop):
"""Test admin toggling shop status"""
response = client.put(
f"/api/v1/admin/shops/{test_shop.id}/status", headers=admin_headers
)
assert response.status_code == 200
message = response.json()["message"]
assert "activated" in message or "deactivated" in message
assert test_shop.shop_code in message
def test_toggle_shop_status_not_found(self, client, admin_headers):
"""Test admin toggling status for non-existent shop"""
response = client.put("/api/v1/admin/shops/99999/status", headers=admin_headers)
assert response.status_code == 404
assert "Shop not found" in response.json()["detail"]
def test_get_marketplace_import_jobs_admin(
self, client, admin_headers, test_marketplace_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 len(data) >= 1
# Check that test_marketplace_job is in the response
job_ids = [job["job_id"] for job in data if "job_id" in job]
assert test_marketplace_job.id in job_ids
def test_get_marketplace_import_jobs_with_filters(
self, client, admin_headers, test_marketplace_job
):
"""Test admin getting marketplace import jobs with filters"""
response = client.get(
"/api/v1/admin/marketplace-import-jobs",
params={"marketplace": test_marketplace_job.marketplace},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
assert all(
job["marketplace"] == test_marketplace_job.marketplace for job in data
)
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
assert (
"Access denied" in response.json()["detail"]
or "admin" in response.json()["detail"].lower()
)
def test_admin_pagination_users(self, client, admin_headers, test_user, test_admin):
"""Test user pagination works correctly"""
# Test first page
response = client.get(
"/api/v1/admin/users?skip=0&limit=1", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert len(data) == 1
# Test second page
response = client.get(
"/api/v1/admin/users?skip=1&limit=1", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert len(data) >= 0 # Could be 1 or 0 depending on total users
def test_admin_pagination_shops(self, client, admin_headers, test_shop):
"""Test shop pagination works correctly"""
response = client.get(
"/api/v1/admin/shops?skip=0&limit=1", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert len(data["shops"]) >= 0
assert "skip" in data
assert "limit" in data

127
tests/integration/api/v1/test_authentication_endpoints.py Normal file
View File

@@ -0,0 +1,127 @@
# tests/test_authentication_endpoints.py
import pytest
from fastapi import HTTPException
class TestAuthenticationAPI:
def test_register_user_success(self, client, db):
"""Test successful user registration"""
response = client.post(
"/api/v1/auth/register",
json={
"email": "newuser@example.com",
"username": "newuser",
"password": "securepass123",
},
)
assert response.status_code == 200
data = response.json()
assert data["email"] == "newuser@example.com"
assert data["username"] == "newuser"
assert data["role"] == "user"
assert data["is_active"] is True
assert "hashed_password" not in data
def test_register_user_duplicate_email(self, client, test_user):
"""Test registration with duplicate email"""
response = client.post(
"/api/v1/auth/register",
json={
"email": test_user.email, # Same as test_user
"username": "newuser",
"password": "securepass123",
},
)
assert response.status_code == 400
assert "Email already registered" in response.json()["detail"]
def test_register_user_duplicate_username(self, client, test_user):
"""Test registration with duplicate username"""
response = client.post(
"/api/v1/auth/register",
json={
"email": "new@example.com",
"username": test_user.username, # Same as test_user
"password": "securepass123",
},
)
assert response.status_code == 400
assert "Username already taken" in response.json()["detail"]
def test_login_success(self, client, test_user):
"""Test successful login"""
response = client.post(
"/api/v1/auth/login",
json={"username": test_user.username, "password": "testpass123"},
)
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_user.username
def test_login_wrong_password(self, client, test_user):
"""Test login with wrong password"""
response = client.post(
"/api/v1/auth/login",
json={"username": "testuser", "password": "wrongpassword"},
)
assert response.status_code == 401
assert "Incorrect username or password" in response.json()["detail"]
def test_login_nonexistent_user(self, client, db): # Added db fixture
"""Test login with nonexistent user"""
response = client.post(
"/api/v1/auth/login",
json={"username": "nonexistent", "password": "password123"},
)
assert response.status_code == 401
def test_get_current_user_info(self, client, auth_headers, test_user):
"""Test getting current user info"""
response = client.get("/api/v1/auth/me", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert data["username"] == test_user.username
assert data["email"] == test_user.email
def test_get_current_user_without_auth(self, client):
"""Test getting current user without authentication"""
response = client.get("/api/v1/auth/me")
assert response.status_code == 401 # No authorization header
class TestAuthManager:
def test_hash_password(self, auth_manager):
"""Test password hashing"""
password = "testpassword123"
hashed = auth_manager.hash_password(password)
assert hashed != password
assert len(hashed) > 20 # bcrypt hashes are long
def test_verify_password(self, auth_manager):
"""Test password verification"""
password = "testpassword123"
hashed = auth_manager.hash_password(password)
assert auth_manager.verify_password(password, hashed) is True
assert auth_manager.verify_password("wrongpassword", hashed) is False
def test_create_access_token(self, auth_manager, test_user):
"""Test JWT token creation"""
token_data = auth_manager.create_access_token(test_user)
assert "access_token" in token_data
assert token_data["token_type"] == "bearer"
assert "expires_in" in token_data
assert isinstance(token_data["expires_in"], int)

54
tests/integration/api/v1/test_marketplace_endpoints.py Normal file
View File

@@ -0,0 +1,54 @@
# tests/test_marketplace_endpoints.py
from unittest.mock import AsyncMock, patch
import pytest
class TestMarketplaceAPI:
def test_import_from_marketplace(self, client, auth_headers, test_shop):
"""Test marketplace import endpoint - just test job creation"""
import_data = {
"url": "https://example.com/products.csv",
"marketplace": "TestMarket",
"shop_code": test_shop.shop_code,
}
response = client.post(
"/api/v1/marketplace/import-product", headers=auth_headers, json=import_data
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "pending"
assert data["marketplace"] == "TestMarket"
assert "job_id" in data
# Don't test the background task here - test it separately
def test_import_from_marketplace_invalid_shop(self, client, auth_headers):
"""Test marketplace import with invalid shop"""
import_data = {
"url": "https://example.com/products.csv",
"marketplace": "TestMarket",
"shop_code": "NONEXISTENT",
}
response = client.post(
"/api/v1/marketplace/import-product", headers=auth_headers, json=import_data
)
assert response.status_code == 404
assert "Shop not found" in response.json()["detail"]
def test_get_marketplace_import_jobs(self, client, auth_headers):
"""Test getting marketplace import jobs"""
response = client.get("/api/v1/marketplace/import-jobs", headers=auth_headers)
assert response.status_code == 200
assert isinstance(response.json(), list)
def test_get_marketplace_without_auth(self, client):
"""Test that marketplace endpoints require authentication"""
response = client.get("/api/v1/marketplace/import-jobs")
assert response.status_code == 401 # No authorization header

87
tests/integration/api/v1/test_pagination.py Normal file
View File

@@ -0,0 +1,87 @@
# tests/integration/api/v1/test_pagination.py
import pytest
from models.database_models import Product, Shop
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.database
@pytest.mark.products
@pytest.mark.shops
class TestPagination:
def test_product_pagination(self, client, auth_headers, db):
"""Test pagination for product listing"""
# Create multiple products
products = []
for i in range(25):
product = Product(
product_id=f"PAGE{i:03d}",
title=f"Pagination Test Product {i}",
marketplace="PaginationTest",
)
products.append(product)
db.add_all(products)
db.commit()
# Test first page
response = client.get("/api/v1/product?limit=10&skip=0", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert len(data["products"]) == 10
assert data["total"] == 25
assert data["skip"] == 0
assert data["limit"] == 10
# Test second page
response = client.get("/api/v1/product?limit=10&skip=10", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert len(data["products"]) == 10
assert data["skip"] == 10
# Test last page
response = client.get("/api/v1/product?limit=10&skip=20", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert len(data["products"]) == 5 # Only 5 remaining
def test_pagination_boundaries(self, client, auth_headers):
"""Test pagination boundary conditions"""
# Test negative skip
response = client.get("/api/v1/product?skip=-1", headers=auth_headers)
assert response.status_code == 422 # Validation error
# Test zero limit
response = client.get("/api/v1/product?limit=0", headers=auth_headers)
assert response.status_code == 422 # Validation error
# Test excessive limit
response = client.get("/api/v1/product?limit=10000", headers=auth_headers)
assert response.status_code == 422 # Should be limited
def test_shop_pagination(self, client, admin_headers, db, test_user):
"""Test pagination for shop listing"""
# Create multiple shops for pagination testing
shops = []
for i in range(15):
shop = Shop(
shop_code=f"PAGESHOP{i:03d}",
shop_name=f"Pagination Shop {i}",
owner_id=test_user.id,
is_active=True,
)
shops.append(shop)
db.add_all(shops)
db.commit()
# Test first page
response = client.get("/api/v1/admin/shops?limit=5&skip=0", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert len(data["shops"]) == 5
assert data["total"] >= 15 # At least our test shops
assert data["skip"] == 0
assert data["limit"] == 5

140
tests/integration/api/v1/test_product_endpoints.py Normal file
View File

@@ -0,0 +1,140 @@
# tests/test_product_endpoints.py
import pytest
class TestProductsAPI:
def test_get_products_empty(self, client, auth_headers):
"""Test getting products when none exist"""
response = client.get("/api/v1/product", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert data["products"] == []
assert data["total"] == 0
def test_get_products_with_data(self, client, auth_headers, test_product):
"""Test getting products with data"""
response = client.get("/api/v1/product", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert len(data["products"]) == 1
assert data["total"] == 1
assert data["products"][0]["product_id"] == "TEST001"
def test_get_products_with_filters(self, client, auth_headers, test_product):
"""Test filtering products"""
# Test brand filter
response = client.get("/api/v1/product?brand=TestBrand", headers=auth_headers)
assert response.status_code == 200
assert response.json()["total"] == 1
# Test marketplace filter
response = client.get(
"/api/v1/product?marketplace=Letzshop", headers=auth_headers
)
assert response.status_code == 200
assert response.json()["total"] == 1
# Test search
response = client.get("/api/v1/product?search=Test", headers=auth_headers)
assert response.status_code == 200
assert response.json()["total"] == 1
def test_create_product(self, client, auth_headers):
"""Test creating a new product"""
product_data = {
"product_id": "NEW001",
"title": "New Product",
"description": "A new product",
"price": "15.99",
"brand": "NewBrand",
"gtin": "9876543210987",
"availability": "in stock",
"marketplace": "Amazon",
}
response = client.post(
"/api/v1/product", headers=auth_headers, json=product_data
)
assert response.status_code == 200
data = response.json()
assert data["product_id"] == "NEW001"
assert data["title"] == "New Product"
assert data["marketplace"] == "Amazon"
def test_create_product_duplicate_id(self, client, auth_headers, test_product):
"""Test creating product with duplicate ID"""
product_data = {
"product_id": test_product.product_id,
"title": test_product.title,
"description": "A new product",
"price": "15.99",
"brand": "NewBrand",
"gtin": "9876543210987",
"availability": "in stock",
"marketplace": "Amazon",
}
response = client.post(
"/api/v1/product", headers=auth_headers, json=product_data
)
# Debug output
print(f"Status Code: {response.status_code}")
print(f"Response Content: {response.content}")
try:
print(f"Response JSON: {response.json()}")
except:
print("Could not parse response as JSON")
assert response.status_code == 400
assert "already exists" in response.json()["detail"]
def test_get_product_by_id(self, client, auth_headers, test_product):
"""Test getting specific product"""
response = client.get(
f"/api/v1/product/{test_product.product_id}", headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert data["product"]["product_id"] == test_product.product_id
assert data["product"]["title"] == test_product.title
def test_get_nonexistent_product(self, client, auth_headers):
"""Test getting nonexistent product"""
response = client.get("/api/v1/product/NONEXISTENT", headers=auth_headers)
assert response.status_code == 404
def test_update_product(self, client, auth_headers, test_product):
"""Test updating product"""
update_data = {"title": "Updated Product Title", "price": "25.99"}
response = client.put(
f"/api/v1/product/{test_product.product_id}",
headers=auth_headers,
json=update_data,
)
assert response.status_code == 200
data = response.json()
assert data["title"] == "Updated Product Title"
assert data["price"] == "25.99"
def test_delete_product(self, client, auth_headers, test_product):
"""Test deleting product"""
response = client.delete(
f"/api/v1/product/{test_product.product_id}", headers=auth_headers
)
assert response.status_code == 200
assert "deleted successfully" in response.json()["message"]
def test_get_product_without_auth(self, client):
"""Test that product endpoints require authentication"""
response = client.get("/api/v1/product")
assert response.status_code == 401 # No authorization header

57
tests/integration/api/v1/test_shop_endpoints.py Normal file
View File

@@ -0,0 +1,57 @@
# tests/test_shop_endpoints.py
import pytest
class TestShopsAPI:
def test_create_shop(self, client, auth_headers):
"""Test creating a new shop"""
shop_data = {
"shop_code": "NEWSHOP",
"shop_name": "New Shop",
"description": "A new test shop",
}
response = client.post("/api/v1/shop", headers=auth_headers, json=shop_data)
assert response.status_code == 200
data = response.json()
assert data["shop_code"] == "NEWSHOP"
assert data["shop_name"] == "New Shop"
assert data["is_active"] is True
def test_create_shop_duplicate_code(self, client, auth_headers, test_shop):
"""Test creating shop with duplicate code"""
shop_data = {
"shop_code": test_shop.shop_code, # Same as test_shop
"shop_name": test_shop.shop_name,
}
response = client.post("/api/v1/shop", headers=auth_headers, json=shop_data)
assert response.status_code == 400
assert "already exists" in response.json()["detail"]
def test_get_shops(self, client, auth_headers, test_shop):
"""Test getting shops list"""
response = client.get("/api/v1/shop", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert len(data["shops"]) >= 1
def test_get_shop_by_code(self, client, auth_headers, test_shop):
"""Test getting specific shop"""
response = client.get(
f"/api/v1/shop/{test_shop.shop_code}", headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert data["shop_code"] == test_shop.shop_code
assert data["shop_name"] == test_shop.shop_name
def test_get_shop_without_auth(self, client):
"""Test that shop endpoints require authentication"""
response = client.get("/api/v1/shop")
assert response.status_code == 401 # No authorization header

33
tests/integration/api/v1/test_stats_endpoints.py Normal file
View File

@@ -0,0 +1,33 @@
# tests/test_stats_endpoints.py
import pytest
class TestStatsAPI:
def test_get_basic_stats(self, client, auth_headers, test_product):
"""Test getting basic statistics"""
response = client.get("/api/v1/stats", headers=auth_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_shops" in data
assert data["total_products"] >= 1
def test_get_marketplace_stats(self, client, auth_headers, test_product):
"""Test getting marketplace statistics"""
response = client.get("/api/v1/stats/marketplace", headers=auth_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]
def test_get_stats_without_auth(self, client):
"""Test that stats endpoints require authentication"""
response = client.get("/api/v1/stats")
assert response.status_code == 401 # No authorization header

154
tests/integration/api/v1/test_stock_endpoints.py Normal file
View File

@@ -0,0 +1,154 @@
# tests/test_stock_endpoints.py
import pytest
from models.database_models import Stock
class TestStockAPI:
def test_set_stock_new(self, client, auth_headers):
"""Test setting stock for new GTIN"""
stock_data = {
"gtin": "1234567890123",
"location": "WAREHOUSE_A",
"quantity": 100,
}
response = client.post("/api/v1/stock", headers=auth_headers, json=stock_data)
assert response.status_code == 200
data = response.json()
assert data["gtin"] == "1234567890123"
assert data["location"] == "WAREHOUSE_A"
assert data["quantity"] == 100
def test_set_stock_existing(self, client, auth_headers, db):
"""Test updating existing stock"""
# Create initial stock
stock = Stock(gtin="1234567890123", location="WAREHOUSE_A", quantity=50)
db.add(stock)
db.commit()
stock_data = {
"gtin": "1234567890123",
"location": "WAREHOUSE_A",
"quantity": 75,
}
response = client.post("/api/v1/stock", headers=auth_headers, json=stock_data)
assert response.status_code == 200
data = response.json()
assert data["quantity"] == 75 # Should be replaced, not added
def test_add_stock(self, client, auth_headers, db):
"""Test adding to existing stock"""
# Create initial stock
stock = Stock(gtin="1234567890123", location="WAREHOUSE_A", quantity=50)
db.add(stock)
db.commit()
stock_data = {
"gtin": "1234567890123",
"location": "WAREHOUSE_A",
"quantity": 25,
}
response = client.post(
"/api/v1/stock/add", headers=auth_headers, json=stock_data
)
assert response.status_code == 200
data = response.json()
assert data["quantity"] == 75 # 50 + 25
def test_remove_stock(self, client, auth_headers, db):
"""Test removing from existing stock"""
# Create initial stock
stock = Stock(gtin="1234567890123", location="WAREHOUSE_A", quantity=50)
db.add(stock)
db.commit()
stock_data = {
"gtin": "1234567890123",
"location": "WAREHOUSE_A",
"quantity": 15,
}
response = client.post(
"/api/v1/stock/remove", headers=auth_headers, json=stock_data
)
assert response.status_code == 200
data = response.json()
assert data["quantity"] == 35 # 50 - 15
def test_remove_stock_insufficient(self, client, auth_headers, db):
"""Test removing more stock than available"""
# Create initial stock
stock = Stock(gtin="1234567890123", location="WAREHOUSE_A", quantity=10)
db.add(stock)
db.commit()
stock_data = {
"gtin": "1234567890123",
"location": "WAREHOUSE_A",
"quantity": 20,
}
response = client.post(
"/api/v1/stock/remove", headers=auth_headers, json=stock_data
)
assert response.status_code == 400
assert "Insufficient stock" in response.json()["detail"]
def test_get_stock_by_gtin(self, client, auth_headers, db):
"""Test getting stock summary for GTIN"""
# Create stock in multiple locations
stock1 = Stock(gtin="1234567890123", location="WAREHOUSE_A", quantity=50)
stock2 = Stock(gtin="1234567890123", location="WAREHOUSE_B", quantity=25)
db.add_all([stock1, stock2])
db.commit()
response = client.get("/api/v1/stock/1234567890123", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert data["gtin"] == "1234567890123"
assert data["total_quantity"] == 75
assert len(data["locations"]) == 2
def test_get_total_stock(self, client, auth_headers, db):
"""Test getting total stock for GTIN"""
# Create stock in multiple locations
stock1 = Stock(gtin="1234567890123", location="WAREHOUSE_A", quantity=50)
stock2 = Stock(gtin="1234567890123", location="WAREHOUSE_B", quantity=25)
db.add_all([stock1, stock2])
db.commit()
response = client.get("/api/v1/stock/1234567890123/total", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert data["gtin"] == "1234567890123"
assert data["total_quantity"] == 75
assert data["locations_count"] == 2
def test_get_all_stock(self, client, auth_headers, db):
"""Test getting all stock entries"""
# Create some stock entries
stock1 = Stock(gtin="1234567890123", location="WAREHOUSE_A", quantity=50)
stock2 = Stock(gtin="9876543210987", location="WAREHOUSE_B", quantity=25)
db.add_all([stock1, stock2])
db.commit()
response = client.get("/api/v1/stock", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert len(data) == 2
def test_get_stock_without_auth(self, client):
"""Test that stock endpoints require authentication"""
response = client.get("/api/v1/stock")
assert response.status_code == 401 # No authorization header

6
tests/integration/conftest.py Normal file
View File

@@ -0,0 +1,6 @@
# tests/integration/conftest.py
"""Integration test specific fixtures."""
import pytest
# Add any integration-specific fixtures here if needed

3
tests/integration/security/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
# tests/integration/security/__init__.py
"""Security integration tests."""

75
tests/integration/security/test_authentication.py Normal file
View File

@@ -0,0 +1,75 @@
# tests/integration/security/test_authentication.py
import pytest
@pytest.mark.integration
@pytest.mark.security
@pytest.mark.auth
class TestAuthentication:
def test_protected_endpoint_without_auth(self, client):
"""Test that protected endpoints reject unauthenticated requests"""
protected_endpoints = [
"/api/v1/admin/users",
"/api/v1/admin/shops",
"/api/v1/marketplace/import-jobs",
"/api/v1/product",
"/api/v1/shop",
"/api/v1/stats",
"/api/v1/stock",
]
for endpoint in protected_endpoints:
response = client.get(endpoint)
assert response.status_code == 401 # Authentication missing
def test_protected_endpoint_with_invalid_token(self, client):
"""Test protected endpoints with invalid token"""
headers = {"Authorization": "Bearer invalid_token_here"}
response = client.get("/api/v1/product", headers=headers)
assert response.status_code == 401 # Token is not valid
def test_debug_direct_bearer(self, client):
"""Test HTTPBearer directly"""
response = client.get("/api/v1/debug-bearer")
print(f"Direct Bearer - Status: {response.status_code}")
print(
f"Direct Bearer - Response: {response.json() if response.content else 'No content'}"
)
def test_debug_dependencies(self, client):
"""Debug the dependency chain step by step"""
# Test 1: Direct endpoint with no auth
response = client.get("/api/v1/admin/users")
print(f"Admin endpoint - Status: {response.status_code}")
try:
print(f"Admin endpoint - Response: {response.json()}")
except:
print(f"Admin endpoint - Raw: {response.content}")
# Test 2: Try a regular endpoint that uses get_current_user
response2 = client.get("/api/v1/product") # or any endpoint with get_current_user
print(f"Regular endpoint - Status: {response2.status_code}")
try:
print(f"Regular endpoint - Response: {response2.json()}")
except:
print(f"Regular endpoint - Raw: {response2.content}")
def test_debug_available_routes(self, client):
"""Debug test to see all available routes"""
print("\n=== All Available Routes ===")
for route in client.app.routes:
if hasattr(route, "path") and hasattr(route, "methods"):
print(f"{list(route.methods)} {route.path}")
print("\n=== Testing Product Endpoint Variations ===")
variations = [
"/api/v1/product", # Your current attempt
"/api/v1/product/", # With trailing slash
"/api/v1/product/list", # With list endpoint
"/api/v1/product/all", # With all endpoint
]
for path in variations:
response = client.get(path)
print(f"{path}: Status {response.status_code}")

46
tests/integration/security/test_authorization.py Normal file
View File

@@ -0,0 +1,46 @@
# tests/integration/security/test_authorization.py
import pytest
@pytest.mark.integration
@pytest.mark.security
@pytest.mark.auth
class TestAuthorization:
def test_admin_endpoint_requires_admin_role(self, client, auth_headers):
"""Test that admin endpoints require admin role"""
response = client.get("/api/v1/admin/users", headers=auth_headers)
assert response.status_code == 403
# Regular user should be denied access
def test_admin_endpoints_with_admin_access(self, client, admin_headers):
"""Test that admin users can access admin endpoints"""
admin_endpoints = [
"/api/v1/admin/users",
"/api/v1/admin/shops",
"/api/v1/admin/marketplace-import-jobs",
]
for endpoint in admin_endpoints:
response = client.get(endpoint, headers=admin_headers)
assert response.status_code == 200 # Admin should have access
def test_regular_endpoints_with_user_access(self, client, auth_headers):
"""Test that regular users can access non-admin endpoints"""
user_endpoints = [
"/api/v1/product",
"/api/v1/stats",
"/api/v1/stock",
]
for endpoint in user_endpoints:
response = client.get(endpoint, headers=auth_headers)
assert response.status_code == 200 # Regular user should have access
def test_shop_owner_access_control(self, client, auth_headers, test_shop, other_user):
"""Test that users can only access their own shops"""
# Test accessing own shop (should work)
response = client.get(f"/api/v1/shop/{test_shop.shop_code}", headers=auth_headers)
# Response depends on your implementation - could be 200 or 404 if shop doesn't belong to user
# The exact assertion depends on your shop access control implementation
assert response.status_code in [200, 403, 404]

65
tests/integration/security/test_input_validation.py Normal file
View File

@@ -0,0 +1,65 @@
# tests/integration/security/test_input_validation.py
import pytest
@pytest.mark.integration
@pytest.mark.security
class TestInputValidation:
def test_sql_injection_prevention(self, client, auth_headers):
"""Test SQL injection prevention in search parameters"""
# Try SQL injection in search parameter
malicious_search = "'; DROP TABLE products; --"
response = client.get(
f"/api/v1/product?search={malicious_search}", headers=auth_headers
)
# Should not crash and should return normal response
assert response.status_code == 200
# Database should still be intact (no products dropped)
# def test_input_validation(self, client, auth_headers):
# # TODO: implement sanitization
# """Test input validation and sanitization"""
# # Test XSS attempt in product creation
# xss_payload = "<script>alert('xss')</script>"
#
# product_data = {
# "product_id": "XSS_TEST",
# "title": xss_payload,
# "description": xss_payload,
# }
#
# response = client.post("/api/v1/product", headers=auth_headers, json=product_data)
#
# assert response.status_code == 200
# data = response.json()
# assert "<script>" not in data["title"]
# assert "&lt;script&gt;" in data["title"]
def test_parameter_validation(self, client, auth_headers):
"""Test parameter validation for API endpoints"""
# Test invalid pagination parameters
response = client.get("/api/v1/product?limit=-1", headers=auth_headers)
assert response.status_code == 422 # Validation error
response = client.get("/api/v1/product?skip=-1", headers=auth_headers)
assert response.status_code == 422 # Validation error
def test_json_validation(self, client, auth_headers):
"""Test JSON validation for POST requests"""
# Test invalid JSON structure
response = client.post(
"/api/v1/product",
headers=auth_headers,
content="invalid json content"
)
assert response.status_code == 422 # JSON decode error
# Test missing required fields
response = client.post(
"/api/v1/product",
headers=auth_headers,
json={"title": "Test Product"} # Missing required product_id
)
assert response.status_code == 422 # Validation error