7bbe4e7295
- 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.
110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
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
|