Spaces:
Sleeping
Sleeping
File size: 2,722 Bytes
96d8d92 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | 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()
|