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

refactor: remove db.expunge() anti-pattern from test fixtures

Browse Source 
Remove db.expunge() calls that were causing DetachedInstanceError
when accessing lazy-loaded relationships in tests.

Changes:
- conftest.py: Add documentation about fixture best practices
- auth_fixtures: Remove expunge, keep objects attached to session
- customer_fixtures: Remove expunge, add proper relationship loading
- vendor_fixtures: Remove expunge, add test_company and other_company
  fixtures for proper company-vendor relationship setup
- marketplace_import_job_fixtures: Remove expunge calls
- marketplace_product_fixtures: Remove expunge calls

The db fixture already provides test isolation by dropping/recreating
tables after each test, so expunge is unnecessary and harmful.

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

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Samir Boulahtit
2025-12-05 21:41:50 +01:00
parent 50cb4b0985
commit aaff799b5e
6 changed files with 193 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
tests/fixtures/auth_fixtures.py vendored
View File

@@ -1,4 +1,10 @@
# tests/fixtures/auth_fixtures.py
"""
Authentication-related test fixtures.
Note: Fixtures should NOT use db.expunge() as it breaks lazy loading.
See tests/conftest.py for details on fixture best practices.
"""
import uuid
import pytest
@@ -9,14 +15,14 @@ from models.database.user import User
@pytest.fixture(scope="session")
def auth_manager():
"""Create auth manager instance (session scope since it's stateless)"""
"""Create auth manager instance (session scope since it's stateless)."""
return AuthManager()
@pytest.fixture
def test_user(db, auth_manager):
"""Create a test user with unique username"""
unique_id = str(uuid.uuid4())[:8] # Short unique identifier
"""Create a test user with unique username."""
unique_id = str(uuid.uuid4())[:8]
hashed_password = auth_manager.hash_password("testpass123")
user = User(
email=f"test_{unique_id}@example.com",
@@ -28,16 +34,13 @@ def test_user(db, auth_manager):
db.add(user)
db.commit()
db.refresh(user)
# Expunge user from session to prevent ResourceWarning about unclosed connections
# This detaches the object from the session so it doesn't hold a reference
db.expunge(user)
return user
@pytest.fixture
def test_admin(db, auth_manager):
"""Create a test admin user with unique username"""
unique_id = str(uuid.uuid4())[:8] # Short unique identifier
"""Create a test admin user with unique username."""
unique_id = str(uuid.uuid4())[:8]
hashed_password = auth_manager.hash_password("adminpass123")
admin = User(
email=f"admin_{unique_id}@example.com",
@@ -49,15 +52,13 @@ def test_admin(db, auth_manager):
db.add(admin)
db.commit()
db.refresh(admin)
# Expunge admin from session to prevent ResourceWarning about unclosed connections
db.expunge(admin)
return admin
@pytest.fixture
def another_admin(db, auth_manager):
"""Create another test admin user for testing admin-to-admin interactions"""
unique_id = str(uuid.uuid4())[:8] # Short unique identifier
"""Create another test admin user for testing admin-to-admin interactions."""
unique_id = str(uuid.uuid4())[:8]
hashed_password = auth_manager.hash_password("anotheradminpass123")
admin = User(
email=f"another_admin_{unique_id}@example.com",
@@ -69,14 +70,12 @@ def another_admin(db, auth_manager):
db.add(admin)
db.commit()
db.refresh(admin)
# Expunge admin from session to prevent ResourceWarning about unclosed connections
db.expunge(admin)
return admin
@pytest.fixture
def other_user(db, auth_manager):
"""Create a different user for testing access controls"""
"""Create a different user for testing access controls."""
unique_id = str(uuid.uuid4())[:8]
hashed_password = auth_manager.hash_password("otherpass123")
user = User(
@@ -89,8 +88,6 @@ def other_user(db, auth_manager):
db.add(user)
db.commit()
db.refresh(user)
# Expunge user from session to prevent ResourceWarning about unclosed connections
db.expunge(user)
return user
@@ -120,7 +117,7 @@ def admin_headers(client, test_admin):
@pytest.fixture
def test_vendor_user(db, auth_manager):
"""Create a test vendor user with unique username"""
"""Create a test vendor user with unique username."""
unique_id = str(uuid.uuid4())[:8]
hashed_password = auth_manager.hash_password("vendorpass123")
user = User(
@@ -133,7 +130,6 @@ def test_vendor_user(db, auth_manager):
db.add(user)
db.commit()
db.refresh(user)
db.expunge(user)
return user

35
tests/fixtures/customer_fixtures.py vendored
View File

@@ -1,12 +1,19 @@
# tests/fixtures/customer_fixtures.py
"""
Customer-related test fixtures.
Note: Fixtures should NOT use db.expunge() as it breaks lazy loading.
See tests/conftest.py for details on fixture best practices.
"""
import pytest
from models.database.customer import Customer, CustomerAddress
from models.database.order import Order
@pytest.fixture
def test_customer(db, test_vendor):
"""Create a test customer"""
"""Create a test customer."""
customer = Customer(
vendor_id=test_vendor.id,
email="testcustomer@example.com",
@@ -19,14 +26,12 @@ def test_customer(db, test_vendor):
db.add(customer)
db.commit()
db.refresh(customer)
# Expunge customer from session to prevent ResourceWarning about unclosed connections
db.expunge(customer)
return customer
@pytest.fixture
def test_customer_address(db, test_vendor, test_customer):
"""Create a test customer address"""
"""Create a test customer address."""
address = CustomerAddress(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
@@ -42,6 +47,24 @@ def test_customer_address(db, test_vendor, test_customer):
db.add(address)
db.commit()
db.refresh(address)
# Expunge address from session to prevent ResourceWarning about unclosed connections
db.expunge(address)
return address
@pytest.fixture
def test_order(db, test_vendor, test_customer, test_customer_address):
"""Create a test order."""
order = Order(
vendor_id=test_vendor.id,
customer_id=test_customer.id,
order_number="TEST-ORD-001",
status="pending",
subtotal=99.99,
total_amount=99.99,
currency="EUR",
shipping_address_id=test_customer_address.id,
billing_address_id=test_customer_address.id,
)
db.add(order)
db.commit()
db.refresh(order)
return order

16
tests/fixtures/marketplace_import_job_fixtures.py vendored
View File

@@ -1,4 +1,10 @@
# tests/fixtures/marketplace_import_job_fixtures.py
"""
Marketplace import job test fixtures.
Note: Fixtures should NOT use db.expunge() as it breaks lazy loading.
See tests/conftest.py for details on fixture best practices.
"""
import pytest
from models.database.marketplace_import_job import MarketplaceImportJob
@@ -6,10 +12,9 @@ from models.database.marketplace_import_job import MarketplaceImportJob
@pytest.fixture
def test_marketplace_import_job(db, test_vendor, test_user):
"""Create a test marketplace import job"""
"""Create a test marketplace import job."""
job = MarketplaceImportJob(
marketplace="amazon",
# REMOVED: vendor_name field doesn't exist
status="completed",
source_url="https://test-marketplace.example.com/import",
vendor_id=test_vendor.id,
@@ -23,16 +28,13 @@ def test_marketplace_import_job(db, test_vendor, test_user):
db.add(job)
db.commit()
db.refresh(job)
# Expunge job from session to prevent ResourceWarning about unclosed connections
db.expunge(job)
return job
def create_test_marketplace_import_job(db, vendor_id, user_id, **kwargs):
"""Helper function to create MarketplaceImportJob with defaults"""
"""Helper function to create MarketplaceImportJob with defaults."""
defaults = {
"marketplace": "test",
# REMOVED: name field
"status": "pending",
"source_url": "https://test.example.com/import",
"vendor_id": vendor_id,
@@ -49,6 +51,4 @@ def create_test_marketplace_import_job(db, vendor_id, user_id, **kwargs):
db.add(job)
db.commit()
db.refresh(job)
# Expunge job from session to prevent ResourceWarning about unclosed connections
db.expunge(job)
return job

22
tests/fixtures/marketplace_product_fixtures.py vendored
View File

@@ -1,4 +1,10 @@
# tests/fixtures/marketplace_product_fixtures.py
"""
Marketplace product test fixtures.
Note: Fixtures should NOT use db.expunge() as it breaks lazy loading.
See tests/conftest.py for details on fixture best practices.
"""
import uuid
import pytest
@@ -8,7 +14,7 @@ from models.database.marketplace_product import MarketplaceProduct
@pytest.fixture
def test_marketplace_product(db):
"""Create a test product"""
"""Create a test product."""
marketplace_product = MarketplaceProduct(
marketplace_product_id="TEST001",
title="Test MarketplaceProduct",
@@ -24,14 +30,12 @@ def test_marketplace_product(db):
db.add(marketplace_product)
db.commit()
db.refresh(marketplace_product)
# Expunge marketplace_product from session to prevent ResourceWarning about unclosed connections
db.expunge(marketplace_product)
return marketplace_product
@pytest.fixture
def unique_product(db):
"""Create a unique product for tests that need isolated product data"""
"""Create a unique product for tests that need isolated product data."""
unique_id = str(uuid.uuid4())[:8]
marketplace_product = MarketplaceProduct(
marketplace_product_id=f"UNIQUE_{unique_id}",
@@ -49,14 +53,12 @@ def unique_product(db):
db.add(marketplace_product)
db.commit()
db.refresh(marketplace_product)
# Expunge marketplace_product from session to prevent ResourceWarning about unclosed connections
db.expunge(marketplace_product)
return marketplace_product
@pytest.fixture
def multiple_products(db):
"""Create multiple products for testing statistics and pagination"""
"""Create multiple products for testing statistics and pagination."""
unique_id = str(uuid.uuid4())[:8]
marketplace_products = []
@@ -79,13 +81,11 @@ def multiple_products(db):
db.commit()
for product in marketplace_products:
db.refresh(product)
# Expunge each product from session to prevent ResourceWarning about unclosed connections
db.expunge(product)
return marketplace_products
def create_unique_marketplace_product_factory():
"""Factory function to create unique products in tests"""
"""Factory function to create unique products in tests."""
def _marketplace_create_product(db, **kwargs):
unique_id = str(uuid.uuid4())[:8]
@@ -103,8 +103,6 @@ def create_unique_marketplace_product_factory():
db.add(marketplace_product)
db.commit()
db.refresh(marketplace_product)
# Expunge marketplace_product from session to prevent ResourceWarning about unclosed connections
db.expunge(marketplace_product)
return marketplace_product
return _marketplace_create_product

136
tests/fixtures/vendor_fixtures.py vendored
View File

@@ -1,44 +1,96 @@
# tests/fixtures/vendor_fixtures.py
"""
Vendor-related test fixtures.
Note: Fixtures should NOT use db.expunge() as it breaks lazy loading.
See tests/conftest.py for details on fixture best practices.
"""
import uuid
import pytest
from models.database.company import Company
from models.database.inventory import Inventory
from models.database.product import Product
from models.database.vendor import Vendor
@pytest.fixture
def test_vendor(db, test_user):
"""Create a test vendor with unique vendor code"""
def test_company(db, test_user):
"""Create a test company owned by test_user."""
unique_id = str(uuid.uuid4())[:8]
company = Company(
name=f"Test Company {unique_id}",
owner_user_id=test_user.id,
contact_email=f"test{unique_id}@company.com",
is_active=True,
is_verified=True,
)
db.add(company)
db.commit()
db.refresh(company)
return company
@pytest.fixture
def other_company(db, other_user):
"""Create a test company owned by other_user."""
unique_id = str(uuid.uuid4())[:8]
company = Company(
name=f"Other Company {unique_id}",
owner_user_id=other_user.id,
contact_email=f"other{unique_id}@company.com",
is_active=True,
is_verified=True,
)
db.add(company)
db.commit()
db.refresh(company)
return company
@pytest.fixture
def test_vendor(db, test_company):
"""Create a test vendor with unique vendor code."""
unique_id = str(uuid.uuid4())[:8].upper()
vendor = Vendor(
company_id=test_company.id,
vendor_code=f"TESTVENDOR_{unique_id}",
subdomain=f"testvendor{unique_id.lower()}", # ADDED
name=f"Test Vendor {unique_id.lower()}", # FIXED
owner_user_id=test_user.id,
subdomain=f"testvendor{unique_id.lower()}",
name=f"Test Vendor {unique_id.lower()}",
is_active=True,
is_verified=True,
)
db.add(vendor)
db.commit()
db.refresh(vendor)
# Expunge vendor from session to prevent ResourceWarning about unclosed connections
db.expunge(vendor)
return vendor
@pytest.fixture
def test_vendor_with_vendor_user(db, test_vendor_user):
"""Create a vendor owned by a vendor user (for testing vendor API endpoints)"""
from models.database.vendor import VendorUser
"""Create a vendor owned by a vendor user (for testing vendor API endpoints)."""
from models.database.vendor import VendorUser, VendorUserType
unique_id = str(uuid.uuid4())[:8].upper()
# Create company first
company = Company(
name=f"Vendor API Company {unique_id}",
owner_user_id=test_vendor_user.id,
contact_email=f"vendorapi{unique_id}@company.com",
is_active=True,
is_verified=True,
)
db.add(company)
db.commit()
db.refresh(company)
vendor = Vendor(
company_id=company.id,
vendor_code=f"VENDORAPI_{unique_id}",
subdomain=f"vendorapi{unique_id.lower()}",
name=f"Vendor API Test {unique_id}",
owner_user_id=test_vendor_user.id,
is_active=True,
is_verified=True,
)
@@ -50,81 +102,73 @@ def test_vendor_with_vendor_user(db, test_vendor_user):
vendor_user = VendorUser(
vendor_id=vendor.id,
user_id=test_vendor_user.id,
is_owner=True,
user_type=VendorUserType.OWNER.value,
is_active=True,
)
db.add(vendor_user)
db.commit()
db.refresh(vendor)
db.expunge(vendor)
return vendor
@pytest.fixture
def unique_vendor(db, test_user):
"""Create a unique vendor for tests that need isolated vendor data"""
def unique_vendor(db, test_company):
"""Create a unique vendor for tests that need isolated vendor data."""
unique_id = str(uuid.uuid4())[:8]
vendor = Vendor(
company_id=test_company.id,
vendor_code=f"UNIQUEVENDOR_{unique_id.upper()}",
subdomain=f"uniquevendor{unique_id.lower()}", # ADDED
name=f"Unique Test Vendor {unique_id}", # FIXED
subdomain=f"uniquevendor{unique_id.lower()}",
name=f"Unique Test Vendor {unique_id}",
description=f"A unique test vendor {unique_id}",
owner_user_id=test_user.id,
is_active=True,
is_verified=True,
)
db.add(vendor)
db.commit()
db.refresh(vendor)
# Expunge vendor from session to prevent ResourceWarning about unclosed connections
db.expunge(vendor)
return vendor
@pytest.fixture
def inactive_vendor(db, other_user):
"""Create an inactive vendor owned by other_user"""
def inactive_vendor(db, other_company):
"""Create an inactive vendor owned by other_user."""
unique_id = str(uuid.uuid4())[:8]
vendor = Vendor(
company_id=other_company.id,
vendor_code=f"INACTIVE_{unique_id.upper()}",
subdomain=f"inactive{unique_id.lower()}", # ADDED
name=f"Inactive Vendor {unique_id}", # FIXED
owner_user_id=other_user.id,
subdomain=f"inactive{unique_id.lower()}",
name=f"Inactive Vendor {unique_id}",
is_active=False,
is_verified=False,
)
db.add(vendor)
db.commit()
db.refresh(vendor)
# Expunge vendor from session to prevent ResourceWarning about unclosed connections
db.expunge(vendor)
return vendor
@pytest.fixture
def verified_vendor(db, other_user):
"""Create a verified vendor owned by other_user"""
def verified_vendor(db, other_company):
"""Create a verified vendor owned by other_user."""
unique_id = str(uuid.uuid4())[:8]
vendor = Vendor(
company_id=other_company.id,
vendor_code=f"VERIFIED_{unique_id.upper()}",
subdomain=f"verified{unique_id.lower()}", # ADDED
name=f"Verified Vendor {unique_id}", # FIXED
owner_user_id=other_user.id,
subdomain=f"verified{unique_id.lower()}",
name=f"Verified Vendor {unique_id}",
is_active=True,
is_verified=True,
)
db.add(vendor)
db.commit()
db.refresh(vendor)
# Expunge vendor from session to prevent ResourceWarning about unclosed connections
db.expunge(vendor)
return vendor
@pytest.fixture
def test_product(db, test_vendor, unique_product):
"""Create a vendor product relationship"""
"""Create a vendor product relationship."""
product = Product(
vendor_id=test_vendor.id,
marketplace_product_id=unique_product.id,
@@ -136,8 +180,6 @@ def test_product(db, test_vendor, unique_product):
db.add(product)
db.commit()
db.refresh(product)
# Expunge product from session to prevent ResourceWarning about unclosed connections
db.expunge(product)
return product
@@ -151,19 +193,17 @@ def test_inventory(db, test_product):
location=f"WAREHOUSE_A_{unique_id}",
quantity=100,
reserved_quantity=10,
gtin=test_product.marketplace_product.gtin, # Optional reference
gtin=test_product.marketplace_product.gtin,
)
db.add(inventory)
db.commit()
db.refresh(inventory)
# Expunge inventory from session to prevent ResourceWarning about unclosed connections
db.expunge(inventory)
return inventory
@pytest.fixture
def multiple_inventory_entries(db, multiple_products, test_vendor):
"""Create multiple inventory entries for testing"""
"""Create multiple inventory entries for testing."""
inventory_entries = []
for i, product in enumerate(multiple_products):
inventory = Inventory(
@@ -179,21 +219,19 @@ def multiple_inventory_entries(db, multiple_products, test_vendor):
db.commit()
for inventory in inventory_entries:
db.refresh(inventory)
# Expunge each inventory from session to prevent ResourceWarning about unclosed connections
db.expunge(inventory)
return inventory_entries
def create_unique_vendor_factory():
"""Factory function to create unique vendors in tests"""
"""Factory function to create unique vendors in tests."""
def _create_vendor(db, owner_user_id, **kwargs):
def _create_vendor(db, company_id, **kwargs):
unique_id = str(uuid.uuid4())[:8]
defaults = {
"company_id": company_id,
"vendor_code": f"FACTORY_{unique_id.upper()}",
"subdomain": f"factory{unique_id.lower()}", # ADDED
"name": f"Factory Vendor {unique_id}", # FIXED
"owner_user_id": owner_user_id,
"subdomain": f"factory{unique_id.lower()}",
"name": f"Factory Vendor {unique_id}",
"is_active": True,
"is_verified": False,
}
@@ -210,5 +248,5 @@ def create_unique_vendor_factory():
@pytest.fixture
def vendor_factory():
"""Fixture that provides a vendor factory function"""
"""Fixture that provides a vendor factory function."""
return create_unique_vendor_factory()