modbison/backend/tests/test_social_models.py

"""Model-level tests for the social graph: friendships, follows, blocks.

app/models/social.py / alembic/versions/0007_add_social_graph_tables.py
(docs/planning/social.md US-1). This is the models+migration task only --
there are no API endpoints yet (a sibling task owns the router), 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.social import Friendship, Follow, Block, FriendshipStatus
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 TestFriendshipRequestAccept:
    async def test_request_defaults_to_pending(self):
        client = TestClient(app)
        a = _register(client, "sg_fr_a1")
        b = _register(client, "sg_fr_b1")

        engine, factory = _session_factory()
        try:
            async with factory() as session:
                fr = Friendship(requester_id=a, addressee_id=b)
                session.add(fr)
                await session.commit()
                await session.refresh(fr)
                assert fr.status == FriendshipStatus.PENDING
                assert fr.requester_id == a
                assert fr.addressee_id == b
        finally:
            await engine.dispose()

    async def test_accept_updates_status(self):
        client = TestClient(app)
        a = _register(client, "sg_fr_a2")
        b = _register(client, "sg_fr_b2")

        engine, factory = _session_factory()
        try:
            async with factory() as session:
                fr = Friendship(requester_id=a, addressee_id=b)
                session.add(fr)
                await session.commit()
                fr_id = fr.id

            async with factory() as session:
                fr = await session.get(Friendship, fr_id)
                fr.status = FriendshipStatus.ACCEPTED
                await session.commit()

            async with factory() as session:
                fr = await session.get(Friendship, fr_id)
                assert fr.status == FriendshipStatus.ACCEPTED
        finally:
            await engine.dispose()


class TestUniquePairConstraints:
    async def test_duplicate_friendship_pair_rejected(self):
        client = TestClient(app)
        a = _register(client, "sg_uq_fr_a")
        b = _register(client, "sg_uq_fr_b")

        engine, factory = _session_factory()
        try:
            async with factory() as session:
                session.add(Friendship(requester_id=a, addressee_id=b))
                await session.commit()

            async with factory() as session:
                session.add(Friendship(requester_id=a, addressee_id=b))
                with pytest.raises(IntegrityError):
                    await session.commit()
        finally:
            await engine.dispose()

    async def test_duplicate_follow_pair_rejected(self):
        client = TestClient(app)
        a = _register(client, "sg_uq_fw_a")
        b = _register(client, "sg_uq_fw_b")

        engine, factory = _session_factory()
        try:
            async with factory() as session:
                session.add(Follow(follower_id=a, followed_id=b))
                await session.commit()

            async with factory() as session:
                session.add(Follow(follower_id=a, followed_id=b))
                with pytest.raises(IntegrityError):
                    await session.commit()
        finally:
            await engine.dispose()

    async def test_duplicate_block_pair_rejected(self):
        client = TestClient(app)
        a = _register(client, "sg_uq_bl_a")
        b = _register(client, "sg_uq_bl_b")

        engine, factory = _session_factory()
        try:
            async with factory() as session:
                session.add(Block(blocker_id=a, blocked_id=b))
                await session.commit()

            async with factory() as session:
                session.add(Block(blocker_id=a, blocked_id=b))
                with pytest.raises(IntegrityError):
                    await session.commit()
        finally:
            await engine.dispose()


class TestFollow:
    async def test_create_follow(self):
        client = TestClient(app)
        a = _register(client, "sg_follow_a")
        b = _register(client, "sg_follow_b")

        engine, factory = _session_factory()
        try:
            async with factory() as session:
                session.add(Follow(follower_id=a, followed_id=b))
                await session.commit()

            async with factory() as session:
                rows = (await session.execute(
                    select(Follow).where(Follow.follower_id == a, Follow.followed_id == b)
                )).scalars().all()
                assert len(rows) == 1
        finally:
            await engine.dispose()


class TestBlock:
    async def test_create_block(self):
        client = TestClient(app)
        a = _register(client, "sg_block_a")
        b = _register(client, "sg_block_b")

        engine, factory = _session_factory()
        try:
            async with factory() as session:
                session.add(Block(blocker_id=a, blocked_id=b))
                await session.commit()

            async with factory() as session:
                rows = (await session.execute(
                    select(Block).where(Block.blocker_id == a, Block.blocked_id == b)
                )).scalars().all()
                assert len(rows) == 1
        finally:
            await engine.dispose()


class TestCascadeDelete:
    async def test_deleting_user_removes_rows_in_all_three_tables(self):
        """Deleting a user removes friendships/follows/blocks referencing them
        in *either* FK column, via the real DB ON DELETE CASCADE (deleted with
        a Core statement, bypassing any ORM-level cascade, to prove the
        database constraint itself does the job)."""
        client = TestClient(app)
        a = _register(client, "sg_cascade_a")
        b = _register(client, "sg_cascade_b")
        c = _register(client, "sg_cascade_c")

        engine, factory = _session_factory()
        try:
            async with factory() as session:
                # a as requester/follower/blocker, and also as addressee (c -> a)
                session.add(Friendship(requester_id=a, addressee_id=b))
                session.add(Friendship(requester_id=c, addressee_id=a))
                session.add(Follow(follower_id=a, followed_id=b))
                session.add(Follow(follower_id=c, followed_id=a))
                session.add(Block(blocker_id=a, blocked_id=b))
                session.add(Block(blocker_id=c, blocked_id=a))
                await session.commit()

            async with factory() as session:
                await session.execute(delete(User).where(User.id == a))
                await session.commit()

            async with factory() as session:
                friendships = (await session.execute(
                    select(Friendship).where(
                        (Friendship.requester_id == a) | (Friendship.addressee_id == a)
                    )
                )).scalars().all()
                follows = (await session.execute(
                    select(Follow).where(
                        (Follow.follower_id == a) | (Follow.followed_id == a)
                    )
                )).scalars().all()
                blocks = (await session.execute(
                    select(Block).where(
                        (Block.blocker_id == a) | (Block.blocked_id == a)
                    )
                )).scalars().all()
                assert friendships == []
                assert follows == []
                assert blocks == []

                # Unrelated rows (b, c only) survive.
                remaining_users = (await session.execute(
                    select(User.id).where(User.id.in_([b, c]))
                )).scalars().all()
                assert set(remaining_users) == {b, c}
        finally:
            await engine.dispose()