tests/integration/test_catalog_create_admin_rpc.py

import uuid

import asyncpg
import pytest

from app.core.config import get_settings
from prisma import Prisma


@pytest.mark.integration
@pytest.mark.asyncio
async def test_upsert_article_by_main_ref_forbidden_for_non_admin() -> None:
    conn = await asyncpg.connect(get_settings().database_url)
    test_user_id = str(uuid.uuid4())
    try:
        with pytest.raises(asyncpg.PostgresError) as exc_info:
            async with conn.transaction():
                await conn.execute("set local role authenticated")
                await conn.execute("select set_config('request.jwt.claim.sub', $1, true)", test_user_id)
                await conn.fetchval(
                    """
                    select public.upsert_article_by_main_ref($1::text, $2::text, $3::text, $4::uuid)
                    """,
                    f"forbidden-{uuid.uuid4().hex[:8]}",
                    "Label",
                    None,
                    None,
                )
        assert exc_info.value.sqlstate == "42501"
    finally:
        await conn.close()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_upsert_article_by_main_ref_allowed_for_admin(prisma: Prisma) -> None:
    conn = await asyncpg.connect(get_settings().database_url)
    admin_user_id = str(uuid.uuid4())
    ref = f"zadmin-rpc-{uuid.uuid4().hex[:10]}"
    article_id: str | None = None
    try:
        await conn.execute(
            """
            insert into public.user_roles (user_id, role)
            values ($1::uuid, 'admin'::public.app_role)
            """,
            admin_user_id,
        )
        async with conn.transaction():
            await conn.execute("set local role authenticated")
            await conn.execute("select set_config('request.jwt.claim.sub', $1, true)", admin_user_id)
            article_id = str(
                await conn.fetchval(
                    """
                    select public.upsert_article_by_main_ref($1::text, $2::text, $3::text, $4::uuid)
                    """,
                    ref,
                    "Admin Label",
                    None,
                    None,
                )
            )
        assert article_id is not None
        ar = await prisma.article_references.find_first(
            where={"reference_number": ref, "main": True}
        )
        assert ar is not None
        assert ar.article_id == article_id
    finally:
        await conn.execute("delete from public.user_roles where user_id = $1::uuid", admin_user_id)
        if article_id is not None:
            await prisma.article_references.delete_many(where={"reference_number": ref})
            await prisma.articles.delete(where={"id": article_id})
        await conn.close()