modbison/backend/tests/test_group_models.py

"""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()