Sprint 2 : moteur de documents + sanctuaire
Backend: - CRUD complet documents/items/versions (update, delete, accept, reject, reorder) - Service IPFS (upload/retrieve/pin via kubo HTTP API) - Service sanctuaire : pipeline SHA-256 + IPFS + on-chain (system.remark) - Verification integrite des entrees sanctuaire - Recherche par reference (document -> entrees sanctuaire) - Serialisation deterministe des documents pour archivage - 14 tests unitaires supplementaires (document service) Frontend: - 9 composants : StatusBadge, MarkdownRenderer, DiffView, ItemCard, ItemVersionDiff, DocumentList, SanctuaryEntry, IPFSLink, ChainAnchor - Page detail item avec historique des versions et diff - Page detail sanctuaire avec verification integrite - Modal de creation de document + proposition de version - Archivage document vers sanctuaire depuis la page detail Documentation: - API reference mise a jour (9 nouveaux endpoints) - Guides utilisateur documents et sanctuaire enrichis Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Yvv
418
backend/app/tests/test_documents.py
Normal file
418
backend/app/tests/test_documents.py
Normal file
@@ -0,0 +1,418 @@
|
||||
"""Tests for document service: apply_version, reject_version, and serialization.
|
||||
|
||||
These are pure unit tests that mock the database layer to test
|
||||
the service logic in isolation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
sqlalchemy = pytest.importorskip("sqlalchemy", reason="sqlalchemy required for document service tests")
|
||||
|
||||
from app.services.document_service import ( # noqa: E402
|
||||
apply_version,
|
||||
reject_version,
|
||||
serialize_document_to_text,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers: mock objects that behave like SQLAlchemy models
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_item(
|
||||
item_id: uuid.UUID | None = None,
|
||||
document_id: uuid.UUID | None = None,
|
||||
current_text: str = "Texte original",
|
||||
position: str = "1",
|
||||
item_type: str = "clause",
|
||||
title: str | None = None,
|
||||
sort_order: int = 0,
|
||||
) -> MagicMock:
|
||||
"""Create a mock DocumentItem."""
|
||||
item = MagicMock()
|
||||
item.id = item_id or uuid.uuid4()
|
||||
item.document_id = document_id or uuid.uuid4()
|
||||
item.current_text = current_text
|
||||
item.position = position
|
||||
item.item_type = item_type
|
||||
item.title = title
|
||||
item.sort_order = sort_order
|
||||
item.created_at = datetime.now(timezone.utc)
|
||||
item.updated_at = datetime.now(timezone.utc)
|
||||
return item
|
||||
|
||||
|
||||
def _make_version(
|
||||
version_id: uuid.UUID | None = None,
|
||||
item_id: uuid.UUID | None = None,
|
||||
proposed_text: str = "Texte propose",
|
||||
status: str = "proposed",
|
||||
) -> MagicMock:
|
||||
"""Create a mock ItemVersion."""
|
||||
version = MagicMock()
|
||||
version.id = version_id or uuid.uuid4()
|
||||
version.item_id = item_id or uuid.uuid4()
|
||||
version.proposed_text = proposed_text
|
||||
version.status = status
|
||||
version.diff_text = None
|
||||
version.rationale = None
|
||||
version.decision_id = None
|
||||
version.proposed_by_id = None
|
||||
version.created_at = datetime.now(timezone.utc)
|
||||
return version
|
||||
|
||||
|
||||
def _make_document(
|
||||
doc_id: uuid.UUID | None = None,
|
||||
slug: str = "test-doc",
|
||||
title: str = "Document de test",
|
||||
doc_type: str = "licence",
|
||||
version: str = "1.0.0",
|
||||
status: str = "active",
|
||||
description: str | None = "Description de test",
|
||||
items: list | None = None,
|
||||
) -> MagicMock:
|
||||
"""Create a mock Document."""
|
||||
doc = MagicMock()
|
||||
doc.id = doc_id or uuid.uuid4()
|
||||
doc.slug = slug
|
||||
doc.title = title
|
||||
doc.doc_type = doc_type
|
||||
doc.version = version
|
||||
doc.status = status
|
||||
doc.description = description
|
||||
doc.ipfs_cid = None
|
||||
doc.chain_anchor = None
|
||||
doc.items = items or []
|
||||
doc.created_at = datetime.now(timezone.utc)
|
||||
doc.updated_at = datetime.now(timezone.utc)
|
||||
return doc
|
||||
|
||||
|
||||
def _make_async_db(
|
||||
item: MagicMock | None = None,
|
||||
version: MagicMock | None = None,
|
||||
other_versions: list[MagicMock] | None = None,
|
||||
) -> AsyncMock:
|
||||
"""Create a mock async database session.
|
||||
|
||||
The mock session's execute() returns appropriate results based on
|
||||
the query being run. It supports multiple sequential calls:
|
||||
1st call -> item lookup
|
||||
2nd call -> version lookup
|
||||
3rd call (optional) -> other versions lookup (for apply_version)
|
||||
"""
|
||||
db = AsyncMock()
|
||||
|
||||
call_results = []
|
||||
|
||||
# Item result
|
||||
item_result = MagicMock()
|
||||
item_result.scalar_one_or_none.return_value = item
|
||||
call_results.append(item_result)
|
||||
|
||||
# Version result
|
||||
version_result = MagicMock()
|
||||
version_result.scalar_one_or_none.return_value = version
|
||||
call_results.append(version_result)
|
||||
|
||||
# Other versions result (for apply_version)
|
||||
if other_versions is not None:
|
||||
other_result = MagicMock()
|
||||
other_scalars = MagicMock()
|
||||
other_scalars.__iter__ = MagicMock(return_value=iter(other_versions))
|
||||
other_result.scalars.return_value = other_scalars
|
||||
call_results.append(other_result)
|
||||
|
||||
db.execute = AsyncMock(side_effect=call_results)
|
||||
db.commit = AsyncMock()
|
||||
db.refresh = AsyncMock()
|
||||
|
||||
return db
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: apply_version
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestApplyVersion:
|
||||
"""Test document_service.apply_version."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_apply_version_updates_text(self):
|
||||
"""Applying a version replaces item's current_text with proposed_text."""
|
||||
item_id = uuid.uuid4()
|
||||
version_id = uuid.uuid4()
|
||||
|
||||
item = _make_item(item_id=item_id, current_text="Ancien texte")
|
||||
version = _make_version(
|
||||
version_id=version_id,
|
||||
item_id=item_id,
|
||||
proposed_text="Nouveau texte",
|
||||
)
|
||||
|
||||
db = _make_async_db(item=item, version=version, other_versions=[])
|
||||
|
||||
result = await apply_version(item_id, version_id, db)
|
||||
|
||||
assert result.current_text == "Nouveau texte"
|
||||
assert version.status == "accepted"
|
||||
db.commit.assert_awaited_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_apply_version_rejects_other_pending(self):
|
||||
"""Applying a version rejects other pending/voting versions."""
|
||||
item_id = uuid.uuid4()
|
||||
version_id = uuid.uuid4()
|
||||
|
||||
item = _make_item(item_id=item_id)
|
||||
version = _make_version(version_id=version_id, item_id=item_id)
|
||||
|
||||
other_v1 = _make_version(item_id=item_id, status="proposed")
|
||||
other_v2 = _make_version(item_id=item_id, status="voting")
|
||||
|
||||
db = _make_async_db(
|
||||
item=item,
|
||||
version=version,
|
||||
other_versions=[other_v1, other_v2],
|
||||
)
|
||||
|
||||
await apply_version(item_id, version_id, db)
|
||||
|
||||
assert other_v1.status == "rejected"
|
||||
assert other_v2.status == "rejected"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_apply_version_item_not_found(self):
|
||||
"""ValueError is raised when item does not exist."""
|
||||
db = _make_async_db(item=None, version=None)
|
||||
|
||||
with pytest.raises(ValueError, match="Element de document introuvable"):
|
||||
await apply_version(uuid.uuid4(), uuid.uuid4(), db)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_apply_version_version_not_found(self):
|
||||
"""ValueError is raised when version does not exist."""
|
||||
item = _make_item()
|
||||
db = _make_async_db(item=item, version=None)
|
||||
|
||||
with pytest.raises(ValueError, match="Version introuvable"):
|
||||
await apply_version(item.id, uuid.uuid4(), db)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_apply_version_wrong_item(self):
|
||||
"""ValueError is raised when version belongs to a different item."""
|
||||
item_id = uuid.uuid4()
|
||||
other_item_id = uuid.uuid4()
|
||||
version_id = uuid.uuid4()
|
||||
|
||||
item = _make_item(item_id=item_id)
|
||||
version = _make_version(version_id=version_id, item_id=other_item_id)
|
||||
|
||||
db = _make_async_db(item=item, version=version)
|
||||
|
||||
with pytest.raises(ValueError, match="n'appartient pas"):
|
||||
await apply_version(item_id, version_id, db)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: reject_version
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRejectVersion:
|
||||
"""Test document_service.reject_version."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reject_version_sets_status(self):
|
||||
"""Rejecting a version sets its status to 'rejected'."""
|
||||
item_id = uuid.uuid4()
|
||||
version_id = uuid.uuid4()
|
||||
|
||||
item = _make_item(item_id=item_id)
|
||||
version = _make_version(
|
||||
version_id=version_id,
|
||||
item_id=item_id,
|
||||
status="proposed",
|
||||
)
|
||||
|
||||
db = _make_async_db(item=item, version=version)
|
||||
|
||||
result = await reject_version(item_id, version_id, db)
|
||||
|
||||
assert result.status == "rejected"
|
||||
db.commit.assert_awaited_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reject_version_item_not_found(self):
|
||||
"""ValueError is raised when item does not exist."""
|
||||
db = _make_async_db(item=None, version=None)
|
||||
|
||||
with pytest.raises(ValueError, match="Element de document introuvable"):
|
||||
await reject_version(uuid.uuid4(), uuid.uuid4(), db)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reject_version_version_not_found(self):
|
||||
"""ValueError is raised when version does not exist."""
|
||||
item = _make_item()
|
||||
db = _make_async_db(item=item, version=None)
|
||||
|
||||
with pytest.raises(ValueError, match="Version introuvable"):
|
||||
await reject_version(item.id, uuid.uuid4(), db)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reject_version_wrong_item(self):
|
||||
"""ValueError is raised when version belongs to a different item."""
|
||||
item_id = uuid.uuid4()
|
||||
other_item_id = uuid.uuid4()
|
||||
version_id = uuid.uuid4()
|
||||
|
||||
item = _make_item(item_id=item_id)
|
||||
version = _make_version(version_id=version_id, item_id=other_item_id)
|
||||
|
||||
db = _make_async_db(item=item, version=version)
|
||||
|
||||
with pytest.raises(ValueError, match="n'appartient pas"):
|
||||
await reject_version(item_id, version_id, db)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: serialize_document_to_text
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSerializeDocumentToText:
|
||||
"""Test document serialization for archival."""
|
||||
|
||||
def test_basic_serialization(self):
|
||||
"""A document with items serializes to the expected text format."""
|
||||
doc_id = uuid.uuid4()
|
||||
|
||||
item1 = _make_item(
|
||||
document_id=doc_id,
|
||||
position="1",
|
||||
title="Preambule",
|
||||
item_type="preamble",
|
||||
current_text="Le texte du preambule.",
|
||||
sort_order=0,
|
||||
)
|
||||
item2 = _make_item(
|
||||
document_id=doc_id,
|
||||
position="2",
|
||||
title="Article premier",
|
||||
item_type="clause",
|
||||
current_text="Le texte de l'article premier.",
|
||||
sort_order=1,
|
||||
)
|
||||
item3 = _make_item(
|
||||
document_id=doc_id,
|
||||
position="2.1",
|
||||
title=None,
|
||||
item_type="rule",
|
||||
current_text="Sous-article sans titre.",
|
||||
sort_order=2,
|
||||
)
|
||||
|
||||
doc = _make_document(
|
||||
doc_id=doc_id,
|
||||
title="Licence G1",
|
||||
version="2.0.0",
|
||||
doc_type="licence",
|
||||
status="active",
|
||||
description="La licence monetaire de la G1",
|
||||
items=[item1, item2, item3],
|
||||
)
|
||||
|
||||
text = serialize_document_to_text(doc)
|
||||
|
||||
assert "# Licence G1" in text
|
||||
assert "Version: 2.0.0" in text
|
||||
assert "Type: licence" in text
|
||||
assert "Statut: active" in text
|
||||
assert "Description: La licence monetaire de la G1" in text
|
||||
|
||||
# Items
|
||||
assert "## 1 - Preambule [preamble]" in text
|
||||
assert "Le texte du preambule." in text
|
||||
assert "## 2 - Article premier [clause]" in text
|
||||
assert "Le texte de l'article premier." in text
|
||||
assert "## 2.1 [rule]" in text
|
||||
assert "Sous-article sans titre." in text
|
||||
|
||||
def test_serialization_ordering(self):
|
||||
"""Items are serialized in sort_order, not insertion order."""
|
||||
doc_id = uuid.uuid4()
|
||||
|
||||
item_b = _make_item(
|
||||
document_id=doc_id,
|
||||
position="2",
|
||||
title="Second",
|
||||
current_text="Texte B",
|
||||
sort_order=1,
|
||||
)
|
||||
item_a = _make_item(
|
||||
document_id=doc_id,
|
||||
position="1",
|
||||
title="Premier",
|
||||
current_text="Texte A",
|
||||
sort_order=0,
|
||||
)
|
||||
|
||||
# Insert in reverse order
|
||||
doc = _make_document(doc_id=doc_id, items=[item_b, item_a])
|
||||
|
||||
text = serialize_document_to_text(doc)
|
||||
|
||||
# "Premier" should appear before "Second"
|
||||
idx_a = text.index("Texte A")
|
||||
idx_b = text.index("Texte B")
|
||||
assert idx_a < idx_b, "Items should be ordered by sort_order"
|
||||
|
||||
def test_serialization_without_description(self):
|
||||
"""Document without description omits that line."""
|
||||
doc = _make_document(description=None, items=[])
|
||||
text = serialize_document_to_text(doc)
|
||||
|
||||
assert "Description:" not in text
|
||||
|
||||
def test_serialization_hash_is_deterministic(self):
|
||||
"""Same document content produces the same SHA-256 hash."""
|
||||
doc_id = uuid.uuid4()
|
||||
item = _make_item(
|
||||
document_id=doc_id,
|
||||
position="1",
|
||||
title="Test",
|
||||
current_text="Contenu identique",
|
||||
sort_order=0,
|
||||
)
|
||||
|
||||
doc1 = _make_document(doc_id=doc_id, title="Doc", version="1.0", items=[item])
|
||||
doc2 = _make_document(doc_id=doc_id, title="Doc", version="1.0", items=[item])
|
||||
|
||||
text1 = serialize_document_to_text(doc1)
|
||||
text2 = serialize_document_to_text(doc2)
|
||||
|
||||
hash1 = hashlib.sha256(text1.encode("utf-8")).hexdigest()
|
||||
hash2 = hashlib.sha256(text2.encode("utf-8")).hexdigest()
|
||||
|
||||
assert hash1 == hash2
|
||||
|
||||
def test_empty_document(self):
|
||||
"""A document with no items serializes header only."""
|
||||
doc = _make_document(items=[])
|
||||
text = serialize_document_to_text(doc)
|
||||
|
||||
assert "# Document de test" in text
|
||||
assert "Version: 1.0.0" in text
|
||||
# Should end with just a newline after the header block
|
||||
lines = text.strip().split("\n")
|
||||
assert len(lines) >= 4 # title, version, type, status
|
||||
Reference in New Issue
Block a user