Benjamin Admin
299375e486
DSMS Stufe 3 — making the parent_cid chain useful end-to-end.
Gateway (dsms-gateway):
- /api/v1/documents/{cid}/history alias added next to the legacy
/documents/{cid}/history (history endpoint itself was already there,
just under an inconsistent prefix).
- NEW /api/v1/documents/{cid_a}/diff/{cid_b}: fetches both packages from
IPFS, computes a metadata diff (per-field old/new), and renders a
unified text diff for utf-8 payloads. Binary payloads return only
metadata diff with a "binary — compare via rendered export" note.
- 4 new pytest cases (mocking ipfs_cat): text diff, binary fallback,
fetch error, history chain depth — all green.
Frontend (admin-compliance):
- CIDHistoryModal: lazy-loads /dsms/documents/:cid/history, renders the
version chain as a vertical timeline, marks the AKTUELL entry, and
per-step exposes a "Diff zu V<n>" button that loads + renders the diff
inline (metadata table + unified text diff in a monospace panel).
- AuditTimelinePage: existing CID badge now sits next to a "Verlauf
anzeigen" link that opens the modal. Handles both Python's plain-CID
audit values and the Go techfile flow's JSON envelope {cid, filename,
size} via extractCID() helper.
This makes "show me how this CE-Akte changed between V2 and V3"
self-service in the UI instead of a curl-against-IPFS workflow.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
"""
|
|
Tests for the version-chain diff endpoint added in DSMS Stufe 3.
|
|
|
|
Mocks ipfs_cat so the test does not require a running IPFS node.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from main import app
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def _wrap(metadata: dict, content_text: str) -> str:
|
|
"""Mimic the JSON envelope that routers.documents.ipfs_cat returns."""
|
|
return json.dumps(
|
|
{
|
|
"metadata": metadata,
|
|
"content_base64": base64.b64encode(content_text.encode("utf-8")).decode("ascii"),
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_diff_text_documents_returns_unified_diff():
|
|
pkg_a = _wrap({"version": "1", "document_type": "ce_techfile"}, "alpha\nbeta\ngamma\n")
|
|
pkg_b = _wrap({"version": "2", "document_type": "ce_techfile"}, "alpha\nDELTA\ngamma\n")
|
|
|
|
async def fake_cat(cid: str):
|
|
return pkg_a if cid == "cidA" else pkg_b
|
|
|
|
with patch("routers.documents.ipfs_cat", new=AsyncMock(side_effect=fake_cat)):
|
|
resp = client.get("/api/v1/documents/cidA/diff/cidB")
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["kind"] == "text"
|
|
assert body["cid_a"] == "cidA"
|
|
assert body["cid_b"] == "cidB"
|
|
assert body["added_lines"] >= 1
|
|
assert body["removed_lines"] >= 1
|
|
assert "DELTA" in body["diff"]
|
|
assert body["metadata_diff"] == {"version": {"old": "1", "new": "2"}}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_diff_binary_documents_returns_metadata_only():
|
|
# Use raw bytes that are not utf-8 decodable
|
|
invalid_utf8 = b"\xff\xfe\xfd\xfc"
|
|
pkg_a = json.dumps(
|
|
{"metadata": {"version": "1"}, "content_base64": base64.b64encode(invalid_utf8).decode()}
|
|
)
|
|
pkg_b = json.dumps(
|
|
{"metadata": {"version": "2"}, "content_base64": base64.b64encode(invalid_utf8 + b"\x00").decode()}
|
|
)
|
|
|
|
async def fake_cat(cid: str):
|
|
return pkg_a if cid == "cidA" else pkg_b
|
|
|
|
with patch("routers.documents.ipfs_cat", new=AsyncMock(side_effect=fake_cat)):
|
|
resp = client.get("/api/v1/documents/cidA/diff/cidB")
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["kind"] == "binary"
|
|
assert body["metadata_diff"] == {"version": {"old": "1", "new": "2"}}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_diff_handles_fetch_error():
|
|
async def fake_cat(cid: str):
|
|
raise RuntimeError("not pinned")
|
|
|
|
with patch("routers.documents.ipfs_cat", new=AsyncMock(side_effect=fake_cat)):
|
|
resp = client.get("/api/v1/documents/cidA/diff/cidB")
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "error" in body
|
|
assert body["cid_a"] == "cidA"
|
|
assert body["cid_b"] == "cidB"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_history_endpoint_follows_parent_chain():
|
|
"""Sanity check that the existing history endpoint still works after route alias."""
|
|
|
|
chain = {
|
|
"v3": _wrap({"version": "3", "parent_cid": "v2"}, "x"),
|
|
"v2": _wrap({"version": "2", "parent_cid": "v1"}, "x"),
|
|
"v1": _wrap({"version": "1", "parent_cid": None}, "x"),
|
|
}
|
|
|
|
async def fake_cat(cid: str):
|
|
return chain[cid]
|
|
|
|
with patch("routers.documents.ipfs_cat", new=AsyncMock(side_effect=fake_cat)):
|
|
resp = client.get("/api/v1/documents/v3/history")
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["depth"] == 3
|
|
assert [h["version"] for h in body["history"]] == ["3", "2", "1"]
|