daddy
712d21e746
docs/planning/social.md US-5 (task 13). Models + Alembic migration only, per the sibling-dependency boundary with the Groups API task. - groups: id, name, description, owner_id (FK users, ON DELETE SET NULL -- a group is a durable multi-member community, so it and its full roster survive the owner deleting their account; the API task handles surfacing an ownerless group for reassignment). - group_members: group_id/user_id (both FK, ON DELETE CASCADE -- pure join-table rows with no meaning without either side), role (owner/admin/member native enum, default member), joined_at, unique constraint on (group_id, user_id). - User gets owned_groups / group_memberships relationships. - tests/conftest.py TEST_TABLES updated for truncation between tests. - New tests/test_group_models.py: role assignment, unique-membership rejection, group-delete cascade, member-delete cascade, and owner-delete SET NULL survival -- all verified against real Postgres. Verified: upgrade/downgrade/upgrade round-trip, \d-inspected schema matches spec (FKs, cascade behavior, unique constraint, indexes), full backend suite 166 passed / 0 regressions. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y2FESCaMXeJpC9gSctiEeD
259 lines
10 KiB
Python
259 lines
10 KiB
Python
"""Model-level tests for groups: groups and group_members.
|
|
|
|
app/models/groups.py / alembic/versions/0009_add_group_tables.py
|
|
(docs/planning/social.md US-5). This is the models+migration task only --
|
|
there are no API endpoints yet (a sibling task, "Groups API", owns
|
|
create/join/leave/role authorization and the group page endpoint), so these
|
|
tests exercise the ORM models and the real Postgres constraints directly
|
|
rather than going through HTTP.
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
from app.config import settings
|
|
from app.main import app
|
|
from app.models.groups import Group, GroupMember, GroupRole
|
|
from app.models.user import User
|
|
|
|
|
|
def _register(client: TestClient, username: str) -> int:
|
|
resp = client.post(
|
|
"/api/auth/register",
|
|
json={
|
|
"username": username,
|
|
"email": f"{username}@example.com",
|
|
"password": "secret123",
|
|
"date_of_birth": "1990-01-01",
|
|
},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
headers = {"Authorization": f"Bearer {resp.json()['access_token']}"}
|
|
me = client.get("/api/users/me", headers=headers)
|
|
assert me.status_code == 200, me.text
|
|
return me.json()["id"]
|
|
|
|
|
|
def _session_factory():
|
|
engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
|
|
return engine, async_sessionmaker(engine, expire_on_commit=False)
|
|
|
|
|
|
class TestGroupCreationAndRoles:
|
|
async def test_create_group_with_members_of_different_roles(self):
|
|
client = TestClient(app)
|
|
owner = _register(client, "grp_role_owner")
|
|
admin = _register(client, "grp_role_admin")
|
|
member = _register(client, "grp_role_member")
|
|
|
|
engine, factory = _session_factory()
|
|
try:
|
|
async with factory() as session:
|
|
group = Group(name="Explorers", description="We explore worlds", owner_id=owner)
|
|
session.add(group)
|
|
await session.flush()
|
|
session.add_all(
|
|
[
|
|
GroupMember(group_id=group.id, user_id=owner, role=GroupRole.OWNER),
|
|
GroupMember(group_id=group.id, user_id=admin, role=GroupRole.ADMIN),
|
|
GroupMember(group_id=group.id, user_id=member, role=GroupRole.MEMBER),
|
|
]
|
|
)
|
|
await session.commit()
|
|
group_id = group.id
|
|
|
|
async with factory() as session:
|
|
fetched = await session.get(Group, group_id)
|
|
members = (
|
|
await session.execute(
|
|
select(GroupMember).where(GroupMember.group_id == group_id)
|
|
)
|
|
).scalars().all()
|
|
by_user = {m.user_id: m.role for m in members}
|
|
assert fetched.owner_id == owner
|
|
assert len(members) == 3
|
|
assert by_user[owner] == GroupRole.OWNER
|
|
assert by_user[admin] == GroupRole.ADMIN
|
|
assert by_user[member] == GroupRole.MEMBER
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
async def test_membership_role_defaults_to_member(self):
|
|
client = TestClient(app)
|
|
owner = _register(client, "grp_default_owner")
|
|
joiner = _register(client, "grp_default_joiner")
|
|
|
|
engine, factory = _session_factory()
|
|
try:
|
|
async with factory() as session:
|
|
group = Group(name="Defaults", owner_id=owner)
|
|
session.add(group)
|
|
await session.flush()
|
|
session.add(GroupMember(group_id=group.id, user_id=joiner))
|
|
await session.commit()
|
|
group_id = group.id
|
|
|
|
async with factory() as session:
|
|
gm = (
|
|
await session.execute(
|
|
select(GroupMember).where(
|
|
GroupMember.group_id == group_id, GroupMember.user_id == joiner
|
|
)
|
|
)
|
|
).scalar_one()
|
|
assert gm.role == GroupRole.MEMBER
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
class TestUniqueMembershipConstraint:
|
|
async def test_duplicate_membership_rejected(self):
|
|
client = TestClient(app)
|
|
owner = _register(client, "grp_uq_owner")
|
|
member = _register(client, "grp_uq_member")
|
|
|
|
engine, factory = _session_factory()
|
|
try:
|
|
async with factory() as session:
|
|
group = Group(name="Uniqueness", owner_id=owner)
|
|
session.add(group)
|
|
await session.flush()
|
|
session.add(GroupMember(group_id=group.id, user_id=member))
|
|
await session.commit()
|
|
group_id = group.id
|
|
|
|
async with factory() as session:
|
|
session.add(GroupMember(group_id=group_id, user_id=member))
|
|
with pytest.raises(IntegrityError):
|
|
await session.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
class TestCascadeDelete:
|
|
async def test_deleting_group_deletes_members(self):
|
|
client = TestClient(app)
|
|
owner = _register(client, "grp_cd_group_owner")
|
|
member = _register(client, "grp_cd_group_member")
|
|
|
|
engine, factory = _session_factory()
|
|
try:
|
|
async with factory() as session:
|
|
group = Group(name="To Delete", owner_id=owner)
|
|
session.add(group)
|
|
await session.flush()
|
|
session.add_all(
|
|
[
|
|
GroupMember(group_id=group.id, user_id=owner, role=GroupRole.OWNER),
|
|
GroupMember(group_id=group.id, user_id=member),
|
|
]
|
|
)
|
|
await session.commit()
|
|
group_id = group.id
|
|
|
|
async with factory() as session:
|
|
# Core delete against groups directly, so the real ON DELETE
|
|
# CASCADE does the work (not the ORM).
|
|
await session.execute(delete(Group).where(Group.id == group_id))
|
|
await session.commit()
|
|
|
|
async with factory() as session:
|
|
members = (
|
|
await session.execute(
|
|
select(GroupMember).where(GroupMember.group_id == group_id)
|
|
)
|
|
).scalars().all()
|
|
assert members == []
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
async def test_deleting_member_user_removes_only_their_membership(self):
|
|
client = TestClient(app)
|
|
owner = _register(client, "grp_cd_user_owner")
|
|
member = _register(client, "grp_cd_user_member")
|
|
|
|
engine, factory = _session_factory()
|
|
try:
|
|
async with factory() as session:
|
|
group = Group(name="Membership Cascade", owner_id=owner)
|
|
session.add(group)
|
|
await session.flush()
|
|
session.add_all(
|
|
[
|
|
GroupMember(group_id=group.id, user_id=owner, role=GroupRole.OWNER),
|
|
GroupMember(group_id=group.id, user_id=member),
|
|
]
|
|
)
|
|
await session.commit()
|
|
group_id = group.id
|
|
|
|
async with factory() as session:
|
|
await session.execute(delete(User).where(User.id == member))
|
|
await session.commit()
|
|
|
|
async with factory() as session:
|
|
remaining = (
|
|
await session.execute(
|
|
select(GroupMember).where(GroupMember.group_id == group_id)
|
|
)
|
|
).scalars().all()
|
|
# Only the deleted member's row is gone; the group and the
|
|
# owner's membership survive.
|
|
assert len(remaining) == 1
|
|
assert remaining[0].user_id == owner
|
|
|
|
group = await session.get(Group, group_id)
|
|
assert group is not None
|
|
assert group.owner_id == owner
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
async def test_deleting_owner_survives_with_owner_id_null(self):
|
|
"""Deleting the owner's account must not delete the group or its
|
|
other members -- only ``owner_id`` is nulled out (SET NULL), per the
|
|
module docstring in app/models/groups.py."""
|
|
client = TestClient(app)
|
|
owner = _register(client, "grp_cd_owner_owner")
|
|
member = _register(client, "grp_cd_owner_member")
|
|
|
|
engine, factory = _session_factory()
|
|
try:
|
|
async with factory() as session:
|
|
group = Group(name="Survives Owner Deletion", owner_id=owner)
|
|
session.add(group)
|
|
await session.flush()
|
|
session.add_all(
|
|
[
|
|
GroupMember(group_id=group.id, user_id=owner, role=GroupRole.OWNER),
|
|
GroupMember(group_id=group.id, user_id=member),
|
|
]
|
|
)
|
|
await session.commit()
|
|
group_id = group.id
|
|
|
|
async with factory() as session:
|
|
await session.execute(delete(User).where(User.id == owner))
|
|
await session.commit()
|
|
|
|
async with factory() as session:
|
|
group = await session.get(Group, group_id)
|
|
assert group is not None
|
|
assert group.owner_id is None
|
|
assert group.name == "Survives Owner Deletion"
|
|
|
|
# The owner's own membership row is gone (user_id CASCADE),
|
|
# but the other member's row survives untouched.
|
|
members = (
|
|
await session.execute(
|
|
select(GroupMember).where(GroupMember.group_id == group_id)
|
|
)
|
|
).scalars().all()
|
|
assert len(members) == 1
|
|
assert members[0].user_id == member
|
|
finally:
|
|
await engine.dispose()
|