Refactoring code for modular approach
This commit is contained in:
313
tests/test_marketplace_service.py
Normal file
313
tests/test_marketplace_service.py
Normal file
@@ -0,0 +1,313 @@
|
||||
# tests/test_marketplace_service.py
|
||||
import pytest
|
||||
from app.services.marketplace_service import MarketplaceService
|
||||
from models.api_models import MarketplaceImportRequest
|
||||
from models.database_models import MarketplaceImportJob, Shop, User
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class TestMarketplaceService:
|
||||
def setup_method(self):
|
||||
self.service = MarketplaceService()
|
||||
|
||||
def test_validate_shop_access_success(self, db, test_shop, test_user):
|
||||
"""Test successful shop access validation"""
|
||||
# Set the shop owner to the test user
|
||||
test_shop.owner_id = test_user.id
|
||||
db.commit()
|
||||
|
||||
result = self.service.validate_shop_access(db, test_shop.shop_code, test_user)
|
||||
|
||||
assert result.shop_code == test_shop.shop_code
|
||||
assert result.owner_id == test_user.id
|
||||
|
||||
def test_validate_shop_access_admin_can_access_any_shop(self, db, test_shop, admin_user):
|
||||
"""Test that admin users can access any shop"""
|
||||
result = self.service.validate_shop_access(db, test_shop.shop_code, admin_user)
|
||||
|
||||
assert result.shop_code == test_shop.shop_code
|
||||
|
||||
def test_validate_shop_access_shop_not_found(self, db, test_user):
|
||||
"""Test shop access validation when shop doesn't exist"""
|
||||
with pytest.raises(ValueError, match="Shop not found"):
|
||||
self.service.validate_shop_access(db, "NONEXISTENT", test_user)
|
||||
|
||||
def test_validate_shop_access_permission_denied(self, db, test_shop, test_user, other_user):
|
||||
"""Test shop access validation when user doesn't own the shop"""
|
||||
# Set the shop owner to a different user
|
||||
test_shop.owner_id = other_user.id
|
||||
db.commit()
|
||||
|
||||
with pytest.raises(PermissionError, match="Access denied to this shop"):
|
||||
self.service.validate_shop_access(db, test_shop.shop_code, test_user)
|
||||
|
||||
def test_create_import_job_success(self, db, test_shop, test_user):
|
||||
"""Test successful creation of import job"""
|
||||
# Set the shop owner to the test user
|
||||
test_shop.owner_id = test_user.id
|
||||
db.commit()
|
||||
|
||||
request = MarketplaceImportRequest(
|
||||
url="https://example.com/products.csv",
|
||||
marketplace="Amazon",
|
||||
shop_code=test_shop.shop_code,
|
||||
batch_size=1000
|
||||
)
|
||||
|
||||
result = self.service.create_import_job(db, request, test_user)
|
||||
|
||||
assert result.marketplace == "Amazon"
|
||||
assert result.shop_code == test_shop.shop_code
|
||||
assert result.user_id == test_user.id
|
||||
assert result.status == "pending"
|
||||
assert result.source_url == "https://example.com/products.csv"
|
||||
|
||||
def test_create_import_job_invalid_shop(self, db, test_user):
|
||||
"""Test import job creation with invalid shop"""
|
||||
request = MarketplaceImportRequest(
|
||||
url="https://example.com/products.csv",
|
||||
marketplace="Amazon",
|
||||
shop_code="INVALID_SHOP",
|
||||
batch_size=1000
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="Shop not found"):
|
||||
self.service.create_import_job(db, request, test_user)
|
||||
|
||||
def test_get_import_job_by_id_success(self, db, test_import_job, test_user):
|
||||
"""Test getting import job by ID for job owner"""
|
||||
result = self.service.get_import_job_by_id(db, test_import_job.id, test_user)
|
||||
|
||||
assert result.id == test_import_job.id
|
||||
assert result.user_id == test_user.id
|
||||
|
||||
def test_get_import_job_by_id_admin_access(self, db, test_import_job, admin_user):
|
||||
"""Test that admin can access any import job"""
|
||||
result = self.service.get_import_job_by_id(db, test_import_job.id, admin_user)
|
||||
|
||||
assert result.id == test_import_job.id
|
||||
|
||||
def test_get_import_job_by_id_not_found(self, db, test_user):
|
||||
"""Test getting non-existent import job"""
|
||||
with pytest.raises(ValueError, match="Marketplace import job not found"):
|
||||
self.service.get_import_job_by_id(db, 99999, test_user)
|
||||
|
||||
def test_get_import_job_by_id_access_denied(self, db, test_import_job, other_user):
|
||||
"""Test access denied when user doesn't own the job"""
|
||||
with pytest.raises(PermissionError, match="Access denied to this import job"):
|
||||
self.service.get_import_job_by_id(db, test_import_job.id, other_user)
|
||||
|
||||
def test_get_import_jobs_user_filter(self, db, test_import_job, test_user):
|
||||
"""Test getting import jobs filtered by user"""
|
||||
jobs = self.service.get_import_jobs(db, test_user)
|
||||
|
||||
assert len(jobs) == 1
|
||||
assert jobs[0].id == test_import_job.id
|
||||
assert jobs[0].user_id == test_user.id
|
||||
|
||||
def test_get_import_jobs_admin_sees_all(self, db, test_import_job, admin_user):
|
||||
"""Test that admin sees all import jobs"""
|
||||
jobs = self.service.get_import_jobs(db, admin_user)
|
||||
|
||||
assert len(jobs) >= 1
|
||||
assert any(job.id == test_import_job.id for job in jobs)
|
||||
|
||||
def test_get_import_jobs_with_marketplace_filter(self, db, test_import_job, test_user):
|
||||
"""Test getting import jobs with marketplace filter"""
|
||||
jobs = self.service.get_import_jobs(
|
||||
db, test_user, marketplace=test_import_job.marketplace
|
||||
)
|
||||
|
||||
assert len(jobs) == 1
|
||||
assert jobs[0].marketplace == test_import_job.marketplace
|
||||
|
||||
def test_get_import_jobs_with_pagination(self, db, test_user):
|
||||
"""Test getting import jobs with pagination"""
|
||||
# Create multiple import jobs
|
||||
for i in range(5):
|
||||
job = MarketplaceImportJob(
|
||||
status="completed",
|
||||
marketplace=f"Marketplace_{i}",
|
||||
shop_code="TEST_SHOP",
|
||||
user_id=test_user.id,
|
||||
created_at=datetime.utcnow()
|
||||
)
|
||||
db.add(job)
|
||||
db.commit()
|
||||
|
||||
jobs = self.service.get_import_jobs(db, test_user, skip=2, limit=2)
|
||||
|
||||
assert len(jobs) == 2
|
||||
|
||||
def test_update_job_status_success(self, db, test_import_job):
|
||||
"""Test updating job status"""
|
||||
result = self.service.update_job_status(
|
||||
db,
|
||||
test_import_job.id,
|
||||
"completed",
|
||||
imported_count=100,
|
||||
total_processed=100
|
||||
)
|
||||
|
||||
assert result.status == "completed"
|
||||
assert result.imported_count == 100
|
||||
assert result.total_processed == 100
|
||||
|
||||
def test_update_job_status_not_found(self, db):
|
||||
"""Test updating non-existent job status"""
|
||||
with pytest.raises(ValueError, match="Marketplace import job not found"):
|
||||
self.service.update_job_status(db, 99999, "completed")
|
||||
|
||||
def test_get_job_stats_user(self, db, test_import_job, test_user):
|
||||
"""Test getting job statistics for user"""
|
||||
stats = self.service.get_job_stats(db, test_user)
|
||||
|
||||
assert stats["total_jobs"] >= 1
|
||||
assert "pending_jobs" in stats
|
||||
assert "running_jobs" in stats
|
||||
assert "completed_jobs" in stats
|
||||
assert "failed_jobs" in stats
|
||||
|
||||
def test_get_job_stats_admin(self, db, test_import_job, admin_user):
|
||||
"""Test getting job statistics for admin"""
|
||||
stats = self.service.get_job_stats(db, admin_user)
|
||||
|
||||
assert stats["total_jobs"] >= 1
|
||||
|
||||
def test_convert_to_response_model(self, test_import_job):
|
||||
"""Test converting database model to response model"""
|
||||
response = self.service.convert_to_response_model(test_import_job)
|
||||
|
||||
assert response.job_id == test_import_job.id
|
||||
assert response.status == test_import_job.status
|
||||
assert response.marketplace == test_import_job.marketplace
|
||||
assert response.imported == (test_import_job.imported_count or 0)
|
||||
|
||||
def test_cancel_import_job_success(self, db, test_user):
|
||||
"""Test cancelling a pending import job"""
|
||||
# Create a pending job
|
||||
job = MarketplaceImportJob(
|
||||
status="pending",
|
||||
marketplace="Amazon",
|
||||
shop_code="TEST_SHOP",
|
||||
user_id=test_user.id,
|
||||
created_at=datetime.utcnow()
|
||||
)
|
||||
db.add(job)
|
||||
db.commit()
|
||||
db.refresh(job)
|
||||
|
||||
result = self.service.cancel_import_job(db, job.id, test_user)
|
||||
|
||||
assert result.status == "cancelled"
|
||||
assert result.completed_at is not None
|
||||
|
||||
def test_cancel_import_job_invalid_status(self, db, test_import_job, test_user):
|
||||
"""Test cancelling a job that can't be cancelled"""
|
||||
# Set job status to completed
|
||||
test_import_job.status = "completed"
|
||||
db.commit()
|
||||
|
||||
with pytest.raises(ValueError, match="Cannot cancel job with status: completed"):
|
||||
self.service.cancel_import_job(db, test_import_job.id, test_user)
|
||||
|
||||
def test_delete_import_job_success(self, db, test_user):
|
||||
"""Test deleting a completed import job"""
|
||||
# Create a completed job
|
||||
job = MarketplaceImportJob(
|
||||
status="completed",
|
||||
marketplace="Amazon",
|
||||
shop_code="TEST_SHOP",
|
||||
user_id=test_user.id,
|
||||
created_at=datetime.utcnow()
|
||||
)
|
||||
db.add(job)
|
||||
db.commit()
|
||||
db.refresh(job)
|
||||
job_id = job.id
|
||||
|
||||
result = self.service.delete_import_job(db, job_id, test_user)
|
||||
|
||||
assert result is True
|
||||
|
||||
# Verify the job is actually deleted
|
||||
deleted_job = db.query(MarketplaceImportJob).filter(MarketplaceImportJob.id == job_id).first()
|
||||
assert deleted_job is None
|
||||
|
||||
def test_delete_import_job_invalid_status(self, db, test_user):
|
||||
"""Test deleting a job that can't be deleted"""
|
||||
# Create a pending job
|
||||
job = MarketplaceImportJob(
|
||||
status="pending",
|
||||
marketplace="Amazon",
|
||||
shop_code="TEST_SHOP",
|
||||
user_id=test_user.id,
|
||||
created_at=datetime.utcnow()
|
||||
)
|
||||
db.add(job)
|
||||
db.commit()
|
||||
db.refresh(job)
|
||||
|
||||
with pytest.raises(ValueError, match="Cannot delete job with status: pending"):
|
||||
self.service.delete_import_job(db, job.id, test_user)
|
||||
|
||||
|
||||
# Additional fixtures for marketplace tests
|
||||
@pytest.fixture
|
||||
def test_shop(db):
|
||||
"""Create a test shop"""
|
||||
shop = Shop(
|
||||
shop_code="TEST_SHOP",
|
||||
shop_name="Test Shop",
|
||||
owner_id=1 # Will be updated in tests
|
||||
)
|
||||
db.add(shop)
|
||||
db.commit()
|
||||
db.refresh(shop)
|
||||
return shop
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_user(db):
|
||||
"""Create a test admin user"""
|
||||
user = User(
|
||||
username="admin_user",
|
||||
email="admin@test.com",
|
||||
role="admin",
|
||||
hashed_password="hashed_password"
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def other_user(db):
|
||||
"""Create another test user"""
|
||||
user = User(
|
||||
username="other_user",
|
||||
email="other@test.com",
|
||||
role="user",
|
||||
hashed_password="hashed_password"
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_import_job(db, test_user):
|
||||
"""Create a test import job"""
|
||||
job = MarketplaceImportJob(
|
||||
status="pending",
|
||||
marketplace="Amazon",
|
||||
shop_code="TEST_SHOP",
|
||||
user_id=test_user.id,
|
||||
created_at=datetime.utcnow()
|
||||
)
|
||||
db.add(job)
|
||||
db.commit()
|
||||
db.refresh(job)
|
||||
return job
|
||||
@@ -1,4 +1,4 @@
|
||||
# tests/test_services.py
|
||||
# tests/test_product_service.py
|
||||
import pytest
|
||||
from app.services.product_service import ProductService
|
||||
from models.api_models import ProductCreate
|
||||
322
tests/test_stock_service.py
Normal file
322
tests/test_stock_service.py
Normal file
@@ -0,0 +1,322 @@
|
||||
# tests/test_stock_service.py
|
||||
import pytest
|
||||
from app.services.stock_service import StockService
|
||||
from models.api_models import StockCreate, StockAdd, StockUpdate
|
||||
from models.database_models import Stock, Product
|
||||
|
||||
|
||||
class TestStockService:
|
||||
def setup_method(self):
|
||||
self.service = StockService()
|
||||
|
||||
def test_normalize_gtin_valid(self):
|
||||
"""Test GTIN normalization with valid GTINs"""
|
||||
# Test various valid GTIN formats
|
||||
assert self.service.normalize_gtin("1234567890123") == "1234567890123"
|
||||
assert self.service.normalize_gtin("123456789012") == "123456789012"
|
||||
assert self.service.normalize_gtin("12345678") == "12345678"
|
||||
|
||||
def test_normalize_gtin_invalid(self):
|
||||
"""Test GTIN normalization with invalid GTINs"""
|
||||
assert self.service.normalize_gtin("invalid") is None
|
||||
assert self.service.normalize_gtin("123") is None
|
||||
assert self.service.normalize_gtin("") is None
|
||||
assert self.service.normalize_gtin(None) is None
|
||||
|
||||
def test_set_stock_new_entry(self, db):
|
||||
"""Test setting stock for a new GTIN/location combination"""
|
||||
stock_data = StockCreate(
|
||||
gtin="1234567890123",
|
||||
location="WAREHOUSE_A",
|
||||
quantity=100
|
||||
)
|
||||
|
||||
result = self.service.set_stock(db, stock_data)
|
||||
|
||||
assert result.gtin == "1234567890123"
|
||||
assert result.location == "WAREHOUSE_A"
|
||||
assert result.quantity == 100
|
||||
|
||||
def test_set_stock_existing_entry(self, db, test_stock):
|
||||
"""Test setting stock for an existing GTIN/location combination"""
|
||||
stock_data = StockCreate(
|
||||
gtin=test_stock.gtin,
|
||||
location=test_stock.location,
|
||||
quantity=200
|
||||
)
|
||||
|
||||
result = self.service.set_stock(db, stock_data)
|
||||
|
||||
assert result.gtin == test_stock.gtin
|
||||
assert result.location == test_stock.location
|
||||
assert result.quantity == 200 # Should replace the original quantity
|
||||
|
||||
def test_set_stock_invalid_gtin(self, db):
|
||||
"""Test setting stock with invalid GTIN"""
|
||||
stock_data = StockCreate(
|
||||
gtin="invalid_gtin",
|
||||
location="WAREHOUSE_A",
|
||||
quantity=100
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid GTIN format"):
|
||||
self.service.set_stock(db, stock_data)
|
||||
|
||||
def test_add_stock_new_entry(self, db):
|
||||
"""Test adding stock for a new GTIN/location combination"""
|
||||
stock_data = StockAdd(
|
||||
gtin="1234567890123",
|
||||
location="WAREHOUSE_B",
|
||||
quantity=50
|
||||
)
|
||||
|
||||
result = self.service.add_stock(db, stock_data)
|
||||
|
||||
assert result.gtin == "1234567890123"
|
||||
assert result.location == "WAREHOUSE_B"
|
||||
assert result.quantity == 50
|
||||
|
||||
def test_add_stock_existing_entry(self, db, test_stock):
|
||||
"""Test adding stock to an existing GTIN/location combination"""
|
||||
original_quantity = test_stock.quantity
|
||||
stock_data = StockAdd(
|
||||
gtin=test_stock.gtin,
|
||||
location=test_stock.location,
|
||||
quantity=25
|
||||
)
|
||||
|
||||
result = self.service.add_stock(db, stock_data)
|
||||
|
||||
assert result.gtin == test_stock.gtin
|
||||
assert result.location == test_stock.location
|
||||
assert result.quantity == original_quantity + 25
|
||||
|
||||
def test_add_stock_invalid_gtin(self, db):
|
||||
"""Test adding stock with invalid GTIN"""
|
||||
stock_data = StockAdd(
|
||||
gtin="invalid_gtin",
|
||||
location="WAREHOUSE_A",
|
||||
quantity=50
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid GTIN format"):
|
||||
self.service.add_stock(db, stock_data)
|
||||
|
||||
def test_remove_stock_success(self, db, test_stock):
|
||||
"""Test removing stock successfully"""
|
||||
original_quantity = test_stock.quantity
|
||||
remove_quantity = 10
|
||||
|
||||
stock_data = StockAdd(
|
||||
gtin=test_stock.gtin,
|
||||
location=test_stock.location,
|
||||
quantity=remove_quantity
|
||||
)
|
||||
|
||||
result = self.service.remove_stock(db, stock_data)
|
||||
|
||||
assert result.gtin == test_stock.gtin
|
||||
assert result.location == test_stock.location
|
||||
assert result.quantity == original_quantity - remove_quantity
|
||||
|
||||
def test_remove_stock_insufficient_stock(self, db, test_stock):
|
||||
"""Test removing more stock than available"""
|
||||
stock_data = StockAdd(
|
||||
gtin=test_stock.gtin,
|
||||
location=test_stock.location,
|
||||
quantity=test_stock.quantity + 10 # More than available
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="Insufficient stock"):
|
||||
self.service.remove_stock(db, stock_data)
|
||||
|
||||
def test_remove_stock_nonexistent_entry(self, db):
|
||||
"""Test removing stock from non-existent GTIN/location"""
|
||||
stock_data = StockAdd(
|
||||
gtin="9999999999999",
|
||||
location="NONEXISTENT",
|
||||
quantity=10
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="No stock found"):
|
||||
self.service.remove_stock(db, stock_data)
|
||||
|
||||
def test_remove_stock_invalid_gtin(self, db):
|
||||
"""Test removing stock with invalid GTIN"""
|
||||
stock_data = StockAdd(
|
||||
gtin="invalid_gtin",
|
||||
location="WAREHOUSE_A",
|
||||
quantity=10
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid GTIN format"):
|
||||
self.service.remove_stock(db, stock_data)
|
||||
|
||||
def test_get_stock_by_gtin_success(self, db, test_stock, test_product):
|
||||
"""Test getting stock summary by GTIN"""
|
||||
result = self.service.get_stock_by_gtin(db, test_stock.gtin)
|
||||
|
||||
assert result.gtin == test_stock.gtin
|
||||
assert result.total_quantity == test_stock.quantity
|
||||
assert len(result.locations) == 1
|
||||
assert result.locations[0].location == test_stock.location
|
||||
assert result.locations[0].quantity == test_stock.quantity
|
||||
assert result.product_title == test_product.title
|
||||
|
||||
def test_get_stock_by_gtin_multiple_locations(self, db):
|
||||
"""Test getting stock summary with multiple locations"""
|
||||
gtin = "1234567890123"
|
||||
|
||||
# Create multiple stock entries for the same GTIN
|
||||
stock1 = Stock(gtin=gtin, location="WAREHOUSE_A", quantity=50)
|
||||
stock2 = Stock(gtin=gtin, location="WAREHOUSE_B", quantity=30)
|
||||
|
||||
db.add(stock1)
|
||||
db.add(stock2)
|
||||
db.commit()
|
||||
|
||||
result = self.service.get_stock_by_gtin(db, gtin)
|
||||
|
||||
assert result.gtin == gtin
|
||||
assert result.total_quantity == 80
|
||||
assert len(result.locations) == 2
|
||||
|
||||
def test_get_stock_by_gtin_not_found(self, db):
|
||||
"""Test getting stock for non-existent GTIN"""
|
||||
with pytest.raises(ValueError, match="No stock found"):
|
||||
self.service.get_stock_by_gtin(db, "9999999999999")
|
||||
|
||||
def test_get_stock_by_gtin_invalid_gtin(self, db):
|
||||
"""Test getting stock with invalid GTIN"""
|
||||
with pytest.raises(ValueError, match="Invalid GTIN format"):
|
||||
self.service.get_stock_by_gtin(db, "invalid_gtin")
|
||||
|
||||
def test_get_total_stock_success(self, db, test_stock, test_product):
|
||||
"""Test getting total stock for a GTIN"""
|
||||
result = self.service.get_total_stock(db, test_stock.gtin)
|
||||
|
||||
assert result["gtin"] == test_stock.gtin
|
||||
assert result["total_quantity"] == test_stock.quantity
|
||||
assert result["product_title"] == test_product.title
|
||||
assert result["locations_count"] == 1
|
||||
|
||||
def test_get_total_stock_invalid_gtin(self, db):
|
||||
"""Test getting total stock with invalid GTIN"""
|
||||
with pytest.raises(ValueError, match="Invalid GTIN format"):
|
||||
self.service.get_total_stock(db, "invalid_gtin")
|
||||
|
||||
def test_get_all_stock_no_filters(self, db, test_stock):
|
||||
"""Test getting all stock without filters"""
|
||||
result = self.service.get_all_stock(db)
|
||||
|
||||
assert len(result) >= 1
|
||||
assert any(stock.gtin == test_stock.gtin for stock in result)
|
||||
|
||||
def test_get_all_stock_with_location_filter(self, db, test_stock):
|
||||
"""Test getting all stock with location filter"""
|
||||
result = self.service.get_all_stock(db, location=test_stock.location)
|
||||
|
||||
assert len(result) >= 1
|
||||
assert all(stock.location.upper() == test_stock.location.upper() for stock in result)
|
||||
|
||||
def test_get_all_stock_with_gtin_filter(self, db, test_stock):
|
||||
"""Test getting all stock with GTIN filter"""
|
||||
result = self.service.get_all_stock(db, gtin=test_stock.gtin)
|
||||
|
||||
assert len(result) >= 1
|
||||
assert all(stock.gtin == test_stock.gtin for stock in result)
|
||||
|
||||
def test_get_all_stock_with_pagination(self, db):
|
||||
"""Test getting all stock with pagination"""
|
||||
# Create multiple stock entries
|
||||
for i in range(5):
|
||||
stock = Stock(
|
||||
gtin=f"123456789012{i}",
|
||||
location=f"WAREHOUSE_{i}",
|
||||
quantity=10
|
||||
)
|
||||
db.add(stock)
|
||||
db.commit()
|
||||
|
||||
result = self.service.get_all_stock(db, skip=2, limit=2)
|
||||
|
||||
assert len(result) == 2
|
||||
|
||||
def test_update_stock_success(self, db, test_stock):
|
||||
"""Test updating stock quantity"""
|
||||
stock_update = StockUpdate(quantity=150)
|
||||
|
||||
result = self.service.update_stock(db, test_stock.id, stock_update)
|
||||
|
||||
assert result.id == test_stock.id
|
||||
assert result.quantity == 150
|
||||
|
||||
def test_update_stock_not_found(self, db):
|
||||
"""Test updating non-existent stock entry"""
|
||||
stock_update = StockUpdate(quantity=150)
|
||||
|
||||
with pytest.raises(ValueError, match="Stock entry not found"):
|
||||
self.service.update_stock(db, 99999, stock_update)
|
||||
|
||||
def test_delete_stock_success(self, db, test_stock):
|
||||
"""Test deleting stock entry"""
|
||||
stock_id = test_stock.id
|
||||
|
||||
result = self.service.delete_stock(db, stock_id)
|
||||
|
||||
assert result is True
|
||||
|
||||
# Verify the stock is actually deleted
|
||||
deleted_stock = db.query(Stock).filter(Stock.id == stock_id).first()
|
||||
assert deleted_stock is None
|
||||
|
||||
def test_delete_stock_not_found(self, db):
|
||||
"""Test deleting non-existent stock entry"""
|
||||
with pytest.raises(ValueError, match="Stock entry not found"):
|
||||
self.service.delete_stock(db, 99999)
|
||||
|
||||
def test_get_stock_by_id_success(self, db, test_stock):
|
||||
"""Test getting stock entry by ID"""
|
||||
result = self.service.get_stock_by_id(db, test_stock.id)
|
||||
|
||||
assert result is not None
|
||||
assert result.id == test_stock.id
|
||||
assert result.gtin == test_stock.gtin
|
||||
|
||||
def test_get_stock_by_id_not_found(self, db):
|
||||
"""Test getting non-existent stock entry by ID"""
|
||||
result = self.service.get_stock_by_id(db, 99999)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
# Additional fixtures that might be needed for stock tests
|
||||
@pytest.fixture
|
||||
def test_stock(db):
|
||||
"""Create a test stock entry"""
|
||||
stock = Stock(
|
||||
gtin="1234567890123",
|
||||
location="WAREHOUSE_MAIN",
|
||||
quantity=50
|
||||
)
|
||||
db.add(stock)
|
||||
db.commit()
|
||||
db.refresh(stock)
|
||||
return stock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_product_with_stock(db, test_stock):
|
||||
"""Create a test product that corresponds to the test stock"""
|
||||
product = Product(
|
||||
product_id="STOCK_TEST_001",
|
||||
title="Stock Test Product",
|
||||
gtin=test_stock.gtin,
|
||||
price="29.99",
|
||||
brand="TestBrand",
|
||||
marketplace="Letzshop"
|
||||
)
|
||||
db.add(product)
|
||||
db.commit()
|
||||
db.refresh(product)
|
||||
return product
|
||||
Reference in New Issue
Block a user