66 lines
1.7 KiB
Python
66 lines
1.7 KiB
Python
"""RabbitMQ message schema unit tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from contract_check.core.mq.messages import DocumentExtracted, DocumentUploaded
|
|
|
|
|
|
def _ids() -> tuple[uuid.UUID, uuid.UUID, uuid.UUID]:
|
|
return uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
|
|
|
|
|
|
def test_document_uploaded_roundtrip() -> None:
|
|
cid, did, uid = _ids()
|
|
msg = DocumentUploaded(
|
|
correlation_id=cid,
|
|
document_id=did,
|
|
user_id=uid,
|
|
s3_key=f"users/{uid}/docs/{did}.pdf",
|
|
filename="contract.pdf",
|
|
mime="application/pdf",
|
|
)
|
|
parsed = DocumentUploaded.model_validate_json(msg.model_dump_json())
|
|
assert parsed == msg
|
|
assert parsed.attempt == 0
|
|
|
|
|
|
def test_next_attempt_increments() -> None:
|
|
cid, did, uid = _ids()
|
|
msg = DocumentUploaded(
|
|
correlation_id=cid, document_id=did, user_id=uid, s3_key="k", filename="f", mime="m"
|
|
)
|
|
assert msg.attempt == 0
|
|
assert msg.next_attempt().attempt == 1
|
|
assert msg.next_attempt().next_attempt().attempt == 2
|
|
|
|
|
|
def test_negative_attempt_rejected() -> None:
|
|
cid, did, uid = _ids()
|
|
with pytest.raises(ValueError):
|
|
DocumentUploaded(
|
|
correlation_id=cid,
|
|
document_id=did,
|
|
user_id=uid,
|
|
s3_key="k",
|
|
filename="f",
|
|
mime="m",
|
|
attempt=-1,
|
|
)
|
|
|
|
|
|
def test_document_extracted_fields() -> None:
|
|
cid, did, uid = _ids()
|
|
msg = DocumentExtracted(
|
|
correlation_id=cid,
|
|
document_id=did,
|
|
user_id=uid,
|
|
extracted_s3_key=f"users/{uid}/docs/{did}.txt",
|
|
char_count=4242,
|
|
ocr_used=True,
|
|
)
|
|
assert msg.char_count == 4242
|
|
assert msg.ocr_used is True
|