Spaces:
Running
Running
File size: 4,993 Bytes
0f8d56c | 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 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | 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()
|