diff --git a/app/routers/__init__.py b/app/routers/__init__.py new file mode 100644 index 0000000..fa50519 --- /dev/null +++ b/app/routers/__init__.py @@ -0,0 +1,15 @@ +from app.routers.auth import router as auth_router +from app.routers.business_types import router as business_types_router +from app.routers.plans import router as plans_router +from app.routers.tenants import router as tenants_router +from app.routers.subscriptions import router as subscriptions_router +from app.routers.users import router as users_router + +__all__ = [ + "auth_router", + "business_types_router", + "plans_router", + "tenants_router", + "subscriptions_router", + "users_router", +] diff --git a/app/routers/auth.py b/app/routers/auth.py new file mode 100644 index 0000000..8771b72 --- /dev/null +++ b/app/routers/auth.py @@ -0,0 +1,46 @@ +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.orm import Session + +from app.database import get_db +from app.models.user import User +from app.models.user_profile import UserProfile +from app.schemas.user import UserCreate, UserResponse +from app.security import hash_password + +router = APIRouter(prefix="/auth", tags=["auth"]) + + +@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED) +def register(data: UserCreate, db: Session = Depends(get_db)): + existing = db.query(User).filter( + (User.email == data.email) | (User.username == data.username) + ).first() + if existing: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"message": "Email or username already registered"}, + ) + + user = User( + email=data.email, + username=data.username, + password=hash_password(data.password), + ) + db.add(user) + db.flush() + + try: + profile = UserProfile( + user_id=user.id, + full_name=data.full_name, + phone=data.phone, + timezone=data.timezone, + ) + db.add(profile) + db.commit() + except: + db.rollback() + raise + + db.refresh(user) + return user diff --git a/app/routers/business_types.py b/app/routers/business_types.py new file mode 100644 index 0000000..be53acc --- /dev/null +++ b/app/routers/business_types.py @@ -0,0 +1,50 @@ +from fastapi import APIRouter, Depends, Request, Response, status +from sqlalchemy.orm import Session + +from app.database import get_db +from app.responses import created_response +from app.schemas.business_type import ( + BusinessTypeCreate, + BusinessTypeResponse, + BusinessTypeUpdate, +) +from app.services.business_type_service import ( + create_business_type, + delete_business_type, + get_business_type_by_id, + get_business_type_list, + update_business_type, +) + +router = APIRouter(prefix="/business-types", tags=["business-types"]) + + +@router.get("/", response_model=list[BusinessTypeResponse]) +def list_business_types(db: Session = Depends(get_db)): + return get_business_type_list(db) + + +@router.get("/{id}", response_model=BusinessTypeResponse) +def get_business_type(id: str, db: Session = Depends(get_db)): + return get_business_type_by_id(db, id) + + +@router.post("/", status_code=status.HTTP_201_CREATED) +def create_business_type_route( + request: Request, data: BusinessTypeCreate, db: Session = Depends(get_db) +): + bt = create_business_type(db, data) + location = f"{request.url.path}{bt.id}" + return created_response(location) + + +@router.put("/{id}", response_model=BusinessTypeResponse) +def update_business_type_route( + id: str, data: BusinessTypeUpdate, db: Session = Depends(get_db) +): + return update_business_type(db, id, data) + + +@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_business_type_route(id: str, db: Session = Depends(get_db)): + delete_business_type(db, id) diff --git a/app/routers/plans.py b/app/routers/plans.py new file mode 100644 index 0000000..d3be617 --- /dev/null +++ b/app/routers/plans.py @@ -0,0 +1,42 @@ +from fastapi import APIRouter, Depends, Request, status +from sqlalchemy.orm import Session + +from app.database import get_db +from app.responses import created_response +from app.schemas.plan import PlanCreate, PlanResponse, PlanUpdate +from app.services.plan_service import ( + create_plan, + delete_plan, + get_plan_by_id, + get_plan_list, + update_plan, +) + +router = APIRouter(prefix="/plans", tags=["plans"]) + + +@router.get("/", response_model=list[PlanResponse]) +def list_plans(db: Session = Depends(get_db)): + return get_plan_list(db) + + +@router.get("/{id}", response_model=PlanResponse) +def get_plan(id: str, db: Session = Depends(get_db)): + return get_plan_by_id(db, id) + + +@router.post("/", status_code=status.HTTP_201_CREATED) +def create_plan_route(request: Request, data: PlanCreate, db: Session = Depends(get_db)): + plan = create_plan(db, data) + location = f"{request.url.path}{plan.id}" + return created_response(location) + + +@router.put("/{id}", response_model=PlanResponse) +def update_plan_route(id: str, data: PlanUpdate, db: Session = Depends(get_db)): + return update_plan(db, id, data) + + +@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_plan_route(id: str, db: Session = Depends(get_db)): + delete_plan(db, id) diff --git a/app/routers/subscriptions.py b/app/routers/subscriptions.py new file mode 100644 index 0000000..b1b6f41 --- /dev/null +++ b/app/routers/subscriptions.py @@ -0,0 +1,50 @@ +from fastapi import APIRouter, Depends, Request, status +from sqlalchemy.orm import Session + +from app.database import get_db +from app.responses import created_response +from app.schemas.subscription import ( + SubscriptionCreate, + SubscriptionResponse, + SubscriptionUpdate, +) +from app.services.subscription_service import ( + create_subscription, + delete_subscription, + get_subscription_by_id, + get_subscription_list, + update_subscription, +) + +router = APIRouter(prefix="/subscriptions", tags=["subscriptions"]) + + +@router.get("/", response_model=list[SubscriptionResponse]) +def list_subscriptions(db: Session = Depends(get_db)): + return get_subscription_list(db) + + +@router.get("/{id}", response_model=SubscriptionResponse) +def get_subscription(id: str, db: Session = Depends(get_db)): + return get_subscription_by_id(db, id) + + +@router.post("/", status_code=status.HTTP_201_CREATED) +def create_subscription_route( + request: Request, data: SubscriptionCreate, db: Session = Depends(get_db) +): + sub = create_subscription(db, data) + location = f"{request.url.path}{sub.id}" + return created_response(location) + + +@router.put("/{id}", response_model=SubscriptionResponse) +def update_subscription_route( + id: str, data: SubscriptionUpdate, db: Session = Depends(get_db) +): + return update_subscription(db, id, data) + + +@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_subscription_route(id: str, db: Session = Depends(get_db)): + delete_subscription(db, id) diff --git a/app/routers/tenants.py b/app/routers/tenants.py new file mode 100644 index 0000000..5940055 --- /dev/null +++ b/app/routers/tenants.py @@ -0,0 +1,50 @@ +from fastapi import APIRouter, Depends, Request, status +from sqlalchemy.orm import Session + +from app.database import get_db +from app.responses import created_response +from app.schemas.tenant import TenantCreate, TenantResponse, TenantUpdate +from app.services.tenant_service import ( + assign_user_to_tenant, + create_tenant, + delete_tenant, + get_tenant_by_id, + get_tenant_list, + update_tenant, +) + +router = APIRouter(prefix="/tenants", tags=["tenants"]) + + +@router.get("/", response_model=list[TenantResponse]) +def list_tenants(db: Session = Depends(get_db)): + return get_tenant_list(db) + + +@router.get("/{id}", response_model=TenantResponse) +def get_tenant(id: str, db: Session = Depends(get_db)): + return get_tenant_by_id(db, id) + + +@router.post("/", status_code=status.HTTP_201_CREATED) +def create_tenant_route( + request: Request, data: TenantCreate, db: Session = Depends(get_db) +): + tenant = create_tenant(db, data) + location = f"{request.url.path}{tenant.id}" + return created_response(location) + + +@router.put("/{id}", response_model=TenantResponse) +def update_tenant_route(id: str, data: TenantUpdate, db: Session = Depends(get_db)): + return update_tenant(db, id, data) + + +@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_tenant_route(id: str, db: Session = Depends(get_db)): + delete_tenant(db, id) + + +@router.post("/{tenant_id}/assign-user") +def assign_user_route(tenant_id: str, user_id: str, db: Session = Depends(get_db)): + return assign_user_to_tenant(db, tenant_id, user_id) diff --git a/app/routers/users.py b/app/routers/users.py new file mode 100644 index 0000000..bfc8c5e --- /dev/null +++ b/app/routers/users.py @@ -0,0 +1,58 @@ +from fastapi import APIRouter, Depends, Request, Response, status +from sqlalchemy.orm import Session + +from app.database import get_db +from app.responses import created_response +from app.schemas.user import ( + UserCreate, + UserDetailResponse, + UserResponse, + UserUpdate, +) +from app.services.user_service import ( + create_user, + delete_user, + get_user_by_id, + get_user_list, + update_user, +) + +router = APIRouter(prefix="/users", tags=["users"]) + + +@router.get("/", response_model=list[UserDetailResponse]) +def list_users( + response: Response, + db: Session = Depends(get_db), + page: int = 1, + limit: int = 10, +): + users, total = get_user_list(db, page, limit) + + response.headers["Pagination-Count"] = str(total) + response.headers["Pagination-Page"] = str(page) + response.headers["Pagination-Limit"] = str(limit) + + return users + + +@router.get("/{id}", response_model=UserDetailResponse) +def get_user(id: str, db: Session = Depends(get_db)): + return get_user_by_id(db, id) + + +@router.post("/", status_code=status.HTTP_201_CREATED) +def create_user_route(request: Request, data: UserCreate, db: Session = Depends(get_db)): + user = create_user(db, data) + location = f"{request.url.path}{user.id}" + return created_response(location) + + +@router.put("/{id}", response_model=UserDetailResponse) +def update_user_route(id: str, data: UserUpdate, db: Session = Depends(get_db)): + return update_user(db, id, data) + + +@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_user_route(id: str, db: Session = Depends(get_db)): + delete_user(db, id) diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py index 515d72c..4fe7009 100644 --- a/app/schemas/__init__.py +++ b/app/schemas/__init__.py @@ -1,17 +1,25 @@ -from app.schemas.user import UserCreate, UserResponse -from app.schemas.business_type import BusinessTypeCreate, BusinessTypeResponse -from app.schemas.plan import PlanCreate, PlanResponse -from app.schemas.tenant import TenantCreate, TenantResponse -from app.schemas.subscription import SubscriptionResponse +from app.schemas.user import UserCreate, UserResponse, UserDetailResponse, UserProfileResponse, UserUpdate +from app.schemas.business_type import BusinessTypeCreate, BusinessTypeResponse, BusinessTypeUpdate +from app.schemas.plan import PlanCreate, PlanResponse, PlanUpdate +from app.schemas.tenant import TenantCreate, TenantResponse, TenantUpdate +from app.schemas.subscription import SubscriptionCreate, SubscriptionResponse, SubscriptionUpdate __all__ = [ "UserCreate", "UserResponse", + "UserDetailResponse", + "UserProfileResponse", + "UserUpdate", "BusinessTypeCreate", "BusinessTypeResponse", + "BusinessTypeUpdate", "PlanCreate", "PlanResponse", + "PlanUpdate", "TenantCreate", "TenantResponse", + "TenantUpdate", + "SubscriptionCreate", "SubscriptionResponse", + "SubscriptionUpdate", ] diff --git a/app/schemas/business_type.py b/app/schemas/business_type.py index 6e44694..fba4c5d 100644 --- a/app/schemas/business_type.py +++ b/app/schemas/business_type.py @@ -9,6 +9,13 @@ class BusinessTypeCreate(BaseModel): description: str | None = None +class BusinessTypeUpdate(BaseModel): + code: str | None = None + name: str | None = None + description: str | None = None + is_active: bool | None = None + + class BusinessTypeResponse(BaseModel): id: str code: str diff --git a/app/schemas/plan.py b/app/schemas/plan.py index 59a691d..8187e37 100644 --- a/app/schemas/plan.py +++ b/app/schemas/plan.py @@ -11,6 +11,15 @@ class PlanCreate(BaseModel): limits: dict | None = None +class PlanUpdate(BaseModel): + code: str | None = None + name: str | None = None + description: str | None = None + price: int | None = None + limits: dict | None = None + is_active: bool | None = None + + class PlanResponse(BaseModel): id: str code: str diff --git a/app/schemas/subscription.py b/app/schemas/subscription.py index 124678b..c9738a1 100644 --- a/app/schemas/subscription.py +++ b/app/schemas/subscription.py @@ -3,6 +3,20 @@ from datetime import datetime from pydantic import BaseModel +class SubscriptionCreate(BaseModel): + tenant_id: str + plan_id: str + limits: dict | None = None + status: str = "TRIAL" + + +class SubscriptionUpdate(BaseModel): + plan_id: str | None = None + limits: dict | None = None + status: str | None = None + ended_at: datetime | None = None + + class SubscriptionResponse(BaseModel): id: str tenant_id: str diff --git a/app/schemas/tenant.py b/app/schemas/tenant.py index 683bed0..f3e8c16 100644 --- a/app/schemas/tenant.py +++ b/app/schemas/tenant.py @@ -14,6 +14,16 @@ class TenantCreate(BaseModel): limits: dict | None = None +class TenantUpdate(BaseModel): + name: str | None = None + slug: str | None = None + business_type_id: str | None = None + phone: str | None = None + address: str | None = None + email: str | None = None + status: str | None = None + + class TenantResponse(BaseModel): id: str name: str diff --git a/app/schemas/user.py b/app/schemas/user.py index d6224de..6a0e2fe 100644 --- a/app/schemas/user.py +++ b/app/schemas/user.py @@ -12,6 +12,16 @@ class UserCreate(BaseModel): timezone: str = "Asia/Jakarta" +class UserUpdate(BaseModel): + email: EmailStr | None = None + username: str | None = None + password: str | None = None + full_name: str | None = None + phone: str | None = None + timezone: str | None = None + status: str | None = None + + class UserResponse(BaseModel): id: str email: str @@ -22,3 +32,16 @@ class UserResponse(BaseModel): class Config: from_attributes = True + + +class UserProfileResponse(BaseModel): + full_name: str + phone: str | None + timezone: str | None + + class Config: + from_attributes = True + + +class UserDetailResponse(UserResponse): + profile: UserProfileResponse | None = None diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/services/business_type_service.py b/app/services/business_type_service.py new file mode 100644 index 0000000..09329b6 --- /dev/null +++ b/app/services/business_type_service.py @@ -0,0 +1,74 @@ +from datetime import datetime + +from fastapi import HTTPException, status +from sqlalchemy.orm import Session + +from app.models.business_type import BusinessType +from app.schemas.business_type import BusinessTypeCreate, BusinessTypeUpdate + + +def get_business_type_list(db: Session): + return db.query(BusinessType).filter(BusinessType.deleted_at.is_(None)).all() + + +def get_business_type_by_id(db: Session, id: str): + bt = db.query(BusinessType).filter(BusinessType.id == id, BusinessType.deleted_at.is_(None)).first() + if not bt: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + return bt + + +def create_business_type(db: Session, data: BusinessTypeCreate): + existing = db.query(BusinessType).filter(BusinessType.code == data.code).first() + if existing: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"message": "Code already exists"}, + ) + + bt = BusinessType(**data.model_dump()) + db.add(bt) + db.commit() + db.refresh(bt) + return bt + + +def update_business_type(db: Session, id: str, data: BusinessTypeUpdate): + bt = db.query(BusinessType).filter(BusinessType.id == id, BusinessType.deleted_at.is_(None)).first() + if not bt: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + + update_data = data.model_dump(exclude_unset=True) + + if "code" in update_data: + dup = db.query(BusinessType).filter(BusinessType.code == update_data["code"], BusinessType.id != id).first() + if dup: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"message": "Code already exists"}, + ) + + for field, value in update_data.items(): + setattr(bt, field, value) + + db.commit() + db.refresh(bt) + return bt + + +def delete_business_type(db: Session, id: str): + bt = db.query(BusinessType).filter(BusinessType.id == id, BusinessType.deleted_at.is_(None)).first() + if not bt: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + + bt.deleted_at = datetime.now() + db.commit() diff --git a/app/services/plan_service.py b/app/services/plan_service.py new file mode 100644 index 0000000..1217fb0 --- /dev/null +++ b/app/services/plan_service.py @@ -0,0 +1,74 @@ +from datetime import datetime + +from fastapi import HTTPException, status +from sqlalchemy.orm import Session + +from app.models.plan import Plan +from app.schemas.plan import PlanCreate, PlanUpdate + + +def get_plan_list(db: Session): + return db.query(Plan).filter(Plan.deleted_at.is_(None)).all() + + +def get_plan_by_id(db: Session, id: str): + plan = db.query(Plan).filter(Plan.id == id, Plan.deleted_at.is_(None)).first() + if not plan: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + return plan + + +def create_plan(db: Session, data: PlanCreate): + existing = db.query(Plan).filter(Plan.code == data.code).first() + if existing: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"message": "Code already exists"}, + ) + + plan = Plan(**data.model_dump()) + db.add(plan) + db.commit() + db.refresh(plan) + return plan + + +def update_plan(db: Session, id: str, data: PlanUpdate): + plan = db.query(Plan).filter(Plan.id == id, Plan.deleted_at.is_(None)).first() + if not plan: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + + update_data = data.model_dump(exclude_unset=True) + + if "code" in update_data: + dup = db.query(Plan).filter(Plan.code == update_data["code"], Plan.id != id).first() + if dup: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"message": "Code already exists"}, + ) + + for field, value in update_data.items(): + setattr(plan, field, value) + + db.commit() + db.refresh(plan) + return plan + + +def delete_plan(db: Session, id: str): + plan = db.query(Plan).filter(Plan.id == id, Plan.deleted_at.is_(None)).first() + if not plan: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + + plan.deleted_at = datetime.now() + db.commit() diff --git a/app/services/subscription_service.py b/app/services/subscription_service.py new file mode 100644 index 0000000..5faacf9 --- /dev/null +++ b/app/services/subscription_service.py @@ -0,0 +1,65 @@ +from datetime import datetime + +from fastapi import HTTPException, status +from sqlalchemy.orm import Session + +from app.models.subscription import Subscription +from app.schemas.subscription import SubscriptionCreate, SubscriptionUpdate + + +def get_subscription_list(db: Session): + return db.query(Subscription).all() + + +def get_subscription_by_id(db: Session, id: str): + sub = db.query(Subscription).filter(Subscription.id == id).first() + if not sub: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + return sub + + +def create_subscription(db: Session, data: SubscriptionCreate): + existing = db.query(Subscription).filter(Subscription.tenant_id == data.tenant_id).first() + if existing: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"message": "Tenant already has a subscription"}, + ) + + sub = Subscription(**data.model_dump()) + db.add(sub) + db.commit() + db.refresh(sub) + return sub + + +def update_subscription(db: Session, id: str, data: SubscriptionUpdate): + sub = db.query(Subscription).filter(Subscription.id == id).first() + if not sub: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + + update_data = data.model_dump(exclude_unset=True) + for field, value in update_data.items(): + setattr(sub, field, value) + + db.commit() + db.refresh(sub) + return sub + + +def delete_subscription(db: Session, id: str): + sub = db.query(Subscription).filter(Subscription.id == id).first() + if not sub: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + + db.delete(sub) + db.commit() diff --git a/app/services/tenant_service.py b/app/services/tenant_service.py new file mode 100644 index 0000000..d09dce4 --- /dev/null +++ b/app/services/tenant_service.py @@ -0,0 +1,118 @@ +from datetime import datetime + +from fastapi import HTTPException, status +from sqlalchemy.orm import Session + +from app.models.subscription import Subscription +from app.models.tenant import Tenant +from app.models.user import User +from app.schemas.tenant import TenantCreate, TenantUpdate + + +def get_tenant_list(db: Session): + return db.query(Tenant).filter(Tenant.deleted_at.is_(None)).all() + + +def get_tenant_by_id(db: Session, id: str): + tenant = db.query(Tenant).filter(Tenant.id == id, Tenant.deleted_at.is_(None)).first() + if not tenant: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + return tenant + + +def create_tenant(db: Session, data: TenantCreate): + existing = db.query(Tenant).filter(Tenant.slug == data.slug).first() + if existing: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"message": "Slug already exists"}, + ) + + try: + tenant = Tenant( + name=data.name, + slug=data.slug, + business_type_id=data.business_type_id, + phone=data.phone, + address=data.address, + email=data.email, + ) + db.add(tenant) + db.flush() + + subscription = Subscription( + tenant_id=tenant.id, + plan_id=data.plan_id, + limits=data.limits, + ) + db.add(subscription) + db.commit() + except: + db.rollback() + raise + + db.refresh(tenant) + return tenant + + +def update_tenant(db: Session, id: str, data: TenantUpdate): + tenant = db.query(Tenant).filter(Tenant.id == id, Tenant.deleted_at.is_(None)).first() + if not tenant: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + + update_data = data.model_dump(exclude_unset=True) + + if "slug" in update_data: + dup = db.query(Tenant).filter(Tenant.slug == update_data["slug"], Tenant.id != id).first() + if dup: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"message": "Slug already exists"}, + ) + + for field, value in update_data.items(): + setattr(tenant, field, value) + + db.commit() + db.refresh(tenant) + return tenant + + +def delete_tenant(db: Session, id: str): + tenant = db.query(Tenant).filter(Tenant.id == id, Tenant.deleted_at.is_(None)).first() + if not tenant: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + + tenant.deleted_at = datetime.now() + db.commit() + + +def assign_user_to_tenant(db: Session, tenant_id: str, user_id: str): + tenant = db.query(Tenant).filter(Tenant.id == tenant_id).first() + if not tenant: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "Tenant not found"}, + ) + + user = db.query(User).filter(User.id == user_id).first() + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "User not found"}, + ) + + user.tenant_id = tenant_id + user.status = "ACTIVE" + db.commit() + + return {"message": "User assigned to tenant"} diff --git a/app/services/user_service.py b/app/services/user_service.py new file mode 100644 index 0000000..e7d4c6e --- /dev/null +++ b/app/services/user_service.py @@ -0,0 +1,120 @@ +from datetime import datetime + +from fastapi import HTTPException, status +from sqlalchemy.orm import Session, selectinload + +from app.models.user import User +from app.models.user_profile import UserProfile +from app.schemas.user import UserCreate, UserUpdate +from app.security import hash_password + + +def get_user_list(db: Session, page: int, limit: int): + query = db.query(User).options(selectinload(User.profile)).filter(User.deleted_at.is_(None)) + total = query.count() + users = query.offset((page - 1) * limit).limit(limit).all() + return users, total + + +def get_user_by_id(db: Session, id: str): + user = db.query(User).options(selectinload(User.profile)).filter(User.id == id, User.deleted_at.is_(None)).first() + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + return user + + +def create_user(db: Session, data: UserCreate): + existing = db.query(User).filter( + (User.email == data.email) | (User.username == data.username) + ).first() + if existing: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"message": "Email or username already registered"}, + ) + + user = User( + email=data.email, + username=data.username, + password=hash_password(data.password), + ) + db.add(user) + db.flush() + + try: + profile = UserProfile( + user_id=user.id, + full_name=data.full_name, + phone=data.phone, + timezone=data.timezone, + ) + db.add(profile) + db.commit() + except: + db.rollback() + raise + + return user + + +def update_user(db: Session, id: str, data: UserUpdate): + user = db.query(User).filter(User.id == id, User.deleted_at.is_(None)).first() + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + + update_data = data.model_dump(exclude_unset=True) + + profile_fields = ["full_name", "phone", "timezone"] + user_fields = {k: v for k, v in update_data.items() if k not in profile_fields} + profile_update = {k: v for k, v in update_data.items() if k in profile_fields} + + if user_fields.get("email"): + dup = db.query(User).filter(User.email == user_fields["email"], User.id != id).first() + if dup: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"message": "Email already used"}, + ) + + if user_fields.get("username"): + dup = db.query(User).filter(User.username == user_fields["username"], User.id != id).first() + if dup: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail={"message": "Username already used"}, + ) + + if "password" in user_fields: + user_fields["password"] = hash_password(user_fields["password"]) + + for field, value in user_fields.items(): + setattr(user, field, value) + + if profile_update: + profile = db.query(UserProfile).filter(UserProfile.user_id == id).first() + if profile: + for field, value in profile_update.items(): + setattr(profile, field, value) + + db.commit() + db.refresh(user) + + return user + + +def delete_user(db: Session, id: str): + user = db.query(User).filter(User.id == id, User.deleted_at.is_(None)).first() + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={"message": "The item does not exist"}, + ) + + user.deleted_at = datetime.now() + db.commit() diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..74b886a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,45 @@ +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.database import Base, engine, get_db +from app.models import * # noqa: F401, F403 — register all models +from main import app + + +@pytest.fixture(scope="session") +def setup_db(): + Base.metadata.create_all(bind=engine) + yield + + +@pytest.fixture +def db_session(setup_db): + connection = engine.connect() + transaction = connection.begin() + + connection.execute(text("SET FOREIGN_KEY_CHECKS = 0")) + for table in reversed(Base.metadata.sorted_tables): + connection.execute(text(f"TRUNCATE TABLE {table.name}")) + connection.execute(text("SET FOREIGN_KEY_CHECKS = 1")) + transaction.commit() + + transaction = connection.begin() + session = Session(bind=connection) + + yield session + + session.close() + transaction.rollback() + connection.close() + + +@pytest.fixture +def client(db_session): + def override_get_db(): + yield db_session + + app.dependency_overrides[get_db] = override_get_db + yield TestClient(app) + app.dependency_overrides.clear() diff --git a/tests/test_business_type.py b/tests/test_business_type.py new file mode 100644 index 0000000..0175b33 --- /dev/null +++ b/tests/test_business_type.py @@ -0,0 +1,117 @@ +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models.business_type import BusinessType +from app.schemas.business_type import BusinessTypeCreate, BusinessTypeUpdate +from app.services.business_type_service import ( + create_business_type, + delete_business_type, + get_business_type_by_id, + get_business_type_list, + update_business_type, +) + + +class TestService: + def test_list_empty(self, db_session: Session): + assert get_business_type_list(db_session) == [] + + def test_create(self, db_session: Session): + data = BusinessTypeCreate(code="RETAIL", name="Retail") + bt = create_business_type(db_session, data) + assert bt.code == "RETAIL" + assert bt.name == "Retail" + + def test_create_duplicate_code(self, db_session: Session): + create_business_type(db_session, BusinessTypeCreate(code="RETAIL", name="A")) + with pytest.raises(HTTPException) as exc: + create_business_type(db_session, BusinessTypeCreate(code="RETAIL", name="B")) + assert exc.value.status_code == 409 + + def test_get_by_id(self, db_session: Session): + bt = create_business_type(db_session, BusinessTypeCreate(code="GROCERY", name="Grocery")) + result = get_business_type_by_id(db_session, bt.id) + assert result.id == bt.id + + def test_get_by_id_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + get_business_type_by_id(db_session, "nonexistent") + assert exc.value.status_code == 404 + + def test_update(self, db_session: Session): + bt = create_business_type(db_session, BusinessTypeCreate(code="CAFE", name="Cafe")) + data = BusinessTypeUpdate(name="Coffee Shop") + updated = update_business_type(db_session, bt.id, data) + assert updated.name == "Coffee Shop" + + def test_update_conflict_code(self, db_session: Session): + create_business_type(db_session, BusinessTypeCreate(code="A", name="A")) + bt = create_business_type(db_session, BusinessTypeCreate(code="B", name="B")) + with pytest.raises(HTTPException) as exc: + update_business_type(db_session, bt.id, BusinessTypeUpdate(code="A")) + assert exc.value.status_code == 409 + + def test_update_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + update_business_type(db_session, "x", BusinessTypeUpdate(name="X")) + assert exc.value.status_code == 404 + + def test_delete(self, db_session: Session): + bt = create_business_type(db_session, BusinessTypeCreate(code="DEL", name="Del")) + delete_business_type(db_session, bt.id) + assert bt.deleted_at is not None + + def test_delete_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + delete_business_type(db_session, "x") + assert exc.value.status_code == 404 + + def test_soft_deleted_excluded(self, db_session: Session): + create_business_type(db_session, BusinessTypeCreate(code="KEEP", name="Keep")) + bt = create_business_type(db_session, BusinessTypeCreate(code="GONE", name="Gone")) + delete_business_type(db_session, bt.id) + assert len(get_business_type_list(db_session)) == 1 + + +class TestAPI: + def test_create(self, client: TestClient): + resp = client.post("/v1/business-types/", json={"code": "TOKO", "name": "Toko"}) + assert resp.status_code == 201 + assert resp.json() == {"message": "The item was created successfully"} + + def test_list(self, client: TestClient, db_session: Session): + db_session.add(BusinessType(code="A", name="A")) + db_session.commit() + resp = client.get("/v1/business-types/") + assert resp.status_code == 200 + assert len(resp.json()) == 1 + + def test_get_by_id(self, client: TestClient, db_session: Session): + bt = BusinessType(code="GET", name="Get") + db_session.add(bt) + db_session.commit() + resp = client.get(f"/v1/business-types/{bt.id}") + assert resp.status_code == 200 + assert resp.json()["code"] == "GET" + + def test_get_not_found(self, client: TestClient): + resp = client.get("/v1/business-types/x") + assert resp.status_code == 404 + assert resp.json() == {"message": "The item does not exist"} + + def test_update(self, client: TestClient, db_session: Session): + bt = BusinessType(code="UPD", name="Old") + db_session.add(bt) + db_session.commit() + resp = client.put(f"/v1/business-types/{bt.id}", json={"name": "New"}) + assert resp.status_code == 200 + assert resp.json()["name"] == "New" + + def test_delete(self, client: TestClient, db_session: Session): + bt = BusinessType(code="DEL", name="Del") + db_session.add(bt) + db_session.commit() + resp = client.delete(f"/v1/business-types/{bt.id}") + assert resp.status_code == 204 diff --git a/tests/test_plan.py b/tests/test_plan.py new file mode 100644 index 0000000..3c3ffad --- /dev/null +++ b/tests/test_plan.py @@ -0,0 +1,109 @@ +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models.plan import Plan +from app.schemas.plan import PlanCreate, PlanUpdate +from app.services.plan_service import ( + create_plan, + delete_plan, + get_plan_by_id, + get_plan_list, + update_plan, +) + + +class TestService: + def test_list_empty(self, db_session: Session): + assert get_plan_list(db_session) == [] + + def test_create(self, db_session: Session): + data = PlanCreate(code="BASIC", name="Basic Plan", price=100000) + plan = create_plan(db_session, data) + assert plan.code == "BASIC" + assert plan.price == 100000 + + def test_create_duplicate_code(self, db_session: Session): + create_plan(db_session, PlanCreate(code="BASIC", name="A", price=0)) + with pytest.raises(HTTPException) as exc: + create_plan(db_session, PlanCreate(code="BASIC", name="B", price=0)) + assert exc.value.status_code == 409 + + def test_get_by_id(self, db_session: Session): + plan = create_plan(db_session, PlanCreate(code="PRO", name="Pro", price=50000)) + result = get_plan_by_id(db_session, plan.id) + assert result.id == plan.id + + def test_get_by_id_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + get_plan_by_id(db_session, "x") + assert exc.value.status_code == 404 + + def test_update(self, db_session: Session): + plan = create_plan(db_session, PlanCreate(code="VIP", name="Vip", price=200)) + updated = update_plan(db_session, plan.id, PlanUpdate(price=300)) + assert updated.price == 300 + + def test_update_conflict_code(self, db_session: Session): + create_plan(db_session, PlanCreate(code="A", name="A", price=0)) + plan = create_plan(db_session, PlanCreate(code="B", name="B", price=0)) + with pytest.raises(HTTPException) as exc: + update_plan(db_session, plan.id, PlanUpdate(code="A")) + assert exc.value.status_code == 409 + + def test_delete(self, db_session: Session): + plan = create_plan(db_session, PlanCreate(code="DEL", name="Del", price=0)) + delete_plan(db_session, plan.id) + assert plan.deleted_at is not None + + def test_delete_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + delete_plan(db_session, "x") + assert exc.value.status_code == 404 + + def test_soft_deleted_excluded(self, db_session: Session): + create_plan(db_session, PlanCreate(code="KEEP", name="Keep", price=0)) + plan = create_plan(db_session, PlanCreate(code="GONE", name="Gone", price=0)) + delete_plan(db_session, plan.id) + assert len(get_plan_list(db_session)) == 1 + + +class TestAPI: + def test_create(self, client: TestClient): + resp = client.post("/v1/plans/", json={"code": "BASIC", "name": "Basic", "price": 50000}) + assert resp.status_code == 201 + + def test_list(self, client: TestClient, db_session: Session): + db_session.add(Plan(code="A", name="A", price=0)) + db_session.commit() + resp = client.get("/v1/plans/") + assert resp.status_code == 200 + assert len(resp.json()) == 1 + + def test_get_by_id(self, client: TestClient, db_session: Session): + plan = Plan(code="GET", name="Get", price=100) + db_session.add(plan) + db_session.commit() + resp = client.get(f"/v1/plans/{plan.id}") + assert resp.status_code == 200 + + def test_not_found(self, client: TestClient): + resp = client.get("/v1/plans/x") + assert resp.status_code == 404 + assert resp.json() == {"message": "The item does not exist"} + + def test_update(self, client: TestClient, db_session: Session): + plan = Plan(code="UPD", name="Old", price=100) + db_session.add(plan) + db_session.commit() + resp = client.put(f"/v1/plans/{plan.id}", json={"price": 999}) + assert resp.status_code == 200 + assert resp.json()["price"] == 999 + + def test_delete(self, client: TestClient, db_session: Session): + plan = Plan(code="DEL", name="Del", price=0) + db_session.add(plan) + db_session.commit() + resp = client.delete(f"/v1/plans/{plan.id}") + assert resp.status_code == 204 diff --git a/tests/test_subscription.py b/tests/test_subscription.py new file mode 100644 index 0000000..cbe9cbd --- /dev/null +++ b/tests/test_subscription.py @@ -0,0 +1,126 @@ +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models.business_type import BusinessType +from app.models.plan import Plan +from app.models.subscription import Subscription +from app.models.tenant import Tenant +from app.schemas.subscription import SubscriptionCreate, SubscriptionUpdate +from app.services.subscription_service import ( + create_subscription, + delete_subscription, + get_subscription_by_id, + get_subscription_list, + update_subscription, +) + + +@pytest.fixture +def bt(db_session: Session): + bt = BusinessType(code="SUBT", name="Sub Test") + db_session.add(bt) + db_session.flush() + return bt + + +@pytest.fixture +def tenant(db_session: Session, bt): + t = Tenant(name="T", slug="t", business_type_id=bt.id) + db_session.add(t) + db_session.flush() + return t + + +@pytest.fixture +def plan(db_session: Session): + p = Plan(code="SUB", name="Sub", price=0) + db_session.add(p) + db_session.flush() + return p + + +class TestService: + def test_create(self, db_session: Session, tenant, plan): + sub = create_subscription(db_session, SubscriptionCreate(tenant_id=tenant.id, plan_id=plan.id)) + assert sub.tenant_id == tenant.id + assert sub.plan_id == plan.id + assert sub.status == "TRIAL" + + def test_create_duplicate_tenant(self, db_session: Session, tenant, plan): + create_subscription(db_session, SubscriptionCreate(tenant_id=tenant.id, plan_id=plan.id)) + with pytest.raises(HTTPException) as exc: + create_subscription(db_session, SubscriptionCreate(tenant_id=tenant.id, plan_id=plan.id)) + assert exc.value.status_code == 409 + + def test_get_by_id(self, db_session: Session, tenant, plan): + sub = create_subscription(db_session, SubscriptionCreate(tenant_id=tenant.id, plan_id=plan.id)) + result = get_subscription_by_id(db_session, sub.id) + assert result.id == sub.id + + def test_get_by_id_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + get_subscription_by_id(db_session, "x") + assert exc.value.status_code == 404 + + def test_list(self, db_session: Session, tenant, plan, bt): + t2 = Tenant(name="T2", slug="t2", business_type_id=bt.id) + db_session.add(t2) + db_session.flush() + create_subscription(db_session, SubscriptionCreate(tenant_id=tenant.id, plan_id=plan.id)) + create_subscription(db_session, SubscriptionCreate(tenant_id=t2.id, plan_id=plan.id)) + assert len(get_subscription_list(db_session)) == 2 + + def test_update(self, db_session: Session, tenant, plan): + sub = create_subscription(db_session, SubscriptionCreate(tenant_id=tenant.id, plan_id=plan.id)) + updated = update_subscription(db_session, sub.id, SubscriptionUpdate(status="ACTIVE")) + assert updated.status == "ACTIVE" + + def test_update_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + update_subscription(db_session, "x", SubscriptionUpdate(status="A")) + assert exc.value.status_code == 404 + + def test_delete(self, db_session: Session, tenant, plan): + sub = create_subscription(db_session, SubscriptionCreate(tenant_id=tenant.id, plan_id=plan.id)) + delete_subscription(db_session, sub.id) + assert db_session.query(Subscription).filter(Subscription.id == sub.id).first() is None + + def test_delete_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + delete_subscription(db_session, "x") + assert exc.value.status_code == 404 + + +class TestAPI: + def test_create(self, client: TestClient, db_session: Session, tenant, plan): + resp = client.post("/v1/subscriptions/", json={ + "tenant_id": tenant.id, "plan_id": plan.id, + }) + assert resp.status_code == 201 + + def test_list(self, client: TestClient, db_session: Session, tenant, plan): + db_session.add(Subscription(tenant_id=tenant.id, plan_id=plan.id)) + db_session.commit() + resp = client.get("/v1/subscriptions/") + assert resp.status_code == 200 + assert len(resp.json()) == 1 + + def test_get_by_id(self, client: TestClient, db_session: Session, tenant, plan): + sub = Subscription(tenant_id=tenant.id, plan_id=plan.id) + db_session.add(sub) + db_session.commit() + resp = client.get(f"/v1/subscriptions/{sub.id}") + assert resp.status_code == 200 + + def test_not_found(self, client: TestClient): + resp = client.get("/v1/subscriptions/x") + assert resp.status_code == 404 + + def test_delete(self, client: TestClient, db_session: Session, tenant, plan): + sub = Subscription(tenant_id=tenant.id, plan_id=plan.id) + db_session.add(sub) + db_session.commit() + resp = client.delete(f"/v1/subscriptions/{sub.id}") + assert resp.status_code == 204 diff --git a/tests/test_tenant.py b/tests/test_tenant.py new file mode 100644 index 0000000..4782863 --- /dev/null +++ b/tests/test_tenant.py @@ -0,0 +1,142 @@ +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models.business_type import BusinessType +from app.models.plan import Plan +from app.models.subscription import Subscription +from app.models.tenant import Tenant +from app.models.user import User +from app.schemas.tenant import TenantCreate, TenantUpdate +from app.services.tenant_service import ( + assign_user_to_tenant, + create_tenant, + delete_tenant, + get_tenant_by_id, + get_tenant_list, + update_tenant, +) + + +@pytest.fixture +def bt(db_session: Session): + bt = BusinessType(code="TEST", name="Test") + db_session.add(bt) + db_session.flush() + return bt + + +@pytest.fixture +def plan(db_session: Session): + plan = Plan(code="TEST", name="Test", price=0) + db_session.add(plan) + db_session.flush() + return plan + + +class TestService: + def test_create(self, db_session: Session, bt, plan): + data = TenantCreate( + name="Toko Saya", slug="toko-saya", + business_type_id=bt.id, plan_id=plan.id, + ) + tenant = create_tenant(db_session, data) + assert tenant.name == "Toko Saya" + assert tenant.slug == "toko-saya" + + sub = db_session.query(Subscription).filter(Subscription.tenant_id == tenant.id).first() + assert sub is not None + + def test_create_duplicate_slug(self, db_session: Session, bt, plan): + create_tenant(db_session, TenantCreate(name="A", slug="same", business_type_id=bt.id, plan_id=plan.id)) + with pytest.raises(HTTPException) as exc: + create_tenant(db_session, TenantCreate(name="B", slug="same", business_type_id=bt.id, plan_id=plan.id)) + assert exc.value.status_code == 409 + + def test_get_by_id(self, db_session: Session, bt, plan): + tenant = create_tenant(db_session, TenantCreate(name="Get", slug="get", business_type_id=bt.id, plan_id=plan.id)) + result = get_tenant_by_id(db_session, tenant.id) + assert result.id == tenant.id + + def test_get_by_id_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + get_tenant_by_id(db_session, "x") + assert exc.value.status_code == 404 + + def test_list(self, db_session: Session, bt, plan): + create_tenant(db_session, TenantCreate(name="A", slug="a", business_type_id=bt.id, plan_id=plan.id)) + create_tenant(db_session, TenantCreate(name="B", slug="b", business_type_id=bt.id, plan_id=plan.id)) + assert len(get_tenant_list(db_session)) == 2 + + def test_update(self, db_session: Session, bt, plan): + tenant = create_tenant(db_session, TenantCreate(name="Old", slug="old", business_type_id=bt.id, plan_id=plan.id)) + updated = update_tenant(db_session, tenant.id, TenantUpdate(name="New")) + assert updated.name == "New" + + def test_update_conflict_slug(self, db_session: Session, bt, plan): + create_tenant(db_session, TenantCreate(name="A", slug="taken", business_type_id=bt.id, plan_id=plan.id)) + tenant = create_tenant(db_session, TenantCreate(name="B", slug="b", business_type_id=bt.id, plan_id=plan.id)) + with pytest.raises(HTTPException) as exc: + update_tenant(db_session, tenant.id, TenantUpdate(slug="taken")) + assert exc.value.status_code == 409 + + def test_delete(self, db_session: Session, bt, plan): + tenant = create_tenant(db_session, TenantCreate(name="Del", slug="del", business_type_id=bt.id, plan_id=plan.id)) + delete_tenant(db_session, tenant.id) + assert tenant.deleted_at is not None + + def test_assign_user(self, db_session: Session, bt, plan): + tenant = create_tenant(db_session, TenantCreate(name="T", slug="t", business_type_id=bt.id, plan_id=plan.id)) + user = User(email="u@ex.com", username="u", password="x") + db_session.add(user) + db_session.commit() + + result = assign_user_to_tenant(db_session, tenant.id, user.id) + assert result["message"] == "User assigned to tenant" + assert user.tenant_id == tenant.id + assert user.status == "ACTIVE" + + def test_assign_user_tenant_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + assign_user_to_tenant(db_session, "x", "y") + assert exc.value.status_code == 404 + + +class TestAPI: + def test_create(self, client: TestClient, db_session: Session, bt, plan): + resp = client.post("/v1/tenants/", json={ + "name": "Toko", "slug": "toko", + "business_type_id": bt.id, "plan_id": plan.id, + }) + assert resp.status_code == 201 + + def test_list(self, client: TestClient, db_session: Session, bt): + db_session.add(Tenant(name="T", slug="t", business_type_id=bt.id)) + db_session.commit() + resp = client.get("/v1/tenants/") + assert resp.status_code == 200 + + def test_not_found(self, client: TestClient): + resp = client.get("/v1/tenants/x") + assert resp.status_code == 404 + + def test_delete(self, client: TestClient, db_session: Session, bt, plan): + from app.models.subscription import Subscription + tenant = Tenant(name="D", slug="d", business_type_id=bt.id) + db_session.add(tenant) + db_session.flush() + db_session.add(Subscription(tenant_id=tenant.id, plan_id=plan.id)) + db_session.commit() + resp = client.delete(f"/v1/tenants/{tenant.id}") + assert resp.status_code == 204 + + def test_assign_user(self, client: TestClient, db_session: Session, bt, plan): + tenant = Tenant(name="A", slug="a", business_type_id=bt.id) + db_session.add(tenant) + user = User(email="assign@ex.com", username="assign", password="x") + db_session.add(user) + db_session.commit() + resp = client.post(f"/v1/tenants/{tenant.id}/assign-user?user_id={user.id}") + assert resp.status_code == 200 + assert resp.json()["message"] == "User assigned to tenant" diff --git a/tests/test_user_api.py b/tests/test_user_api.py new file mode 100644 index 0000000..0c0cbde --- /dev/null +++ b/tests/test_user_api.py @@ -0,0 +1,125 @@ +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models.user import User +from app.models.user_profile import UserProfile + + +class TestListUsers: + def test_empty(self, client: TestClient): + resp = client.get("/v1/users/") + assert resp.status_code == 200 + assert resp.json() == [] + assert resp.headers.get("Pagination-Count") == "0" + assert resp.headers.get("Pagination-Page") == "1" + assert resp.headers.get("Pagination-Limit") == "10" + + def test_pagination_headers(self, client: TestClient, db_session: Session): + for i in range(5): + db_session.add(User(email=f"u{i}@ex.com", username=f"u{i}", password="x")) + db_session.commit() + + resp = client.get("/v1/users/?page=1&limit=2") + data = resp.json() + assert len(data) == 2 + assert resp.headers["Pagination-Count"] == "5" + assert resp.headers["Pagination-Page"] == "1" + assert resp.headers["Pagination-Limit"] == "2" + + def test_includes_profile(self, client: TestClient, db_session: Session): + user = User(email="prof@ex.com", username="prof", password="x") + db_session.add(user) + db_session.flush() + db_session.add(UserProfile(user_id=user.id, full_name="Profile Test")) + db_session.commit() + + resp = client.get("/v1/users/") + data = resp.json() + assert data[0]["profile"]["full_name"] == "Profile Test" + + +class TestGetUser: + def test_found(self, client: TestClient, db_session: Session): + user = User(email="get@ex.com", username="get", password="x") + db_session.add(user) + db_session.commit() + + resp = client.get(f"/v1/users/{user.id}") + assert resp.status_code == 200 + assert resp.json()["email"] == "get@ex.com" + + def test_detail_response(self, client: TestClient, db_session: Session): + user = User(email="det@ex.com", username="det", password="x") + db_session.add(user) + db_session.flush() + db_session.add(UserProfile(user_id=user.id, full_name="Detail Name")) + db_session.commit() + + resp = client.get(f"/v1/users/{user.id}") + assert resp.json()["profile"]["full_name"] == "Detail Name" + + def test_not_found(self, client: TestClient): + resp = client.get("/v1/users/nonexistent") + assert resp.status_code == 404 + assert resp.json() == {"message": "The item does not exist"} + + +class TestCreateUser: + def test_created(self, client: TestClient): + payload = { + "email": "new@ex.com", + "username": "newuser", + "password": "secret", + "full_name": "New User", + } + resp = client.post("/v1/users/", json=payload) + assert resp.status_code == 201 + assert resp.json() == {"message": "The item was created successfully"} + assert resp.headers.get("Location") is not None + + def test_duplicate(self, client: TestClient, db_session: Session): + db_session.add(User(email="dup@ex.com", username="dup", password="x")) + db_session.commit() + + payload = { + "email": "dup@ex.com", + "username": "dup", + "password": "x", + "full_name": "Dup", + } + resp = client.post("/v1/users/", json=payload) + assert resp.status_code == 409 + assert resp.json() == {"message": "Email or username already registered"} + + +class TestUpdateUser: + def test_update(self, client: TestClient, db_session: Session): + user = User(email="upd@ex.com", username="upd", password="x") + db_session.add(user) + db_session.flush() + db_session.add(UserProfile(user_id=user.id, full_name="Original")) + db_session.commit() + + resp = client.put(f"/v1/users/{user.id}", json={"full_name": "Updated"}) + assert resp.status_code == 200 + assert resp.json()["profile"]["full_name"] == "Updated" + + def test_not_found(self, client: TestClient): + resp = client.put("/v1/users/nonexistent", json={"full_name": "X"}) + assert resp.status_code == 404 + assert resp.json() == {"message": "The item does not exist"} + + +class TestDeleteUser: + def test_deleted(self, client: TestClient, db_session: Session): + user = User(email="del@ex.com", username="del", password="x") + db_session.add(user) + db_session.commit() + + resp = client.delete(f"/v1/users/{user.id}") + assert resp.status_code == 204 + + def test_not_found(self, client: TestClient): + resp = client.delete("/v1/users/nonexistent") + assert resp.status_code == 404 + assert resp.json() == {"message": "The item does not exist"} diff --git a/tests/test_user_service.py b/tests/test_user_service.py new file mode 100644 index 0000000..5527a15 --- /dev/null +++ b/tests/test_user_service.py @@ -0,0 +1,201 @@ +import pytest +from fastapi import HTTPException +from sqlalchemy.orm import Session + +from app.models.user import User +from app.models.user_profile import UserProfile +from app.schemas.user import UserCreate, UserUpdate +from app.security import verify_password +from app.services.user_service import ( + create_user, + delete_user, + get_user_by_id, + get_user_list, + update_user, +) + + +class TestGetUserList: + def test_empty(self, db_session: Session): + users, total = get_user_list(db_session, page=1, limit=10) + assert users == [] + assert total == 0 + + def test_with_users(self, db_session: Session): + for i in range(3): + user = User(email=f"test{i}@ex.com", username=f"user{i}", password="x") + db_session.add(user) + db_session.commit() + + users, total = get_user_list(db_session, page=1, limit=10) + assert total == 3 + assert len(users) == 3 + + def test_pagination(self, db_session: Session): + for i in range(5): + user = User(email=f"page{i}@ex.com", username=f"page{i}", password="x") + db_session.add(user) + db_session.commit() + + users, total = get_user_list(db_session, page=1, limit=2) + assert total == 5 + assert len(users) == 2 + + def test_soft_deleted_excluded(self, db_session: Session): + from datetime import datetime + + u1 = User(email="active@ex.com", username="active", password="x") + u2 = User(email="deleted@ex.com", username="deleted", password="x", deleted_at=datetime.now()) + db_session.add_all([u1, u2]) + db_session.commit() + + users, total = get_user_list(db_session, page=1, limit=10) + assert total == 1 + assert users[0].email == "active@ex.com" + + +class TestGetUserById: + def test_found(self, db_session: Session): + user = User(email="find@ex.com", username="find", password="x") + db_session.add(user) + db_session.commit() + + result = get_user_by_id(db_session, user.id) + assert result.id == user.id + assert result.email == "find@ex.com" + + def test_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + get_user_by_id(db_session, "nonexistent") + assert exc.value.status_code == 404 + + def test_soft_deleted(self, db_session: Session): + from datetime import datetime + + user = User(email="gone@ex.com", username="gone", password="x", deleted_at=datetime.now()) + db_session.add(user) + db_session.commit() + + with pytest.raises(HTTPException) as exc: + get_user_by_id(db_session, user.id) + assert exc.value.status_code == 404 + + +class TestCreateUser: + def test_create(self, db_session: Session): + data = UserCreate( + email="new@ex.com", + username="newuser", + password="secret", + full_name="New User", + ) + user = create_user(db_session, data) + + assert user.email == "new@ex.com" + assert user.username == "newuser" + assert verify_password("secret", user.password) + + profile = db_session.query(UserProfile).filter(UserProfile.user_id == user.id).first() + assert profile is not None + assert profile.full_name == "New User" + + def test_duplicate_email(self, db_session: Session): + db_session.add(User(email="dup@ex.com", username="first", password="x")) + db_session.commit() + data = UserCreate( + email="dup@ex.com", + username="second", + password="x", + full_name="Dup", + ) + with pytest.raises(HTTPException) as exc: + create_user(db_session, data) + assert exc.value.status_code == 409 + + def test_duplicate_username(self, db_session: Session): + data = UserCreate( + email="a@ex.com", + username="taken", + password="x", + full_name="A", + ) + create_user(db_session, data) + + dup = UserCreate( + email="b@ex.com", + username="taken", + password="x", + full_name="B", + ) + with pytest.raises(HTTPException) as exc: + create_user(db_session, dup) + assert exc.value.status_code == 409 + + +class TestUpdateUser: + def test_update_email(self, db_session: Session): + user = User(email="old@ex.com", username="old", password="x") + db_session.add(user) + db_session.commit() + + data = UserUpdate(email="new@ex.com") + updated = update_user(db_session, user.id, data) + + assert updated.email == "new@ex.com" + + def test_update_profile(self, db_session: Session): + user = User(email="prof@ex.com", username="prof", password="x") + db_session.add(user) + db_session.flush() + + profile = UserProfile(user_id=user.id, full_name="Old Name") + db_session.add(profile) + db_session.commit() + + data = UserUpdate(full_name="Updated Name") + updated = update_user(db_session, user.id, data) + + assert updated.profile.full_name == "Updated Name" + + def test_update_password(self, db_session: Session): + user = User(email="pw@ex.com", username="pw", password="x") + db_session.add(user) + db_session.commit() + + data = UserUpdate(password="newsecret") + update_user(db_session, user.id, data) + + db_session.refresh(user) + assert verify_password("newsecret", user.password) + + def test_update_not_found(self, db_session: Session): + data = UserUpdate(email="nope@ex.com") + with pytest.raises(HTTPException) as exc: + update_user(db_session, "nonexistent", data) + assert exc.value.status_code == 404 + + def test_update_conflict_email(self, db_session: Session): + User(email="existing@ex.com", username="existing", password="x") + user = User(email="me@ex.com", username="me", password="x") + db_session.add_all([User(email="existing@ex.com", username="existing", password="x"), user]) + db_session.commit() + + data = UserUpdate(email="existing@ex.com") + with pytest.raises(HTTPException) as exc: + update_user(db_session, user.id, data) + assert exc.value.status_code == 409 + + +class TestDeleteUser: + def test_soft_delete(self, db_session: Session): + user = User(email="del@ex.com", username="del", password="x") + db_session.add(user) + db_session.commit() + + delete_user(db_session, user.id) + assert user.deleted_at is not None + + def test_delete_not_found(self, db_session: Session): + with pytest.raises(HTTPException) as exc: + delete_user(db_session, "nonexistent") + assert exc.value.status_code == 404