2bdc731639
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>
419 lines
13 KiB
Python
419 lines
13 KiB
Python
"""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
|