Replace team_id with account_id across all API endpoints (trees, categories, tags, steps, step_categories, admin, auth). Add new accounts and webhooks endpoints. Registration now atomically creates Account + Subscription, with account_invite_code bypassing the platform invite gate. Schemas updated for account_id/account_role. 82 tests passing including 18 new tests for accounts, subscriptions, and permissions. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
206 lines
8.1 KiB
Python
206 lines
8.1 KiB
Python
"""Integration tests for account-based permissions."""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
class TestAccountPermissions:
|
|
"""Test suite for account-based permission checks."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_viewer_cannot_create_tree(self, client: AsyncClient, test_db):
|
|
"""Test that viewers cannot create trees."""
|
|
from sqlalchemy import select
|
|
from app.models.user import User
|
|
from uuid import UUID as PyUUID
|
|
|
|
# Register a user
|
|
reg_resp = await client.post("/api/v1/auth/register", json={
|
|
"email": "viewer@example.com",
|
|
"password": "ViewerPass123!",
|
|
"name": "Viewer User"
|
|
})
|
|
assert reg_resp.status_code == 201
|
|
user_id = PyUUID(reg_resp.json()["id"])
|
|
|
|
# Demote to viewer via ORM
|
|
result = await test_db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one()
|
|
user.account_role = "viewer"
|
|
await test_db.commit()
|
|
|
|
# Login as viewer
|
|
login_resp = await client.post("/api/v1/auth/login/json", json={
|
|
"email": "viewer@example.com",
|
|
"password": "ViewerPass123!"
|
|
})
|
|
viewer_headers = {"Authorization": f"Bearer {login_resp.json()['access_token']}"}
|
|
|
|
# Try to create tree
|
|
response = await client.post("/api/v1/trees", json={
|
|
"name": "Viewer Tree",
|
|
"tree_structure": {"id": "root", "type": "solution", "title": "Test", "description": "Test"}
|
|
}, headers=viewer_headers)
|
|
assert response.status_code == 403
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_viewer_can_list_trees(self, client: AsyncClient, auth_headers: dict, test_db):
|
|
"""Test that viewers can browse/list trees."""
|
|
from sqlalchemy import select
|
|
from app.models.user import User
|
|
from uuid import UUID as PyUUID
|
|
|
|
# Create a public tree as the regular user first
|
|
await client.post("/api/v1/trees", json={
|
|
"name": "Public Tree",
|
|
"tree_structure": {"id": "root", "type": "solution", "title": "T", "description": "T"},
|
|
"is_public": True
|
|
}, headers=auth_headers)
|
|
|
|
# Register viewer
|
|
reg_resp = await client.post("/api/v1/auth/register", json={
|
|
"email": "viewer2@example.com",
|
|
"password": "ViewerPass123!",
|
|
"name": "Viewer 2"
|
|
})
|
|
user_id = PyUUID(reg_resp.json()["id"])
|
|
|
|
result = await test_db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one()
|
|
user.account_role = "viewer"
|
|
await test_db.commit()
|
|
|
|
login_resp = await client.post("/api/v1/auth/login/json", json={
|
|
"email": "viewer2@example.com",
|
|
"password": "ViewerPass123!"
|
|
})
|
|
viewer_headers = {"Authorization": f"Bearer {login_resp.json()['access_token']}"}
|
|
|
|
# Viewer can list trees
|
|
response = await client.get("/api/v1/trees", headers=viewer_headers)
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_owner_can_edit_account_members_tree(self, client: AsyncClient, auth_headers: dict, test_db):
|
|
"""Test that account owner can edit trees created by account members."""
|
|
from sqlalchemy import select
|
|
from app.models.user import User
|
|
from uuid import UUID as PyUUID
|
|
|
|
# Get owner's account
|
|
me_resp = await client.get("/api/v1/auth/me", headers=auth_headers)
|
|
account_id = me_resp.json()["account_id"]
|
|
|
|
# Create invite
|
|
invite_resp = await client.post(
|
|
"/api/v1/accounts/me/invites",
|
|
json={"email": "engineer@example.com", "role": "engineer"},
|
|
headers=auth_headers
|
|
)
|
|
invite_code = invite_resp.json()["code"]
|
|
|
|
# Register engineer in same account
|
|
reg_resp = await client.post("/api/v1/auth/register", json={
|
|
"email": "engineer@example.com",
|
|
"password": "EngineerPass123!",
|
|
"name": "Engineer",
|
|
"account_invite_code": invite_code
|
|
})
|
|
assert reg_resp.status_code == 201
|
|
assert reg_resp.json()["account_id"] == account_id
|
|
|
|
# Login as engineer
|
|
eng_login = await client.post("/api/v1/auth/login/json", json={
|
|
"email": "engineer@example.com",
|
|
"password": "EngineerPass123!"
|
|
})
|
|
eng_headers = {"Authorization": f"Bearer {eng_login.json()['access_token']}"}
|
|
|
|
# Engineer creates a tree
|
|
tree_resp = await client.post("/api/v1/trees", json={
|
|
"name": "Engineer's Tree",
|
|
"tree_structure": {"id": "root", "type": "solution", "title": "T", "description": "T"}
|
|
}, headers=eng_headers)
|
|
assert tree_resp.status_code == 201
|
|
tree_id = tree_resp.json()["id"]
|
|
|
|
# Owner can edit engineer's tree
|
|
update_resp = await client.put(
|
|
f"/api/v1/trees/{tree_id}",
|
|
json={"name": "Owner Updated Name"},
|
|
headers=auth_headers
|
|
)
|
|
assert update_resp.status_code == 200
|
|
assert update_resp.json()["name"] == "Owner Updated Name"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_account_scoped_visibility(self, client: AsyncClient, auth_headers: dict):
|
|
"""Test that account members can see each other's non-public trees."""
|
|
# Get owner's account
|
|
me_resp = await client.get("/api/v1/auth/me", headers=auth_headers)
|
|
account_id = me_resp.json()["account_id"]
|
|
|
|
# Owner creates a private tree
|
|
tree_resp = await client.post("/api/v1/trees", json={
|
|
"name": "Private Account Tree",
|
|
"tree_structure": {"id": "root", "type": "solution", "title": "T", "description": "T"},
|
|
"is_public": False
|
|
}, headers=auth_headers)
|
|
assert tree_resp.status_code == 201
|
|
tree_id = tree_resp.json()["id"]
|
|
|
|
# Create invite and add member
|
|
invite_resp = await client.post(
|
|
"/api/v1/accounts/me/invites",
|
|
json={"email": "teammate@example.com", "role": "engineer"},
|
|
headers=auth_headers
|
|
)
|
|
invite_code = invite_resp.json()["code"]
|
|
|
|
await client.post("/api/v1/auth/register", json={
|
|
"email": "teammate@example.com",
|
|
"password": "TeammatePass123!",
|
|
"name": "Teammate",
|
|
"account_invite_code": invite_code
|
|
})
|
|
|
|
mate_login = await client.post("/api/v1/auth/login/json", json={
|
|
"email": "teammate@example.com",
|
|
"password": "TeammatePass123!"
|
|
})
|
|
mate_headers = {"Authorization": f"Bearer {mate_login.json()['access_token']}"}
|
|
|
|
# Teammate should see the private tree (same account)
|
|
response = await client.get(f"/api/v1/trees/{tree_id}", headers=mate_headers)
|
|
assert response.status_code == 200
|
|
assert response.json()["name"] == "Private Account Tree"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_different_account_cannot_see_private_tree(self, client: AsyncClient, auth_headers: dict):
|
|
"""Test that users from different accounts cannot see private trees."""
|
|
# Owner creates a private tree
|
|
tree_resp = await client.post("/api/v1/trees", json={
|
|
"name": "Secret Tree",
|
|
"tree_structure": {"id": "root", "type": "solution", "title": "T", "description": "T"},
|
|
"is_public": False
|
|
}, headers=auth_headers)
|
|
assert tree_resp.status_code == 201
|
|
tree_id = tree_resp.json()["id"]
|
|
|
|
# Register a completely separate user (different account)
|
|
await client.post("/api/v1/auth/register", json={
|
|
"email": "outsider@example.com",
|
|
"password": "OutsiderPass123!",
|
|
"name": "Outsider"
|
|
})
|
|
|
|
outsider_login = await client.post("/api/v1/auth/login/json", json={
|
|
"email": "outsider@example.com",
|
|
"password": "OutsiderPass123!"
|
|
})
|
|
outsider_headers = {"Authorization": f"Bearer {outsider_login.json()['access_token']}"}
|
|
|
|
# Outsider should NOT see the private tree
|
|
response = await client.get(f"/api/v1/trees/{tree_id}", headers=outsider_headers)
|
|
assert response.status_code == 403
|