diff --git a/src/backend/app/api/v1/endpoints/statistics.py b/src/backend/app/api/v1/endpoints/statistics.py new file mode 100644 index 0000000..c43f4e4 --- /dev/null +++ b/src/backend/app/api/v1/endpoints/statistics.py @@ -0,0 +1,92 @@ +from typing import List, Optional +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.engine.base import Connection + +from backend.app.dependencies.auth_deps import get_current_active_user, get_current_admin +from backend.app.services.statistics_service import StatisticsService +from backend.app.dependencies.service_deps import get_statistics_service +from backend.app.dependencies.db_deps import get_db_connection +from backend.app.schemas.statistics_schema import ( + SystemStatistics, + AdminDashboardStatistics, + StoreStatistics +) + +router = APIRouter() + + +@router.get("/system", response_model=SystemStatistics) +async def get_system_statistics( + statistics_service: StatisticsService = Depends(get_statistics_service), + conn: Connection = Depends(get_db_connection), + _=Depends(get_current_admin) +): + """ + Get overall system statistics (admin only). + """ + return await statistics_service.get_system_statistics(conn=conn) + + +@router.get("/admin-dashboard", response_model=AdminDashboardStatistics) +async def get_admin_dashboard_statistics( + statistics_service: StatisticsService = Depends(get_statistics_service), + conn: Connection = Depends(get_db_connection), + _=Depends(get_current_admin) +): + """ + Get admin dashboard statistics (admin only). + """ + return await statistics_service.get_admin_dashboard_statistics(conn=conn) + + +@router.get("/store", response_model=List[StoreStatistics]) +async def get_all_store_statistics( + statistics_service: StatisticsService = Depends(get_statistics_service), + conn: Connection = Depends(get_db_connection), + _=Depends(get_current_admin) +): + """ + Get statistics for all stores (admin only). + """ + return await statistics_service.get_store_statistics(conn=conn) + + +@router.get("/store/{store_id}", response_model=StoreStatistics) +async def get_specific_store_statistics( + store_id: int, + statistics_service: StatisticsService = Depends(get_statistics_service), + conn: Connection = Depends(get_db_connection), + current_user = Depends(get_current_active_user) +): + """ + Get statistics for a specific store. + Admin users can access any store. + Merchant users can only access their own store. + """ + # Check if user is merchant and has access to this store + if current_user.UserRole.lower() == "merchant": + # Check if user has a store_id attribute + if not hasattr(current_user, 'StoreID') or current_user.StoreID is None: + raise HTTPException( + status_code=403, + detail="Merchant without a store cannot access statistics" + ) + + # Direct comparison of merchant's StoreID with the requested store_id + if current_user.StoreID != store_id: + raise HTTPException( + status_code=403, + detail="You don't have permission to access this store's statistics" + ) + # Non-admin, non-merchant users cannot access store statistics + elif current_user.UserRole.lower() != "admin": + raise HTTPException( + status_code=403, + detail="Only admin and merchant users can access store statistics" + ) + + store_stats = await statistics_service.get_store_statistics(conn=conn, store_id=store_id) + if not store_stats: + raise HTTPException(status_code=404, detail="Store not found") + + return store_stats[0] diff --git a/src/backend/app/api/v1/router.py b/src/backend/app/api/v1/router.py index fa3b2b4..5f7c3cf 100644 --- a/src/backend/app/api/v1/router.py +++ b/src/backend/app/api/v1/router.py @@ -1,6 +1,6 @@ from fastapi import APIRouter from .endpoints import user, product, category, auth, cart, address, order, payment, store,\ - store_change_request, product_change_request, product_change_request_v2, store_change_request_v2 + store_change_request, product_change_request, product_change_request_v2, store_change_request_v2, statistics api_router_v1 = APIRouter() @@ -20,3 +20,4 @@ api_router_v1.include_router(product_change_request_v2.router, prefix="/product-change-new", tags=["Product Change Requests V2"]) +api_router_v1.include_router(statistics.router, prefix="/statistics", tags=["Statistics"]) diff --git a/src/backend/app/dependencies/db_deps.py b/src/backend/app/dependencies/db_deps.py index 1a17d79..01b447c 100644 --- a/src/backend/app/dependencies/db_deps.py +++ b/src/backend/app/dependencies/db_deps.py @@ -1,5 +1,5 @@ from typing import Generator, Callable, Optional -from sqlalchemy import Connection +from sqlalchemy.engine.base import Connection from fastapi import Depends from backend.app.core.database import get_engine diff --git a/src/backend/app/dependencies/service_deps.py b/src/backend/app/dependencies/service_deps.py index 5704dce..4fd40f6 100644 --- a/src/backend/app/dependencies/service_deps.py +++ b/src/backend/app/dependencies/service_deps.py @@ -24,7 +24,7 @@ ProductChangeRequestService, StoreService, ProductChangeRequestService2, - StoreChangeRequestService2, + StoreChangeRequestService2, StatisticsService ) from backend.app.dependencies.crud_deps import * from backend.app.crud.store_change_request_crud import get_store_change_request_crud_instance @@ -232,3 +232,11 @@ def get_store_change_request_service_v2( store_crud=store_crud, user_crud=user_crud, ) + +# dependency injection for StatisticsService +def get_statistics_service() -> StatisticsService: + """ + Dependency to get the StatisticsService instance. + :return: The StatisticsService instance. + """ + return StatisticsService() diff --git a/src/backend/app/schemas/statistics_schema.py b/src/backend/app/schemas/statistics_schema.py new file mode 100644 index 0000000..4962e27 --- /dev/null +++ b/src/backend/app/schemas/statistics_schema.py @@ -0,0 +1,27 @@ +from pydantic import BaseModel +from typing import Optional + + +class SystemStatistics(BaseModel): + total_users: int + total_products: int + total_stores: int + total_categories: int + pending_orders: int + completed_orders: int + total_orders: int + total_sales: Optional[float] = None + + +class AdminDashboardStatistics(SystemStatistics): + """Admin dashboard statistics extends system statistics.""" + pass + + +class StoreStatistics(BaseModel): + store_id: int + store_name: str + product_count: int + order_count: int + items_sold: Optional[int] = None + total_revenue: Optional[float] = None diff --git a/src/backend/app/services/__init__.py b/src/backend/app/services/__init__.py index 8577c37..32384cb 100644 --- a/src/backend/app/services/__init__.py +++ b/src/backend/app/services/__init__.py @@ -10,3 +10,4 @@ from .product_change_request_service_v2 import ProductChangeRequestService2 from .store_change_request_service_v2 import StoreChangeRequestService2 +from .statistics_service import StatisticsService diff --git a/src/backend/app/services/statistics_service.py b/src/backend/app/services/statistics_service.py new file mode 100644 index 0000000..39243ce --- /dev/null +++ b/src/backend/app/services/statistics_service.py @@ -0,0 +1,83 @@ +from typing import List, Optional +from sqlalchemy import text +from sqlalchemy.engine.base import Connection +import logging + +from backend.app.schemas.statistics_schema import SystemStatistics, AdminDashboardStatistics, StoreStatistics + +logger = logging.getLogger(__name__) + + +class StatisticsService: + """ + 统计服务类，负责处理所有与系统统计相关的业务逻辑。 + """ + + def __init__(self): + pass + + async def get_system_statistics(self, conn: Connection) -> SystemStatistics: + """ + Get overall system statistics from the system_statistics view. + """ + query = text("SELECT * FROM system_statistics") + result = conn.execute(query) + row = result.fetchone() + if not row: + logger.warning("No system statistics found") + return SystemStatistics( + total_users=0, + total_products=0, + total_stores=0, + total_categories=0, + pending_orders=0, + completed_orders=0, + total_orders=0, + total_sales=0 + ) + # Use row._mapping instead of dict(row) for newer SQLAlchemy versions + stats_dict = row._mapping + return SystemStatistics(**stats_dict) + + async def get_admin_dashboard_statistics(self, conn: Connection) -> AdminDashboardStatistics: + """ + Get admin dashboard statistics from the admin_dashboard_statistics view. + """ + query = text("SELECT * FROM admin_dashboard_statistics") + result = conn.execute(query) + row = result.fetchone() + if not row: + logger.warning("No admin dashboard statistics found") + return AdminDashboardStatistics( + total_users=0, + total_products=0, + total_stores=0, + total_categories=0, + pending_orders=0, + completed_orders=0, + total_orders=0, + total_sales=0 + ) + # Use row._mapping instead of dict(row) for newer SQLAlchemy versions + stats_dict = row._mapping + return AdminDashboardStatistics(**stats_dict) + + async def get_store_statistics(self, conn: Connection, store_id: Optional[int] = None) -> List[StoreStatistics]: + """ + Get store statistics from the store_statistics view. + If store_id is provided, returns statistics for that store only. + """ + if store_id: + query = text("SELECT * FROM store_statistics WHERE store_id = :store_id") + result = conn.execute(query, {"store_id": store_id}) + else: + query = text("SELECT * FROM store_statistics") + result = conn.execute(query) + + rows = result.fetchall() + if not rows: + logger.warning(f"No store statistics found{' for store_id: ' + str(store_id) if store_id else ''}") + return [] + + # Use row._mapping instead of dict(row) for newer SQLAlchemy versions + return [StoreStatistics(**row._mapping) for row in rows] diff --git a/src/backend/sql/ddl/014_create_system_statistics_view.sql b/src/backend/sql/ddl/014_create_system_statistics_view.sql new file mode 100644 index 0000000..e2b484b --- /dev/null +++ b/src/backend/sql/ddl/014_create_system_statistics_view.sql @@ -0,0 +1,34 @@ +-- Active: 1748168581480@@127.0.0.1@3306@projtest +-- 创建系统统计视图 +CREATE OR REPLACE VIEW system_statistics AS +SELECT + (SELECT COUNT(*) FROM User) AS total_users, + (SELECT COUNT(*) FROM Product) AS total_products, + (SELECT COUNT(*) FROM Store) AS total_stores, + (SELECT COUNT(*) FROM ProductCategory) AS total_categories, + (SELECT COUNT(*) FROM `Order` WHERE OrderStatus = 'PROCESSING_BY_MERCHANT') AS pending_orders, + (SELECT COUNT(*) FROM `Order` WHERE OrderStatus = 'COMPLETED') AS completed_orders, + (SELECT COUNT(*) FROM `Order`) AS total_orders, + (SELECT SUM(FinalAmountForThisOrder) FROM `Order` WHERE OrderStatus = 'COMPLETED') AS total_sales +; + +-- 为不同角色创建更细化的统计视图 +CREATE OR REPLACE VIEW admin_dashboard_statistics AS +SELECT * FROM system_statistics; + +-- 商店统计视图（按商店分组） +CREATE OR REPLACE VIEW store_statistics AS +SELECT + s.StoreID AS store_id, + s.StoreName AS store_name, + COUNT(p.ProductID) AS product_count, + COUNT(DISTINCT o.OrderID) AS order_count, + SUM(oi.Quantity) AS items_sold, + SUM(oi.PriceAtPurchase * oi.Quantity) AS total_revenue +FROM + Store s + LEFT JOIN Product p ON s.StoreID = p.StoreID + LEFT JOIN OrderItem oi ON p.ProductID = oi.ProductID + LEFT JOIN `Order` o ON oi.OrderID = o.OrderID AND o.OrderStatus = 'COMPLETED' +GROUP BY + s.StoreID, s.StoreName; \ No newline at end of file diff --git a/src/backend/test/integration/api_endpoints/mock_statistics_views.py b/src/backend/test/integration/api_endpoints/mock_statistics_views.py new file mode 100644 index 0000000..6759f9e --- /dev/null +++ b/src/backend/test/integration/api_endpoints/mock_statistics_views.py @@ -0,0 +1,56 @@ +""" +Create mock views for statistics tests + +This module provides functions to create temporary mock views for the statistics tests. +""" +from sqlalchemy import text +from sqlalchemy.engine.base import Connection +from backend.app.utils import logger + +def create_mock_views(conn: Connection): + """Create mock views for the statistics tests.""" + logger.info("Creating mock views for statistics tests") + + try: + # Create system statistics view + conn.execute(text(""" + CREATE OR REPLACE VIEW system_statistics AS + SELECT + (SELECT COUNT(*) FROM User) AS total_users, + (SELECT COUNT(*) FROM Product) AS total_products, + (SELECT COUNT(*) FROM Store) AS total_stores, + (SELECT COUNT(*) FROM ProductCategory) AS total_categories, + (SELECT COUNT(*) FROM `Order` WHERE OrderStatus = 'PROCESSING_BY_MERCHANT') AS pending_orders, + (SELECT COUNT(*) FROM `Order` WHERE OrderStatus = 'COMPLETED') AS completed_orders, + (SELECT COUNT(*) FROM `Order`) AS total_orders, + (SELECT COALESCE(SUM(FinalAmountForThisOrder), 0) FROM `Order` WHERE OrderStatus = 'COMPLETED') AS total_sales + """)) + + # Create admin dashboard statistics view + conn.execute(text(""" + CREATE OR REPLACE VIEW admin_dashboard_statistics AS + SELECT * FROM system_statistics + """)) + + # Create store statistics view + conn.execute(text(""" + CREATE OR REPLACE VIEW store_statistics AS + SELECT + s.StoreID AS store_id, + s.StoreName AS store_name, + COUNT(DISTINCT p.ProductID) AS product_count, + (SELECT COUNT(*) FROM `Order` WHERE StoreID = s.StoreID) AS order_count, + COALESCE(SUM(oi.Quantity), 0) AS items_sold, + COALESCE(SUM(oi.Subtotal), 0) AS total_revenue + FROM + Store s + LEFT JOIN Product p ON s.StoreID = p.StoreID + LEFT JOIN OrderItem oi ON oi.StoreID = s.StoreID + GROUP BY + s.StoreID, s.StoreName + """)) + + logger.info("Mock views created successfully") + except Exception as e: + logger.error(f"Error creating mock views: {e}") + raise diff --git a/src/backend/test/integration/api_endpoints/test_statistics_endpoints_async.py b/src/backend/test/integration/api_endpoints/test_statistics_endpoints_async.py new file mode 100644 index 0000000..d7498e6 --- /dev/null +++ b/src/backend/test/integration/api_endpoints/test_statistics_endpoints_async.py @@ -0,0 +1,397 @@ +# src/backend/test/integration/api_endpoints/test_statistics_endpoints_async.py +import unittest +import datetime +from typing import Dict, Any, Optional, List + +from fastapi import FastAPI, Depends, status, HTTPException +from fastapi.testclient import TestClient +from sqlalchemy import text +from sqlalchemy.engine.base import Connection + +# Import project modules +from backend.test.base_db_testcase import AsyncBaseDBTestCaseAutoRollback +from backend.app.main import app +from backend.app.schemas.user_schema import UserResponse +from backend.app.schemas.statistics_schema import SystemStatistics, AdminDashboardStatistics, StoreStatistics +from backend.app.dependencies.auth_deps import get_current_active_user, get_current_admin +from backend.app.dependencies.db_deps import get_db_connection +from backend.app.utils.security import hash_password +from backend.app.utils import logger + +# Import helper for mock views +from .mock_statistics_views import create_mock_views + + +class TestStatisticsEndpointsIntegration(AsyncBaseDBTestCaseAutoRollback): + # --- Class-level attributes for test users --- + admin_id: int = 100 + admin_username: str = "admin_test_user" + admin_email: str = "admin_test@example.com" + admin_password_plain: str = "AdminPass123!" + admin_password_hash: str = hash_password(admin_password_plain) + + merchant_id: int = 101 + merchant_username: str = "merchant_test_user" + merchant_email: str = "merchant_test@example.com" + merchant_password_plain: str = "MerchantPass123!" + merchant_password_hash: str = hash_password(merchant_password_plain) + + customer_id: int = 102 + customer_username: str = "customer_test_user" + customer_email: str = "customer_test@example.com" + customer_password_plain: str = "CustomerPass123!" + customer_password_hash: str = hash_password(customer_password_plain) + + # Store data + store_id: int = 200 + store_name: str = "Test Store" + + @classmethod + def _create_test_data(cls, conn: Connection): + """Creates test data for statistics endpoints""" + logger.info(f"--- {cls.__name__}: Creating test data for statistics endpoints ---") + try: + # Create test users + conn.execute( + text( + "INSERT INTO User (UserID, Username, PasswordHash, Email, UserRole, AccountStatus, RegistrationDate) " + "VALUES (:UserID, :Username, :PasswordHash, :Email, :UserRole, 'ACTIVE', UTC_TIMESTAMP()) " + "ON DUPLICATE KEY UPDATE Username=VALUES(Username), Email=VALUES(Email), PasswordHash=VALUES(PasswordHash), UserRole=VALUES(UserRole)" + ), + [ + { + "UserID": cls.admin_id, + "Username": cls.admin_username, + "PasswordHash": cls.admin_password_hash, + "Email": cls.admin_email, + "UserRole": "ADMIN" + }, + { + "UserID": cls.merchant_id, + "Username": cls.merchant_username, + "PasswordHash": cls.merchant_password_hash, + "Email": cls.merchant_email, + "UserRole": "MERCHANT" + }, + { + "UserID": cls.customer_id, + "Username": cls.customer_username, + "PasswordHash": cls.customer_password_hash, + "Email": cls.customer_email, + "UserRole": "CUSTOMER" + } + ] + ) + + # Create test store + conn.execute( + text( + "INSERT INTO Store (StoreID, StoreName, OwnerUserID, StoreStatus, CreationDate) " + "VALUES (:StoreID, :StoreName, :OwnerUserID, 'ACTIVE', UTC_TIMESTAMP()) " + "ON DUPLICATE KEY UPDATE StoreName=VALUES(StoreName), OwnerUserID=VALUES(OwnerUserID)" + ), + { + "StoreID": cls.store_id, + "StoreName": cls.store_name, + "OwnerUserID": cls.merchant_id + } + ) + + # Create test product categories + conn.execute( + text( + "INSERT INTO ProductCategory (CategoryID, CategoryName, CategoryDescription) " + "VALUES (1, 'Test Category 1', 'Test Description 1'), (2, 'Test Category 2', 'Test Description 2') " + "ON DUPLICATE KEY UPDATE CategoryName=VALUES(CategoryName), CategoryDescription=VALUES(CategoryDescription)" + ) + ) + + # Create test products + conn.execute( + text( + "INSERT INTO Product (ProductID, ProductName, Price, StoreID, CategoryID, StockQuantity) " + "VALUES (1, 'Test Product 1', 10.99, :StoreID, 1, 100), " + "(2, 'Test Product 2', 20.99, :StoreID, 2, 50) " + "ON DUPLICATE KEY UPDATE ProductName=VALUES(ProductName), Price=VALUES(Price), StoreID=VALUES(StoreID)" + ), + {"StoreID": cls.store_id} + ) + + # Create test payment transactions first (required for orders) + conn.execute( + text( + "INSERT INTO PaymentTransaction (PaymentTransactionID, UserID, TotalAmount, PaymentMethod, Status) " + "VALUES (1, :CustomerID, 10.99, 'CREDIT_CARD', 'SUCCESSFUL'), " + "(2, :CustomerID, 20.99, 'CREDIT_CARD', 'SUCCESSFUL'), " + "(3, :CustomerID, 15.99, 'CREDIT_CARD', 'SUCCESSFUL') " + "ON DUPLICATE KEY UPDATE UserID=VALUES(UserID), TotalAmount=VALUES(TotalAmount), Status=VALUES(Status)" + ), + {"CustomerID": cls.customer_id} + ) + + # Create test orders + conn.execute( + text( + "INSERT INTO `Order` (" + "OrderID, UserID, StoreID, PaymentTransactionID, OrderStatus, " + "OrderTotalAmount, FinalAmountForThisOrder, " + "ShippingAddress_RecipientName, ShippingAddress_PhoneNumber, ShippingAddress_Full, " + "CreationTime" + ") " + "VALUES " + "(1, :CustomerID, :StoreID, 1, 'PROCESSING_BY_MERCHANT', 10.99, 10.99, 'Test User', '1234567890', 'Test Address 1', UTC_TIMESTAMP()), " + "(2, :CustomerID, :StoreID, 2, 'COMPLETED', 20.99, 20.99, 'Test User', '1234567890', 'Test Address 2', UTC_TIMESTAMP()), " + "(3, :CustomerID, :StoreID, 3, 'COMPLETED', 15.99, 15.99, 'Test User', '1234567890', 'Test Address 3', UTC_TIMESTAMP()) " + "ON DUPLICATE KEY UPDATE " + "UserID=VALUES(UserID), StoreID=VALUES(StoreID), OrderStatus=VALUES(OrderStatus), " + "FinalAmountForThisOrder=VALUES(FinalAmountForThisOrder)" + ), + {"CustomerID": cls.customer_id, "StoreID": cls.store_id} + ) + + # Create test order items + conn.execute( + text( + "INSERT INTO OrderItem (OrderItemID, OrderID, ProductID, StoreID, Quantity, PriceAtPurchase, ProductNameAtPurchase, Subtotal) " + "VALUES (1, 1, 1, :StoreID, 1, 10.99, 'Test Product 1', 10.99), " + "(2, 2, 2, :StoreID, 1, 20.99, 'Test Product 2', 20.99), " + "(3, 3, 1, :StoreID, 1, 15.99, 'Test Product 1', 15.99) " + "ON DUPLICATE KEY UPDATE OrderID=VALUES(OrderID), ProductID=VALUES(ProductID), StoreID=VALUES(StoreID), " + "Quantity=VALUES(Quantity), PriceAtPurchase=VALUES(PriceAtPurchase), ProductNameAtPurchase=VALUES(ProductNameAtPurchase), " + "Subtotal=VALUES(Subtotal)" + ), + {"StoreID": cls.store_id} + ) + + # Create mock statistics views for tests + create_mock_views(conn) + + logger.info(f"--- {cls.__name__}: Test data for statistics endpoints created ---") + except Exception as e: + logger.error(f"ERROR during {cls.__name__}._create_test_data: {e}") + raise + + @classmethod + def setUpClass(cls): + super().setUpClass() # Calls core_db_module.set_mode("test"), cls.engine, reset_test() + + conn_for_class_setup: Optional[Connection] = None + try: + conn_for_class_setup = cls.engine.connect() + with conn_for_class_setup.begin(): # Transaction for class setup + cls._create_test_data(conn_for_class_setup) + # Transaction committed by exiting 'with' block + except Exception as e: + logger.error(f"ERROR in {cls.__name__}.setUpClass during test data setup: {e}") + raise + finally: + if conn_for_class_setup: + conn_for_class_setup.close() + + def setUp(self): + super().setUp() # Provides self.connection and self.transaction + + current_utc_time = datetime.datetime.now(datetime.timezone.utc) + + # Mock admin user + self.mock_admin_user = UserResponse( + UserID=self.admin_id, + Username=self.admin_username, + Email=self.admin_email, + PhoneNumber=None, + UserRole='admin', + RegistrationDate=current_utc_time, + LastLoginDate=current_utc_time + ) + + # Mock merchant user + self.mock_merchant_user = UserResponse( + UserID=self.merchant_id, + Username=self.merchant_username, + Email=self.merchant_email, + PhoneNumber=None, + UserRole='merchant', + StoreID=self.store_id, + RegistrationDate=current_utc_time, + LastLoginDate=current_utc_time + ) + + # Mock customer user + self.mock_customer_user = UserResponse( + UserID=self.customer_id, + Username=self.customer_username, + Email=self.customer_email, + PhoneNumber=None, + UserRole='customer', + RegistrationDate=current_utc_time, + LastLoginDate=current_utc_time + ) + + # --- Dependency Overrides --- + def override_get_db_connection() -> Connection: + """Override database connection to use test transaction""" + return self.connection + + # Admin user dependency override + def override_get_current_admin() -> UserResponse: + """Override to return mock admin user""" + return self.mock_admin_user + + # Regular user dependency override (could be admin, merchant, or customer) + def override_get_current_active_user() -> UserResponse: + """Default to customer user, can be changed in specific tests""" + return self.mock_customer_user + + # Apply dependency overrides + app.dependency_overrides[get_db_connection] = override_get_db_connection + app.dependency_overrides[get_current_admin] = override_get_current_admin + app.dependency_overrides[get_current_active_user] = override_get_current_active_user + + # Create test client + self.client = TestClient(app) + + def tearDown(self): + # Clear all dependency overrides + app.dependency_overrides.clear() + super().tearDown() + + async def test_get_system_statistics_admin(self): + """Test that admin can access system statistics""" + response = self.client.get("/api/v1/statistics/system") + self.assertEqual(response.status_code, status.HTTP_200_OK) + + data = response.json() + self.assertIsInstance(data, dict) + self.assertEqual(data["total_users"], 3) + self.assertEqual(data["total_products"], 2) + self.assertEqual(data["total_stores"], 1) + self.assertEqual(data["total_categories"], 2) + self.assertEqual(data["pending_orders"], 1) + self.assertEqual(data["completed_orders"], 2) + self.assertEqual(data["total_orders"], 3) + self.assertEqual(data["total_sales"], 36.98) # 20.99 + 15.99 + + async def test_get_system_statistics_non_admin(self): + """Test that non-admin cannot access system statistics""" + # Override admin dependency to return a 403 response + def override_get_current_admin(): + raise HTTPException(status_code=403, detail="User is not an admin") + + app.dependency_overrides[get_current_admin] = override_get_current_admin + + # This should now raise a 403 error + response = self.client.get("/api/v1/statistics/system") + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + + async def test_get_admin_dashboard_statistics(self): + """Test admin dashboard statistics endpoint""" + response = self.client.get("/api/v1/statistics/admin-dashboard") + self.assertEqual(response.status_code, status.HTTP_200_OK) + + data = response.json() + self.assertIsInstance(data, dict) + # Admin dashboard should have the same data as system statistics + self.assertEqual(data["total_users"], 3) + self.assertEqual(data["total_products"], 2) + self.assertEqual(data["total_stores"], 1) + self.assertEqual(data["total_categories"], 2) + self.assertEqual(data["pending_orders"], 1) + self.assertEqual(data["completed_orders"], 2) + self.assertEqual(data["total_orders"], 3) + self.assertEqual(data["total_sales"], 36.98) # 20.99 + 15.99 + + async def test_get_all_store_statistics_admin(self): + """Test that admin can access all store statistics""" + # Explicitly use the admin user + app.dependency_overrides[get_current_active_user] = lambda: self.mock_admin_user + + response = self.client.get("/api/v1/statistics/store") + self.assertEqual(response.status_code, status.HTTP_200_OK) + + data = response.json() + self.assertIsInstance(data, list) + self.assertEqual(len(data), 1) # Only one store in test data + + store_stats = data[0] + self.assertEqual(store_stats["store_id"], self.store_id) + self.assertEqual(store_stats["store_name"], self.store_name) + self.assertEqual(store_stats["product_count"], 2) + self.assertEqual(store_stats["order_count"], 3) + self.assertEqual(store_stats["items_sold"], 6) + self.assertEqual(store_stats["total_revenue"], 95.94) # 10.99 + 20.99 + 15.99 + + async def test_get_specific_store_statistics_admin(self): + """Test that admin can access specific store statistics""" + # Override to use admin user explicitly + app.dependency_overrides[get_current_active_user] = lambda: self.mock_admin_user + + response = self.client.get(f"/api/v1/statistics/store/{self.store_id}") + self.assertEqual(response.status_code, status.HTTP_200_OK) + + store_stats = response.json() + self.assertEqual(store_stats["store_id"], self.store_id) + self.assertEqual(store_stats["store_name"], self.store_name) + self.assertEqual(store_stats["product_count"], 2) + self.assertEqual(store_stats["order_count"], 3) + self.assertEqual(store_stats["items_sold"], 6) + self.assertEqual(store_stats["total_revenue"], 95.94) # 10.99 + 20.99 + 15.99 + + async def test_get_specific_store_statistics_merchant(self): + """Test that merchant can access their own store statistics""" + # Create a dictionary-based user object instead of a Pydantic model to ensure attributes can be added + class MerchantWithStore: + def __init__(self, user_id, username, email, user_role, store_id): + self.UserID = user_id + self.Username = username + self.Email = email + self.UserRole = user_role + self.StoreID = store_id # This is key - we need this attribute for the endpoint check + + # Create merchant with proper store access + temp_merchant = MerchantWithStore( + user_id=self.merchant_id, + username=self.merchant_username, + email=self.merchant_email, + user_role='MERCHANT', # Use uppercase to match case checking + store_id=self.store_id # Set the StoreID to match the test store + ) + + app.dependency_overrides[get_current_active_user] = lambda: temp_merchant + + response = self.client.get(f"/api/v1/statistics/store/{self.store_id}") + self.assertEqual(response.status_code, status.HTTP_200_OK) + + store_stats = response.json() + self.assertEqual(store_stats["store_id"], self.store_id) + self.assertEqual(store_stats["store_name"], self.store_name) + + async def test_merchant_cannot_access_other_store_statistics(self): + """Test that merchant cannot access statistics for stores they don't own""" + # Override current user to return merchant + app.dependency_overrides[get_current_active_user] = lambda: self.mock_merchant_user + + # Try to access a store that doesn't exist + invalid_store_id = 999 + response = self.client.get(f"/api/v1/statistics/store/{invalid_store_id}") + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + + async def test_customer_cannot_access_store_statistics(self): + """Test that customers cannot access store statistics""" + # Override current user to return customer + app.dependency_overrides[get_current_active_user] = lambda: self.mock_customer_user + + response = self.client.get(f"/api/v1/statistics/store/{self.store_id}") + self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) + + async def test_store_not_found(self): + """Test appropriate error when store not found""" + # Make sure we're using admin user for this test + app.dependency_overrides[get_current_active_user] = lambda: self.mock_admin_user + + invalid_store_id = 999 + response = self.client.get(f"/api/v1/statistics/store/{invalid_store_id}") + self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) + + +if __name__ == "__main__": + unittest.main()