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.
126 lines
4.6 KiB
Python
126 lines
4.6 KiB
Python
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"}
|