# tests/unit/services/test_stats_service.py """Unit tests for StatsService following the application's testing patterns.""" import uuid from unittest.mock import patch import pytest from sqlalchemy.exc import SQLAlchemyError from app.modules.analytics.services.stats_service import StatsService from app.modules.inventory.models import Inventory from app.modules.marketplace.models import ( MarketplaceProduct, MarketplaceProductTranslation, ) from app.modules.tenancy.exceptions import ( AdminOperationException, StoreNotFoundException, ) def create_marketplace_product_with_translation( db, marketplace_product_id, title, language="en", **kwargs ): """Helper to create a MarketplaceProduct with its translation.""" product = MarketplaceProduct( marketplace_product_id=marketplace_product_id, **kwargs ) db.add(product) db.flush() # Get the product ID translation = MarketplaceProductTranslation( marketplace_product_id=product.id, language=language, title=title, ) db.add(translation) return product @pytest.mark.unit @pytest.mark.stats class TestStatsService: """Test suite for StatsService following the application's testing patterns.""" def setup_method(self): """Setup method following the same pattern as other service tests.""" self.service = StatsService() # ==================== get_comprehensive_stats Tests ==================== def test_get_comprehensive_stats_basic(self, db, test_marketplace_product): """Test getting comprehensive stats with basic data.""" stats = self.service.get_comprehensive_stats(db) assert "total_products" in stats assert "unique_brands" in stats assert "unique_categories" in stats assert "unique_marketplaces" in stats assert "unique_stores" in stats assert "total_inventory_entries" in stats assert "total_inventory_quantity" in stats # Verify types assert isinstance(stats["total_products"], int) assert isinstance(stats["unique_brands"], int) assert isinstance(stats["unique_categories"], int) assert isinstance(stats["unique_marketplaces"], int) assert isinstance(stats["unique_stores"], int) assert isinstance(stats["total_inventory_entries"], int) assert isinstance(stats["total_inventory_quantity"], int) def test_get_comprehensive_stats_with_products(self, db, test_product): """Test comprehensive stats counts Product records correctly.""" stats = self.service.get_comprehensive_stats(db) assert stats["total_products"] >= 1 def test_get_comprehensive_stats_multiple_marketplaces( self, db, test_marketplace_product ): """Test comprehensive stats with multiple marketplaces.""" unique_id = str(uuid.uuid4())[:8] create_marketplace_product_with_translation( db, marketplace_product_id=f"PROD002_{unique_id}", title="MarketplaceProduct 2", brand="DifferentBrand", google_product_category="Different Category", marketplace="Amazon", store_name="AmazonStore", price="15.99", currency="EUR", ) create_marketplace_product_with_translation( db, marketplace_product_id=f"PROD003_{unique_id}", title="MarketplaceProduct 3", brand="ThirdBrand", google_product_category="Third Category", marketplace="eBay", store_name="eBayStore", price="25.99", currency="USD", ) db.commit() stats = self.service.get_comprehensive_stats(db) # Should count unique marketplaces from MarketplaceProduct table assert stats["unique_marketplaces"] >= 3 # Letzshop, Amazon, eBay def test_get_comprehensive_stats_handles_nulls(self, db): """Test comprehensive stats handles null/empty values correctly.""" unique_id = str(uuid.uuid4())[:8] create_marketplace_product_with_translation( db, marketplace_product_id=f"NULL001_{unique_id}", title="MarketplaceProduct with Nulls", brand=None, google_product_category=None, marketplace=None, store_name=None, price="10.00", currency="EUR", ) create_marketplace_product_with_translation( db, marketplace_product_id=f"EMPTY001_{unique_id}", title="MarketplaceProduct with Empty Values", brand="", google_product_category="", marketplace="", store_name="", price="15.00", currency="EUR", ) db.commit() stats = self.service.get_comprehensive_stats(db) # Should not throw error - null/empty values handled gracefully assert isinstance(stats["unique_brands"], int) assert isinstance(stats["unique_categories"], int) assert isinstance(stats["unique_marketplaces"], int) def test_get_comprehensive_stats_empty_database(self, db): """Test stats with empty database.""" stats = self.service.get_comprehensive_stats(db) assert stats["total_products"] == 0 assert stats["unique_brands"] == 0 assert stats["unique_categories"] == 0 assert stats["unique_marketplaces"] == 0 assert stats["total_inventory_entries"] == 0 assert stats["total_inventory_quantity"] == 0 def test_get_comprehensive_stats_database_error(self, db): """Test comprehensive stats handles database errors.""" with patch.object(db, "query", side_effect=SQLAlchemyError("DB Error")): with pytest.raises(AdminOperationException) as exc_info: self.service.get_comprehensive_stats(db) assert exc_info.value.details.get("operation") == "get_comprehensive_stats" # ==================== get_marketplace_breakdown_stats Tests ==================== def test_get_marketplace_breakdown_stats_basic(self, db, test_marketplace_product): """Test getting marketplace breakdown stats with basic data.""" stats = self.service.get_marketplace_breakdown_stats(db) assert isinstance(stats, list) assert len(stats) >= 1 # Find our test marketplace in the results test_marketplace_stat = next( ( stat for stat in stats if stat["marketplace"] == test_marketplace_product.marketplace ), None, ) assert test_marketplace_stat is not None assert test_marketplace_stat["total_products"] >= 1 assert "unique_stores" in test_marketplace_stat assert "unique_brands" in test_marketplace_stat def test_get_marketplace_breakdown_stats_multiple_marketplaces( self, db, test_marketplace_product ): """Test marketplace breakdown with multiple marketplaces.""" unique_id = str(uuid.uuid4())[:8] create_marketplace_product_with_translation( db, marketplace_product_id=f"AMAZON001_{unique_id}", title="Amazon MarketplaceProduct 1", brand="AmazonBrand1", marketplace="Amazon", store_name="AmazonStore1", price="20.00", currency="EUR", ) create_marketplace_product_with_translation( db, marketplace_product_id=f"AMAZON002_{unique_id}", title="Amazon MarketplaceProduct 2", brand="AmazonBrand2", marketplace="Amazon", store_name="AmazonStore2", price="25.00", currency="EUR", ) create_marketplace_product_with_translation( db, marketplace_product_id=f"EBAY001_{unique_id}", title="eBay MarketplaceProduct", brand="eBayBrand", marketplace="eBay", store_name="eBayStore", price="30.00", currency="USD", ) db.commit() stats = self.service.get_marketplace_breakdown_stats(db) marketplace_names = [stat["marketplace"] for stat in stats] assert "Amazon" in marketplace_names assert "eBay" in marketplace_names # Check Amazon stats amazon_stat = next(stat for stat in stats if stat["marketplace"] == "Amazon") assert amazon_stat["total_products"] == 2 assert amazon_stat["unique_stores"] == 2 assert amazon_stat["unique_brands"] == 2 # Check eBay stats ebay_stat = next(stat for stat in stats if stat["marketplace"] == "eBay") assert ebay_stat["total_products"] == 1 assert ebay_stat["unique_stores"] == 1 assert ebay_stat["unique_brands"] == 1 def test_get_marketplace_breakdown_stats_excludes_nulls(self, db): """Test marketplace breakdown excludes products with null marketplaces.""" unique_id = str(uuid.uuid4())[:8] create_marketplace_product_with_translation( db, marketplace_product_id=f"NULLMARKET001_{unique_id}", title="MarketplaceProduct without marketplace", marketplace=None, store_name="SomeStore", brand="SomeBrand", price="10.00", currency="EUR", ) db.commit() stats = self.service.get_marketplace_breakdown_stats(db) # Should not include any stats for null marketplace marketplace_names = [ stat["marketplace"] for stat in stats if stat["marketplace"] is not None ] assert None not in marketplace_names def test_get_marketplace_breakdown_empty_database(self, db): """Test marketplace breakdown with empty database.""" stats = self.service.get_marketplace_breakdown_stats(db) assert isinstance(stats, list) assert len(stats) == 0 def test_get_marketplace_breakdown_stats_database_error(self, db): """Test marketplace breakdown handles database errors.""" with patch.object(db, "query", side_effect=SQLAlchemyError("DB Error")): with pytest.raises(AdminOperationException) as exc_info: self.service.get_marketplace_breakdown_stats(db) assert ( exc_info.value.details.get("operation") == "get_marketplace_breakdown_stats" ) # ==================== get_store_stats Tests ==================== def test_get_store_stats_success(self, db, test_store, test_product): """Test getting store statistics successfully.""" stats = self.service.get_store_stats(db, test_store.id) # New flat structure compatible with StoreDashboardStatsResponse assert "total_products" in stats assert "active_products" in stats assert "total_orders" in stats assert "total_customers" in stats assert "total_revenue" in stats assert "total_inventory_quantity" in stats assert stats["total_products"] >= 0 assert stats["total_inventory_quantity"] >= 0 def test_get_store_stats_store_not_found(self, db): """Test store stats with non-existent store.""" with pytest.raises(StoreNotFoundException): self.service.get_store_stats(db, 99999) def test_get_store_stats_with_inventory( self, db, test_store, test_product, test_inventory ): """Test store stats includes inventory data.""" stats = self.service.get_store_stats(db, test_store.id) assert stats["total_inventory_quantity"] >= test_inventory.quantity assert stats["reserved_inventory_quantity"] >= 0 def test_get_store_stats_database_error(self, db, test_store): """Test store stats handles database errors after store check.""" # Mock query to fail after first successful call (store check) original_query = db.query call_count = [0] def mock_query(*args, **kwargs): call_count[0] += 1 if call_count[0] > 1: raise SQLAlchemyError("DB Error") return original_query(*args, **kwargs) with patch.object(db, "query", side_effect=mock_query): with pytest.raises(AdminOperationException) as exc_info: self.service.get_store_stats(db, test_store.id) assert exc_info.value.details.get("operation") == "get_store_stats" # ==================== get_store_analytics Tests ==================== def test_get_store_analytics_success(self, db, test_store): """Test getting store analytics successfully.""" analytics = self.service.get_store_analytics(db, test_store.id) assert "period" in analytics assert "start_date" in analytics assert "imports" in analytics assert "catalog" in analytics assert "inventory" in analytics def test_get_store_analytics_different_periods(self, db, test_store): """Test store analytics with different time periods.""" for period in ["7d", "30d", "90d", "1y"]: analytics = self.service.get_store_analytics( db, test_store.id, period=period ) assert analytics["period"] == period def test_get_store_analytics_store_not_found(self, db): """Test store analytics with non-existent store.""" with pytest.raises(StoreNotFoundException): self.service.get_store_analytics(db, 99999) # ==================== get_store_statistics Tests ==================== def test_get_store_statistics_success(self, db, test_store): """Test getting store statistics for admin dashboard.""" stats = self.service.get_store_statistics(db) assert "total" in stats assert "verified" in stats assert "pending" in stats assert "inactive" in stats assert "verification_rate" in stats assert stats["total"] >= 1 def test_get_store_statistics_calculates_rates(self, db, test_store): """Test store statistics calculates rates correctly.""" stats = self.service.get_store_statistics(db) if stats["total"] > 0: expected_rate = stats["verified"] / stats["total"] * 100 assert abs(stats["verification_rate"] - expected_rate) < 0.01 def test_get_store_statistics_database_error(self, db): """Test store statistics handles database errors.""" with patch.object(db, "query", side_effect=SQLAlchemyError("DB Error")): with pytest.raises(AdminOperationException) as exc_info: self.service.get_store_statistics(db) assert exc_info.value.details.get("operation") == "get_store_statistics" # ==================== get_user_statistics Tests ==================== def test_get_user_statistics_success(self, db, test_user): """Test getting user statistics.""" stats = self.service.get_user_statistics(db) assert "total_users" in stats assert "active_users" in stats assert "inactive_users" in stats assert "admin_users" in stats assert "activation_rate" in stats assert stats["total_users"] >= 1 def test_get_user_statistics_calculates_correctly(self, db, test_user, test_admin): """Test user statistics calculates values correctly.""" stats = self.service.get_user_statistics(db) assert stats["total_users"] == stats["active_users"] + stats["inactive_users"] assert stats["admin_users"] >= 1 # test_admin def test_get_user_statistics_database_error(self, db): """Test user statistics handles database errors.""" with patch.object(db, "query", side_effect=SQLAlchemyError("DB Error")): with pytest.raises(AdminOperationException) as exc_info: self.service.get_user_statistics(db) assert exc_info.value.details.get("operation") == "get_user_statistics" # ==================== get_import_statistics Tests ==================== def test_get_import_statistics_success(self, db): """Test getting import statistics.""" stats = self.service.get_import_statistics(db) assert "total" in stats assert "completed" in stats assert "failed" in stats assert "success_rate" in stats def test_get_import_statistics_with_jobs( self, db, test_store, test_marketplace_import_job ): """Test import statistics with existing jobs.""" stats = self.service.get_import_statistics(db) assert stats["total"] >= 1 assert stats["completed"] >= 1 # test job has completed status def test_get_import_statistics_calculates_rate(self, db): """Test import statistics calculates success rate.""" stats = self.service.get_import_statistics(db) if stats["total"] > 0: expected_rate = stats["completed"] / stats["total"] * 100 assert abs(stats["success_rate"] - expected_rate) < 0.01 else: assert stats["success_rate"] == 0 def test_get_import_statistics_handles_errors(self, db): """Test import statistics returns zeros on error.""" with patch.object(db, "query", side_effect=SQLAlchemyError("DB Error")): stats = self.service.get_import_statistics(db) # Should return default values, not raise exception assert stats["total"] == 0 assert stats["completed"] == 0 assert stats["failed"] == 0 assert stats["success_rate"] == 0 # ==================== Private Helper Method Tests ==================== def test_parse_period_known_values(self): """Test period parsing for known values.""" assert self.service._parse_period("7d") == 7 assert self.service._parse_period("30d") == 30 assert self.service._parse_period("90d") == 90 assert self.service._parse_period("1y") == 365 def test_parse_period_unknown_defaults(self): """Test period parsing defaults to 30 for unknown values.""" assert self.service._parse_period("unknown") == 30 assert self.service._parse_period("") == 30 def test_get_unique_brands_count(self, db, test_marketplace_product): """Test getting unique brands count.""" unique_id = str(uuid.uuid4())[:8] create_marketplace_product_with_translation( db, marketplace_product_id=f"BRAND001_{unique_id}", title="Brand MarketplaceProduct 1", brand="BrandA", marketplace="Test", store_name="TestStore", price="10.00", currency="EUR", ) create_marketplace_product_with_translation( db, marketplace_product_id=f"BRAND002_{unique_id}", title="Brand MarketplaceProduct 2", brand="BrandB", marketplace="Test", store_name="TestStore", price="15.00", currency="EUR", ) db.commit() count = self.service._get_unique_brands_count(db) assert count >= 2 # At least BrandA and BrandB assert isinstance(count, int) def test_get_unique_categories_count(self, db, test_marketplace_product): """Test getting unique categories count.""" unique_id = str(uuid.uuid4())[:8] create_marketplace_product_with_translation( db, marketplace_product_id=f"CAT001_{unique_id}", title="Category MarketplaceProduct 1", google_product_category="Electronics", marketplace="Test", store_name="TestStore", price="10.00", currency="EUR", ) create_marketplace_product_with_translation( db, marketplace_product_id=f"CAT002_{unique_id}", title="Category MarketplaceProduct 2", google_product_category="Books", marketplace="Test", store_name="TestStore", price="15.00", currency="EUR", ) db.commit() count = self.service._get_unique_categories_count(db) assert count >= 2 # At least Electronics and Books assert isinstance(count, int) def test_get_inventory_statistics(self, db, test_inventory): """Test getting inventory statistics.""" unique_id = str(uuid.uuid4())[:8] additional_inventory = Inventory( gtin=f"123456789{unique_id[:4]}", warehouse="strassen", bin_location=f"ST-{unique_id[:2]}-01", quantity=25, reserved_quantity=5, store_id=test_inventory.store_id, product_id=test_inventory.product_id, ) db.add(additional_inventory) db.commit() stats = self.service._get_inventory_statistics(db) assert "total_entries" in stats assert "total_quantity" in stats assert "total_reserved" in stats assert "total_available" in stats assert stats["total_entries"] >= 2 assert stats["total_quantity"] >= test_inventory.quantity + 25 def test_get_inventory_statistics_empty(self, db): """Test inventory statistics with empty database.""" stats = self.service._get_inventory_statistics(db) assert stats["total_entries"] == 0 assert stats["total_quantity"] == 0 assert stats["total_reserved"] == 0 assert stats["total_available"] == 0