317 lines
12 KiB
Python
317 lines
12 KiB
Python
# tests/integration/api/v1/test_pagination.py
|
|
import pytest
|
|
|
|
from models.database.product import Product
|
|
from models.database.shop import Shop
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.api
|
|
@pytest.mark.database
|
|
@pytest.mark.products
|
|
class TestPagination:
|
|
|
|
def test_product_pagination_success(self, client, auth_headers, db):
|
|
"""Test pagination for product listing successfully"""
|
|
import uuid
|
|
unique_suffix = str(uuid.uuid4())[:8]
|
|
|
|
# Create multiple products
|
|
products = []
|
|
for i in range(25):
|
|
product = Product(
|
|
product_id=f"PAGE{i:03d}_{unique_suffix}",
|
|
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 # At least our test products
|
|
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 (should have remaining products)
|
|
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 # At least 5 remaining from our test set
|
|
|
|
def test_pagination_boundary_negative_skip_validation_error(self, client, auth_headers):
|
|
"""Test negative skip parameter returns ValidationException"""
|
|
response = client.get("/api/v1/product?skip=-1", headers=auth_headers)
|
|
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
assert data["error_code"] == "VALIDATION_ERROR"
|
|
assert data["status_code"] == 422
|
|
assert "Request validation failed" in data["message"]
|
|
assert "validation_errors" in data["details"]
|
|
|
|
def test_pagination_boundary_zero_limit_validation_error(self, client, auth_headers):
|
|
"""Test zero limit parameter returns ValidationException"""
|
|
response = client.get("/api/v1/product?limit=0", headers=auth_headers)
|
|
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
assert data["error_code"] == "VALIDATION_ERROR"
|
|
assert data["status_code"] == 422
|
|
assert "Request validation failed" in data["message"]
|
|
|
|
def test_pagination_boundary_excessive_limit_validation_error(self, client, auth_headers):
|
|
"""Test excessive limit parameter returns ValidationException"""
|
|
response = client.get("/api/v1/product?limit=10000", headers=auth_headers)
|
|
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
assert data["error_code"] == "VALIDATION_ERROR"
|
|
assert data["status_code"] == 422
|
|
assert "Request validation failed" in data["message"]
|
|
|
|
def test_pagination_beyond_available_records(self, client, auth_headers, db):
|
|
"""Test pagination beyond available records returns empty results"""
|
|
# Test skip beyond available records
|
|
response = client.get("/api/v1/product?skip=10000&limit=10", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["products"] == []
|
|
assert data["skip"] == 10000
|
|
assert data["limit"] == 10
|
|
# total should still reflect actual count
|
|
|
|
def test_pagination_with_filters(self, client, auth_headers, db):
|
|
"""Test pagination combined with filtering"""
|
|
import uuid
|
|
unique_suffix = str(uuid.uuid4())[:8]
|
|
|
|
# Create products with same brand for filtering
|
|
products = []
|
|
for i in range(15):
|
|
product = Product(
|
|
product_id=f"FILTPAGE{i:03d}_{unique_suffix}",
|
|
title=f"Filter Page Product {i}",
|
|
brand="FilterBrand",
|
|
marketplace="FilterMarket",
|
|
)
|
|
products.append(product)
|
|
|
|
db.add_all(products)
|
|
db.commit()
|
|
|
|
# Test first page with filter
|
|
response = client.get(
|
|
"/api/v1/product?brand=FilterBrand&limit=5&skip=0",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["products"]) == 5
|
|
assert data["total"] >= 15 # At least our test products
|
|
|
|
# Verify all products have the filtered brand
|
|
test_products = [p for p in data["products"] if p["product_id"].endswith(unique_suffix)]
|
|
for product in test_products:
|
|
assert product["brand"] == "FilterBrand"
|
|
|
|
# Test second page with same filter
|
|
response = client.get(
|
|
"/api/v1/product?brand=FilterBrand&limit=5&skip=5",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["products"]) == 5
|
|
assert data["skip"] == 5
|
|
|
|
def test_pagination_default_values(self, client, auth_headers):
|
|
"""Test pagination with default values"""
|
|
response = client.get("/api/v1/product", headers=auth_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["skip"] == 0 # Default skip
|
|
assert data["limit"] == 100 # Default limit
|
|
assert len(data["products"]) <= 100 # Should not exceed limit
|
|
|
|
def test_pagination_consistency(self, client, auth_headers, db):
|
|
"""Test pagination consistency across multiple requests"""
|
|
import uuid
|
|
unique_suffix = str(uuid.uuid4())[:8]
|
|
|
|
# Create products with predictable ordering
|
|
products = []
|
|
for i in range(10):
|
|
product = Product(
|
|
product_id=f"CONSIST{i:03d}_{unique_suffix}",
|
|
title=f"Consistent Product {i:03d}",
|
|
marketplace="ConsistentMarket",
|
|
)
|
|
products.append(product)
|
|
|
|
db.add_all(products)
|
|
db.commit()
|
|
|
|
# Get first page
|
|
response1 = client.get("/api/v1/product?limit=5&skip=0", headers=auth_headers)
|
|
assert response1.status_code == 200
|
|
first_page_ids = [p["product_id"] for p in response1.json()["products"]]
|
|
|
|
# Get second page
|
|
response2 = client.get("/api/v1/product?limit=5&skip=5", headers=auth_headers)
|
|
assert response2.status_code == 200
|
|
second_page_ids = [p["product_id"] for p in response2.json()["products"]]
|
|
|
|
# Verify no overlap between pages
|
|
overlap = set(first_page_ids) & set(second_page_ids)
|
|
assert len(overlap) == 0, "Pages should not have overlapping products"
|
|
|
|
def test_shop_pagination_success(self, client, admin_headers, db, test_user):
|
|
"""Test pagination for shop listing successfully"""
|
|
import uuid
|
|
unique_suffix = str(uuid.uuid4())[:8]
|
|
|
|
# Create multiple shops for pagination testing
|
|
from models.database.shop import Shop
|
|
shops = []
|
|
for i in range(15):
|
|
shop = Shop(
|
|
shop_code=f"PAGESHOP{i:03d}_{unique_suffix}",
|
|
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 (assuming admin endpoint exists)
|
|
response = client.get(
|
|
"/api/v1/shop?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
|
|
|
|
def test_stock_pagination_success(self, client, auth_headers, db):
|
|
"""Test pagination for stock listing successfully"""
|
|
import uuid
|
|
unique_suffix = str(uuid.uuid4())[:8]
|
|
|
|
# Create multiple stock entries
|
|
from models.database.stock import Stock
|
|
stocks = []
|
|
for i in range(20):
|
|
stock = Stock(
|
|
gtin=f"123456789{i:04d}",
|
|
location=f"LOC_{unique_suffix}_{i}",
|
|
quantity=10 + i,
|
|
)
|
|
stocks.append(stock)
|
|
|
|
db.add_all(stocks)
|
|
db.commit()
|
|
|
|
# Test first page
|
|
response = client.get("/api/v1/stock?limit=8&skip=0", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 8
|
|
|
|
# Test second page
|
|
response = client.get("/api/v1/stock?limit=8&skip=8", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 8
|
|
|
|
def test_pagination_performance_large_offset(self, client, auth_headers, db):
|
|
"""Test pagination performance with large offset values"""
|
|
# Test with large skip value (should still be reasonable performance)
|
|
import time
|
|
|
|
start_time = time.time()
|
|
response = client.get("/api/v1/product?skip=1000&limit=10", headers=auth_headers)
|
|
end_time = time.time()
|
|
|
|
assert response.status_code == 200
|
|
assert end_time - start_time < 5.0 # Should complete within 5 seconds
|
|
|
|
data = response.json()
|
|
assert data["skip"] == 1000
|
|
assert data["limit"] == 10
|
|
|
|
def test_pagination_with_invalid_parameters_types(self, client, auth_headers):
|
|
"""Test pagination with invalid parameter types returns ValidationException"""
|
|
# Test non-numeric skip
|
|
response = client.get("/api/v1/product?skip=invalid", headers=auth_headers)
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
assert data["error_code"] == "VALIDATION_ERROR"
|
|
|
|
# Test non-numeric limit
|
|
response = client.get("/api/v1/product?limit=invalid", headers=auth_headers)
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
assert data["error_code"] == "VALIDATION_ERROR"
|
|
|
|
# Test float values (should be converted or rejected)
|
|
response = client.get("/api/v1/product?skip=10.5&limit=5.5", headers=auth_headers)
|
|
assert response.status_code in [200, 422] # Depends on implementation
|
|
|
|
def test_empty_dataset_pagination(self, client, auth_headers):
|
|
"""Test pagination behavior with empty dataset"""
|
|
# Use a filter that should return no results
|
|
response = client.get(
|
|
"/api/v1/product?brand=NonexistentBrand999&limit=10&skip=0",
|
|
headers=auth_headers
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["products"] == []
|
|
assert data["total"] == 0
|
|
assert data["skip"] == 0
|
|
assert data["limit"] == 10
|
|
|
|
def test_exception_structure_in_pagination_errors(self, client, auth_headers):
|
|
"""Test that pagination validation errors follow consistent exception structure"""
|
|
response = client.get("/api/v1/product?skip=-1", headers=auth_headers)
|
|
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
|
|
# Verify exception structure matches LetzShopException.to_dict()
|
|
required_fields = ["error_code", "message", "status_code"]
|
|
for field in required_fields:
|
|
assert field in data, f"Missing required field: {field}"
|
|
|
|
assert isinstance(data["error_code"], str)
|
|
assert isinstance(data["message"], str)
|
|
assert isinstance(data["status_code"], int)
|
|
assert data["error_code"] == "VALIDATION_ERROR"
|
|
assert data["status_code"] == 422
|
|
|
|
# Details should contain validation errors
|
|
if "details" in data:
|
|
assert isinstance(data["details"], dict)
|
|
assert "validation_errors" in data["details"]
|