251 lines
10 KiB
Python
251 lines
10 KiB
Python
|
"""Tests for OrgRepository."""
|
||
|
|
|
||
|
|
from uuid import uuid4
|
||
|
|
|
||
|
|
import asyncpg
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from app.repositories.org import OrgRepository
|
||
|
|
from app.repositories.user import UserRepository
|
||
|
|
|
||
|
|
|
||
|
|
class TestOrgRepository:
|
||
|
|
"""Tests for OrgRepository conforming to SPECS.md."""
|
||
|
|
|
||
|
|
async def test_create_org_returns_org_data(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""Creating an org returns the org data."""
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
org_id = uuid4()
|
||
|
|
name = "Test Organization"
|
||
|
|
slug = "test-org"
|
||
|
|
|
||
|
|
result = await repo.create(org_id, name, slug)
|
||
|
|
|
||
|
|
assert result["id"] == org_id
|
||
|
|
assert result["name"] == name
|
||
|
|
assert result["slug"] == slug
|
||
|
|
assert result["created_at"] is not None
|
||
|
|
|
||
|
|
async def test_create_org_slug_must_be_unique(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""Org slug uniqueness constraint per SPECS.md orgs table."""
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
slug = "unique-slug"
|
||
|
|
|
||
|
|
await repo.create(uuid4(), "Org One", slug)
|
||
|
|
|
||
|
|
with pytest.raises(asyncpg.UniqueViolationError):
|
||
|
|
await repo.create(uuid4(), "Org Two", slug)
|
||
|
|
|
||
|
|
async def test_get_by_id_returns_org(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""get_by_id returns the correct organization."""
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
org_id = uuid4()
|
||
|
|
|
||
|
|
await repo.create(org_id, "My Org", "my-org")
|
||
|
|
result = await repo.get_by_id(org_id)
|
||
|
|
|
||
|
|
assert result is not None
|
||
|
|
assert result["id"] == org_id
|
||
|
|
assert result["name"] == "My Org"
|
||
|
|
|
||
|
|
async def test_get_by_id_returns_none_for_nonexistent(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""get_by_id returns None for non-existent org."""
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
result = await repo.get_by_id(uuid4())
|
||
|
|
|
||
|
|
assert result is None
|
||
|
|
|
||
|
|
async def test_get_by_slug_returns_org(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""get_by_slug returns the correct organization."""
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
org_id = uuid4()
|
||
|
|
slug = "slug-lookup"
|
||
|
|
|
||
|
|
await repo.create(org_id, "Slug Test", slug)
|
||
|
|
result = await repo.get_by_slug(slug)
|
||
|
|
|
||
|
|
assert result is not None
|
||
|
|
assert result["id"] == org_id
|
||
|
|
|
||
|
|
async def test_get_by_slug_returns_none_for_nonexistent(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""get_by_slug returns None for non-existent slug."""
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
result = await repo.get_by_slug("nonexistent-slug")
|
||
|
|
|
||
|
|
assert result is None
|
||
|
|
|
||
|
|
async def test_slug_exists_returns_true_when_exists(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""slug_exists returns True when slug exists."""
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
slug = "existing-slug"
|
||
|
|
|
||
|
|
await repo.create(uuid4(), "Existing Org", slug)
|
||
|
|
result = await repo.slug_exists(slug)
|
||
|
|
|
||
|
|
assert result is True
|
||
|
|
|
||
|
|
async def test_slug_exists_returns_false_when_not_exists(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""slug_exists returns False when slug doesn't exist."""
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
result = await repo.slug_exists("no-such-slug")
|
||
|
|
|
||
|
|
assert result is False
|
||
|
|
|
||
|
|
|
||
|
|
class TestOrgMembership:
|
||
|
|
"""Tests for org membership operations per SPECS.md org_members table."""
|
||
|
|
|
||
|
|
async def _create_user(self, conn: asyncpg.Connection, email: str) -> uuid4:
|
||
|
|
"""Helper to create a user."""
|
||
|
|
user_repo = UserRepository(conn)
|
||
|
|
user_id = uuid4()
|
||
|
|
await user_repo.create(user_id, email, "hash")
|
||
|
|
return user_id
|
||
|
|
|
||
|
|
async def _create_org(self, conn: asyncpg.Connection, slug: str) -> uuid4:
|
||
|
|
"""Helper to create an org."""
|
||
|
|
org_repo = OrgRepository(conn)
|
||
|
|
org_id = uuid4()
|
||
|
|
await org_repo.create(org_id, f"Org {slug}", slug)
|
||
|
|
return org_id
|
||
|
|
|
||
|
|
async def test_add_member_creates_membership(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""add_member creates a membership record."""
|
||
|
|
user_id = await self._create_user(db_conn, "member@example.com")
|
||
|
|
org_id = await self._create_org(db_conn, "member-org")
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
result = await repo.add_member(uuid4(), user_id, org_id, "member")
|
||
|
|
|
||
|
|
assert result["user_id"] == user_id
|
||
|
|
assert result["org_id"] == org_id
|
||
|
|
assert result["role"] == "member"
|
||
|
|
assert result["created_at"] is not None
|
||
|
|
|
||
|
|
async def test_add_member_role_must_be_valid(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""Role must be admin, member, or viewer per SPECS.md."""
|
||
|
|
org_id = await self._create_org(db_conn, "role-test-org")
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
# Valid roles should work
|
||
|
|
for role in ["admin", "member", "viewer"]:
|
||
|
|
member_id = uuid4()
|
||
|
|
# Need a new user for each since user+org must be unique
|
||
|
|
new_user_id = await self._create_user(db_conn, f"{role}@example.com")
|
||
|
|
result = await repo.add_member(member_id, new_user_id, org_id, role)
|
||
|
|
assert result["role"] == role
|
||
|
|
|
||
|
|
# Invalid role should fail
|
||
|
|
another_user = await self._create_user(db_conn, "invalid_role@example.com")
|
||
|
|
with pytest.raises(asyncpg.CheckViolationError):
|
||
|
|
await repo.add_member(uuid4(), another_user, org_id, "superuser")
|
||
|
|
|
||
|
|
async def test_add_member_user_org_must_be_unique(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""User can only be member of an org once (unique constraint)."""
|
||
|
|
user_id = await self._create_user(db_conn, "unique_member@example.com")
|
||
|
|
org_id = await self._create_org(db_conn, "unique-member-org")
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
await repo.add_member(uuid4(), user_id, org_id, "member")
|
||
|
|
|
||
|
|
with pytest.raises(asyncpg.UniqueViolationError):
|
||
|
|
await repo.add_member(uuid4(), user_id, org_id, "admin")
|
||
|
|
|
||
|
|
async def test_get_member_returns_membership(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""get_member returns the membership for user and org."""
|
||
|
|
user_id = await self._create_user(db_conn, "get_member@example.com")
|
||
|
|
org_id = await self._create_org(db_conn, "get-member-org")
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
await repo.add_member(uuid4(), user_id, org_id, "admin")
|
||
|
|
result = await repo.get_member(user_id, org_id)
|
||
|
|
|
||
|
|
assert result is not None
|
||
|
|
assert result["user_id"] == user_id
|
||
|
|
assert result["org_id"] == org_id
|
||
|
|
assert result["role"] == "admin"
|
||
|
|
|
||
|
|
async def test_get_member_returns_none_for_nonmember(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""get_member returns None if user is not a member."""
|
||
|
|
user_id = await self._create_user(db_conn, "nonmember@example.com")
|
||
|
|
org_id = await self._create_org(db_conn, "nonmember-org")
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
result = await repo.get_member(user_id, org_id)
|
||
|
|
|
||
|
|
assert result is None
|
||
|
|
|
||
|
|
async def test_get_members_returns_all_org_members(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""get_members returns all members with their emails."""
|
||
|
|
org_id = await self._create_org(db_conn, "all-members-org")
|
||
|
|
user1 = await self._create_user(db_conn, "user1@example.com")
|
||
|
|
user2 = await self._create_user(db_conn, "user2@example.com")
|
||
|
|
user3 = await self._create_user(db_conn, "user3@example.com")
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
await repo.add_member(uuid4(), user1, org_id, "admin")
|
||
|
|
await repo.add_member(uuid4(), user2, org_id, "member")
|
||
|
|
await repo.add_member(uuid4(), user3, org_id, "viewer")
|
||
|
|
|
||
|
|
result = await repo.get_members(org_id)
|
||
|
|
|
||
|
|
assert len(result) == 3
|
||
|
|
emails = {m["email"] for m in result}
|
||
|
|
assert emails == {"user1@example.com", "user2@example.com", "user3@example.com"}
|
||
|
|
|
||
|
|
async def test_get_members_returns_empty_list_for_no_members(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""get_members returns empty list for org with no members."""
|
||
|
|
org_id = await self._create_org(db_conn, "empty-org")
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
result = await repo.get_members(org_id)
|
||
|
|
|
||
|
|
assert result == []
|
||
|
|
|
||
|
|
async def test_get_user_orgs_returns_all_user_memberships(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""get_user_orgs returns all orgs a user belongs to with their role."""
|
||
|
|
user_id = await self._create_user(db_conn, "multi_org@example.com")
|
||
|
|
org1 = await self._create_org(db_conn, "user-org-1")
|
||
|
|
org2 = await self._create_org(db_conn, "user-org-2")
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
await repo.add_member(uuid4(), user_id, org1, "admin")
|
||
|
|
await repo.add_member(uuid4(), user_id, org2, "member")
|
||
|
|
|
||
|
|
result = await repo.get_user_orgs(user_id)
|
||
|
|
|
||
|
|
assert len(result) == 2
|
||
|
|
slugs = {o["slug"] for o in result}
|
||
|
|
assert slugs == {"user-org-1", "user-org-2"}
|
||
|
|
# Check role is included
|
||
|
|
roles = {o["role"] for o in result}
|
||
|
|
assert roles == {"admin", "member"}
|
||
|
|
|
||
|
|
async def test_get_user_orgs_returns_empty_for_no_memberships(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""get_user_orgs returns empty list for user with no memberships."""
|
||
|
|
user_id = await self._create_user(db_conn, "no_orgs@example.com")
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
result = await repo.get_user_orgs(user_id)
|
||
|
|
|
||
|
|
assert result == []
|
||
|
|
|
||
|
|
async def test_member_requires_valid_user_foreign_key(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""org_members.user_id must reference existing user."""
|
||
|
|
org_id = await self._create_org(db_conn, "fk-test-org")
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
with pytest.raises(asyncpg.ForeignKeyViolationError):
|
||
|
|
await repo.add_member(uuid4(), uuid4(), org_id, "member")
|
||
|
|
|
||
|
|
async def test_member_requires_valid_org_foreign_key(self, db_conn: asyncpg.Connection) -> None:
|
||
|
|
"""org_members.org_id must reference existing org."""
|
||
|
|
user_id = await self._create_user(db_conn, "fk_user@example.com")
|
||
|
|
repo = OrgRepository(db_conn)
|
||
|
|
|
||
|
|
with pytest.raises(asyncpg.ForeignKeyViolationError):
|
||
|
|
await repo.add_member(uuid4(), user_id, uuid4(), "member")
|