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.
127 lines
4.9 KiB
Python
127 lines
4.9 KiB
Python
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
|