LeadPilot / backend /tests /test_workspaces.py
Ashraf Al-Kassem
fix: add missing backend application code (app/, migrations, tests, templates)
8ca4657
raw
history blame
4.87 kB
import pytest
from httpx import AsyncClient
async def get_auth_headers(client: AsyncClient, email: str) -> dict:
# Helper to signup/login and get headers
pwd = "password123"
await client.post("/api/v1/auth/signup", json={"email": email, "password": pwd, "full_name": "WS Test"})
login_res = await client.post("/api/v1/auth/login", data={"username": email, "password": pwd})
token = login_res.json()["data"]["access_token"]
# Decode token to get default workspace (or just fetch user to find it if needed, but Login returns context usually?)
# For now, let's just use the token. The backend often infers workspace or requires header.
# We need the workspace ID.
# Let's hit /auth/me or similar if exists, or use the token payload logic if we could decode.
# Better: Signup returns the user, we can assume default workspace is created.
# Let's inspect the side-effects or just use a known endpoint to list workspaces.
# Assuming we need to find the workspace ID.
# Let's try listing workspaces if that endpoint exists, or just create a fresh one.
# Or, simpler: The LOGIN endpoint in this app returns a token scoped to a workspace?
# Checking auth.py: login returns token.
# Let's use `GET /workspaces` (assuming it exists, usually does in this stack).
# Wait, strict checking: we need the workspace ID to trigger X-Workspace-ID checks.
# We'll just assume there is a /api/v1/workspaces endpoint based on standard patterns.
# If not, we'll fail and fix.
ws_res = await client.get("/api/v1/workspaces", headers={"Authorization": f"Bearer {token}"})
if ws_res.status_code == 200:
ws_id = ws_res.json()["data"][0]["id"] # list of workspaces
return {"Authorization": f"Bearer {token}", "X-Workspace-ID": ws_id}
return {"Authorization": f"Bearer {token}"}
@pytest.mark.asyncio
async def test_workspace_isolation(async_client: AsyncClient):
# User 1
headers1 = await get_auth_headers(async_client, "user1@example.com")
# User 2
headers2 = await get_auth_headers(async_client, "user2@example.com")
# User 1 creates a Prompt Config
pc_payload = {
"structured_data": {"basics": {"business_name": "User 1 Bot"}},
"temperature": 0.5
}
create_res = await async_client.post("/api/v1/prompt-config/", json=pc_payload, headers=headers1)
# Note: If /prompt-config doesn't exist, this will fail (we'll implement the test anyway as requested)
# Based on user request "prompt-config create/read", it should exist.
assert create_res.status_code == 200
# Response is Version, not Config
# We don't get config ID directly in data.
# But we can get it via GET /
# User 2 tries to read Config
# The endpoint is singleton per workspace: GET /api/v1/prompt-config/
# So User 2 calling GET / should returned User 2's empty config or default, NOT User 1's.
# Actually, verify User 2 cannot access User 1's data via ID if there was ID based access.
# But current design is Singleton.
# So test is: User 2 calls GET / and sees DIFFERENT data (or default) than User 1.
# Let's verify User 1 sees what they created.
get1 = await async_client.get("/api/v1/prompt-config/", headers=headers1)
v1_data = get1.json()["data"]["versions"][0]
assert v1_data["structured_data"]["basics"]["business_name"] == "User 1 Bot"
# User 2 calls GET /
get2 = await async_client.get("/api/v1/prompt-config/", headers=headers2)
assert get2.status_code == 200
data2 = get2.json()["data"]
# Should be empty or default
if data2["versions"]:
assert data2["versions"][0]["structured_data"].get("basics", {}).get("business_name") != "User 1 Bot"
else:
assert True # Empty versions means clean slate
# If there was an ID based endpoint: GET /prompt-config/{id}
# It seems from audit/code viewing, prompt_config endpoint is singleton "/".
# So "Isolation" means I see my config, you see yours. Verified above.
# If there was an ID based endpoint: GET /prompt-config/{id}
# It seems from audit/code viewing, prompt_config endpoint is singleton "/".
# So "Isolation" means I see my config, you see yours. Verified above.
@pytest.mark.asyncio
async def test_invite_member(async_client: AsyncClient):
# User 1 (Owner)
headers = await get_auth_headers(async_client, "owner@example.com")
# Switch context if get_auth_headers returns X-Workspace-ID
# It does based on my read of previous file content
payload = {"email": "invited@example.com", "role": "member"}
res = await async_client.post("/api/v1/workspaces/members/invite", json=payload, headers=headers)
assert res.status_code == 200
assert "Invite sent" in res.json()["data"]["message"]