tests/integration/test_reference_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_reference_trigger_normalizes_on_insert(prisma: Prisma) -> None:
    ref_stored = f"pytest-trig-{uuid.uuid4().hex[:12]}"
    ref_input = f"a{ref_stored}"
    rows = await prisma.query_raw(
        "SELECT public.upsert_article_by_main_ref($1::text, $2::text, $3::text, $4::uuid) AS id",
        ref_input,
        "L",
        None,
        None,
    )
    article_id = rows[0]["id"]
    ar = await prisma.article_references.find_first(
        where={"reference_number": ref_stored, "main": True}
    )
    assert ar is not None
    assert ar.article_id == article_id

    await prisma.article_references.delete_many(where={"reference_number": ref_stored})
    await prisma.articles.delete(where={"id": article_id})


@pytest.mark.integration
@pytest.mark.asyncio
async def test_reference_trigger_normalizes_on_update(prisma: Prisma) -> None:
    ref = f"pytest-upd-{uuid.uuid4().hex[:12]}"
    rows = await prisma.query_raw(
        "SELECT public.upsert_article_by_main_ref($1::text, $2::text, $3::text, $4::uuid) AS id",
        ref,
        "L",
        None,
        None,
    )
    article_id = rows[0]["id"]
    ar = await prisma.article_references.find_first(
        where={"article_id": article_id, "main": True}
    )
    assert ar is not None

    new_ref = f"new-{uuid.uuid4().hex[:10]}"
    await prisma.article_references.update(
        where={"id": ar.id},
        data={"reference_number": f"a{new_ref}"},
    )
    updated = await prisma.article_references.find_unique(where={"id": ar.id})
    assert updated is not None
    assert updated.reference_number == new_ref

    await prisma.article_references.delete_many(
        where={"reference_number": {"in": [ref, new_ref]}}
    )
    await prisma.articles.delete(where={"id": article_id})


@pytest.mark.integration
@pytest.mark.asyncio
async def test_upsert_article_by_main_ref_a_prefix_matches_plain(prisma: Prisma) -> None:
    ref = f"pytest-norm-{uuid.uuid4().hex[:12]}"
    rows_a = await prisma.query_raw(
        "SELECT public.upsert_article_by_main_ref($1::text, $2::text, $3::text, $4::uuid) AS id",
        f"a{ref}",
        "L1",
        None,
        None,
    )
    id_a = rows_a[0]["id"]
    rows_plain = await prisma.query_raw(
        "SELECT public.upsert_article_by_main_ref($1::text, $2::text, $3::text, $4::uuid) AS id",
        ref,
        "L2",
        None,
        None,
    )
    assert rows_plain[0]["id"] == id_a

    await prisma.article_references.delete_many(where={"reference_number": ref})
    await prisma.articles.delete(where={"id": id_a})


@pytest.mark.integration
@pytest.mark.asyncio
async def test_find_article_references_by_ref_rpc(prisma: Prisma) -> None:
    ref = f"pytest-rpc-{uuid.uuid4().hex[:12]}"
    await prisma.query_raw(
        "SELECT public.upsert_article_by_main_ref($1::text, $2::text, $3::text, $4::uuid) AS id",
        ref,
        "L",
        None,
        None,
    )

    rows_a = await prisma.query_raw(
        "SELECT * FROM public.find_article_references_by_ref($1::text)",
        f"a{ref}",
    )
    rows_plain = await prisma.query_raw(
        "SELECT * FROM public.find_article_references_by_ref($1::text)",
        ref,
    )
    assert len(rows_a) == 1
    assert len(rows_plain) == 1
    assert rows_a[0]["reference_number"] == ref
    assert rows_plain[0]["reference_number"] == ref

    article_id = rows_a[0]["article_id"]
    await prisma.article_references.delete_many(where={"reference_number": ref})
    await prisma.articles.delete(where={"id": article_id})


@pytest.mark.integration
@pytest.mark.asyncio
async def test_direct_insert_normalizes_reference() -> None:
    conn = await asyncpg.connect(get_settings().database_url)
    ref = f"pytest-dir-{uuid.uuid4().hex[:12]}"
    ref_input = f"a{ref}"
    article_id = None
    try:
        async with conn.transaction():
            article_id = await conn.fetchval(
                "INSERT INTO public.articles (label_fr) VALUES ('direct') RETURNING id"
            )
            await conn.execute(
                """
                INSERT INTO public.article_references (article_id, reference_number, main)
                VALUES ($1, $2, true)
                """,
                article_id,
                ref_input,
            )
        stored = await conn.fetchval(
            """
            SELECT reference_number FROM public.article_references
            WHERE article_id = $1 AND main = true
            """,
            article_id,
        )
        assert stored == ref
    finally:
        if article_id is not None:
            await conn.execute(
                "DELETE FROM public.article_references WHERE article_id = $1",
                article_id,
            )
            await conn.execute("DELETE FROM public.articles WHERE id = $1", article_id)
        await conn.close()