mjalmoudy94
717e53a7ce
مخزن اختصاصی برنامهٔ وب؛ محتوای پوشهٔ web در ریشهٔ مخزن قرار گرفت. تاریخچهٔ پروژهٔ ربات (agy-bot) در این مخزن وجود ندارد و هیچ وابستگی runtime به آن نیست.
137 lines
5.3 KiB
Python
137 lines
5.3 KiB
Python
"""Multi-tenancy: the single most important guarantee in the product.
|
|
|
|
A member of organization A must never be able to read or mutate anything that
|
|
belongs to organization B — not by listing, not by guessing an id.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from httpx import AsyncClient
|
|
|
|
from tests.conftest import auth_header, signup
|
|
|
|
|
|
async def _create_org(client: AsyncClient, token: str, name: str) -> dict:
|
|
response = await client.post(
|
|
"/api/organizations", json={"name": name}, headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 201, response.text
|
|
return response.json()
|
|
|
|
|
|
async def test_create_organization_makes_caller_the_owner(client: AsyncClient):
|
|
token = (await signup(client, "owner-a@example.com"))["access_token"]
|
|
org = await _create_org(client, token, "شرکت آلفا")
|
|
assert org["role"] == "owner"
|
|
assert org["slug"]
|
|
assert org["owner_user_id"]
|
|
|
|
|
|
async def test_organization_slug_is_unique_across_tenants(client: AsyncClient):
|
|
token_a = (await signup(client, "slug-a@example.com"))["access_token"]
|
|
token_b = (await signup(client, "slug-b@example.com"))["access_token"]
|
|
first = await _create_org(client, token_a, "My Company")
|
|
second = await _create_org(client, token_b, "My Company")
|
|
assert first["slug"] != second["slug"]
|
|
assert second["slug"].startswith(first["slug"])
|
|
|
|
|
|
async def test_list_only_returns_own_organizations(client: AsyncClient):
|
|
token_a = (await signup(client, "list-a@example.com"))["access_token"]
|
|
token_b = (await signup(client, "list-b@example.com"))["access_token"]
|
|
org_a = await _create_org(client, token_a, "شرکت الف")
|
|
await _create_org(client, token_b, "شرکت ب")
|
|
|
|
listing = await client.get("/api/organizations", headers=auth_header(token_a))
|
|
assert listing.status_code == 200
|
|
ids = [o["id"] for o in listing.json()]
|
|
assert ids == [org_a["id"]]
|
|
|
|
|
|
async def test_foreign_organization_is_not_readable(client: AsyncClient):
|
|
token_a = (await signup(client, "iso-a@example.com"))["access_token"]
|
|
token_b = (await signup(client, "iso-b@example.com"))["access_token"]
|
|
org_a = await _create_org(client, token_a, "سازمان محرمانه")
|
|
|
|
# B knows A's id but has no membership.
|
|
response = await client.get(f"/api/organizations/{org_a['id']}", headers=auth_header(token_b))
|
|
assert response.status_code == 404
|
|
assert response.json()["error"]["code"] == "not_found"
|
|
|
|
|
|
async def test_foreign_member_list_is_not_readable(client: AsyncClient):
|
|
token_a = (await signup(client, "mem-a@example.com"))["access_token"]
|
|
token_b = (await signup(client, "mem-b@example.com"))["access_token"]
|
|
org_a = await _create_org(client, token_a, "سازمان با اعضا")
|
|
|
|
response = await client.get(
|
|
f"/api/organizations/{org_a['id']}/members", headers=auth_header(token_b)
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
async def test_foreign_invite_is_rejected(client: AsyncClient):
|
|
token_a = (await signup(client, "inv-a@example.com"))["access_token"]
|
|
token_b = (await signup(client, "inv-b@example.com"))["access_token"]
|
|
org_a = await _create_org(client, token_a, "سازمان دعوت")
|
|
|
|
response = await client.post(
|
|
f"/api/organizations/{org_a['id']}/invites",
|
|
json={"email": "attacker@example.com", "role": "admin"},
|
|
headers=auth_header(token_b),
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
async def test_owner_can_list_members_and_see_themselves(client: AsyncClient):
|
|
token = (await signup(client, "solo@example.com", name="تکنفره"))["access_token"]
|
|
org = await _create_org(client, token, "تکنفره")
|
|
|
|
response = await client.get(
|
|
f"/api/organizations/{org['id']}/members", headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 200
|
|
members = response.json()
|
|
assert len(members) == 1
|
|
assert members[0]["role"] == "owner"
|
|
assert members[0]["email"] == "solo@example.com"
|
|
|
|
|
|
async def test_invite_returns_token_only_to_the_inviter(client: AsyncClient):
|
|
token = (await signup(client, "inviter@example.com"))["access_token"]
|
|
org = await _create_org(client, token, "دعوتکننده")
|
|
|
|
response = await client.post(
|
|
f"/api/organizations/{org['id']}/invites",
|
|
json={"email": "newbie@example.com", "role": "member"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert response.status_code == 201
|
|
body = response.json()
|
|
assert body["invite_token"]
|
|
assert body["email"] == "newbie@example.com"
|
|
|
|
|
|
async def test_cannot_invite_as_owner(client: AsyncClient):
|
|
token = (await signup(client, "noowner@example.com"))["access_token"]
|
|
org = await _create_org(client, token, "بدون مالک دوم")
|
|
|
|
response = await client.post(
|
|
f"/api/organizations/{org['id']}/invites",
|
|
json={"email": "wannabe@example.com", "role": "owner"},
|
|
headers=auth_header(token),
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
async def test_audit_log_records_organization_creation(client: AsyncClient):
|
|
token = (await signup(client, "audited@example.com"))["access_token"]
|
|
org = await _create_org(client, token, "سازمان دارای audit")
|
|
|
|
response = await client.get(
|
|
f"/api/organizations/{org['id']}/audit", headers=auth_header(token)
|
|
)
|
|
assert response.status_code == 200
|
|
actions = [event["action"] for event in response.json()]
|
|
assert "organization.create" in actions
|