Implement subscription, tenant, and user services with CRUD operations and associated tests
- Added subscription_service.py for managing subscriptions with functions to create, read, update, and delete subscriptions. - Added tenant_service.py for managing tenants with functions to create, read, update, delete tenants, and assign users to tenants. - Added user_service.py for managing users with functions to create, read, update, and delete users. - Created tests for business types, plans, subscriptions, tenants, and users, ensuring proper functionality and error handling. - Implemented pagination in user listing and included user profiles in responses. - Established fixtures for database setup and teardown in tests.
This commit is contained in:
Yoga Pangestu
parent
717ec1e061
commit
7bbe4e7295
15
app/routers/__init__.py
Normal file
15
app/routers/__init__.py
Normal file
@ -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",
|
||||
]
|
||||
46
app/routers/auth.py
Normal file
46
app/routers/auth.py
Normal file
@ -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
|
||||
50
app/routers/business_types.py
Normal file
50
app/routers/business_types.py
Normal file
@ -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)
|
||||
42
app/routers/plans.py
Normal file
42
app/routers/plans.py
Normal file
@ -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)
|
||||
50
app/routers/subscriptions.py
Normal file
50
app/routers/subscriptions.py
Normal file
@ -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)
|
||||
50
app/routers/tenants.py
Normal file
50
app/routers/tenants.py
Normal file
@ -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)
|
||||
58
app/routers/users.py
Normal file
58
app/routers/users.py
Normal file
@ -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)
|
||||
@ -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",
|
||||
]
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
0
app/services/__init__.py
Normal file
0
app/services/__init__.py
Normal file
74
app/services/business_type_service.py
Normal file
74
app/services/business_type_service.py
Normal file
@ -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()
|
||||
74
app/services/plan_service.py
Normal file
74
app/services/plan_service.py
Normal file
@ -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()
|
||||
65
app/services/subscription_service.py
Normal file
65
app/services/subscription_service.py
Normal file
@ -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()
|
||||
118
app/services/tenant_service.py
Normal file
118
app/services/tenant_service.py
Normal file
@ -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"}
|
||||
120
app/services/user_service.py
Normal file
120
app/services/user_service.py
Normal file
@ -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()
|
||||
0
tests/__init__.py
Normal file
0
tests/__init__.py
Normal file
45
tests/conftest.py
Normal file
45
tests/conftest.py
Normal file
@ -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()
|
||||
117
tests/test_business_type.py
Normal file
117
tests/test_business_type.py
Normal file
@ -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
|
||||
109
tests/test_plan.py
Normal file
109
tests/test_plan.py
Normal file
@ -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
|
||||
126
tests/test_subscription.py
Normal file
126
tests/test_subscription.py
Normal file
@ -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
|
||||
142
tests/test_tenant.py
Normal file
142
tests/test_tenant.py
Normal file
@ -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"
|
||||
125
tests/test_user_api.py
Normal file
125
tests/test_user_api.py
Normal file
@ -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"}
|
||||
201
tests/test_user_service.py
Normal file
201
tests/test_user_service.py
Normal file
@ -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
|
||||
Loading…
Reference in New Issue
Block a user