107 lines
4.8 KiB
Python
107 lines
4.8 KiB
Python
# tests/test_sync_lists.py — #197 (Gitea #4) : réconciliation externe en un appel.
|
|
"""Couvre l'exposition de ``mtime``/``size`` dans les listes de fichiers pour
|
|
la synchronisation TikTak ↔ ObsiGate (un seul appel de liste suffit à
|
|
détecter les fichiers modifiés depuis un horodatage, sans relire chaque
|
|
fichier) :
|
|
|
|
- ``GET /api/vault/{vault}/paths`` : chaque entrée fichier expose ``size``,
|
|
``mtime`` (epoch) et ``mtime_iso`` ; les répertoires portent ``None`` ;
|
|
- ``GET /api/browse/{vault}`` : les fichiers exposent ``modified``/``modified_iso`` ;
|
|
- outil MCP ``list_directory`` : même contrat (``size``/``mtime``/``mtime_iso``) ;
|
|
- outil MCP ``list_all_files`` : déjà complet (``size``/``modified``) — parité vérifiée ;
|
|
- scénario de réconciliation : snapshot → modification via l'API → re-liste,
|
|
le fichier modifié est détecté par comparaison ``(mtime, size)``.
|
|
"""
|
|
from datetime import datetime
|
|
|
|
from backend.tools.api import call_tool
|
|
|
|
from tests.test_tools import _ctx
|
|
|
|
|
|
def _paths(client, vault="TestVault", limit=5000):
|
|
resp = client.get(f"/api/vault/{vault}/paths", params={"limit": limit})
|
|
assert resp.status_code == 200, resp.text
|
|
return resp.json()
|
|
|
|
|
|
class TestPathsMetadata:
|
|
def test_file_entries_expose_size_and_mtime(self, client):
|
|
data = _paths(client)
|
|
assert data["vault"] == "TestVault"
|
|
files = [e for e in data["results"] if e["type"] == "file"]
|
|
assert files, "le vault de test doit contenir des fichiers"
|
|
by_path = {e["path"]: e for e in files}
|
|
note = by_path["note1.md"]
|
|
assert isinstance(note["size"], int) and note["size"] > 0
|
|
assert isinstance(note["mtime"], (int, float)) and note["mtime"] > 0
|
|
assert note["mtime_iso"]
|
|
# L'epoch et l'ISO désignent le même instant (tolérance 2 s).
|
|
assert abs(datetime.fromisoformat(note["mtime_iso"]).timestamp() - note["mtime"]) < 2
|
|
|
|
def test_directory_entries_have_null_metadata(self, client):
|
|
data = _paths(client)
|
|
dirs = [e for e in data["results"] if e["type"] == "directory"]
|
|
assert dirs, "le vault de test doit contenir un répertoire (Projets)"
|
|
for entry in dirs:
|
|
assert entry["size"] is None
|
|
assert entry["mtime"] is None
|
|
assert entry["mtime_iso"] is None
|
|
|
|
def test_limit_still_applies(self, client):
|
|
data = _paths(client, limit=2)
|
|
assert data["count"] == 2
|
|
assert len(data["results"]) == 2
|
|
|
|
|
|
class TestReconciliationScenario:
|
|
def test_single_call_detects_changed_file(self, client):
|
|
"""Un client externe snapshot ``{path: (mtime, size)}`` puis, après
|
|
une modification via l'API fichiers, détecte le changement en un seul
|
|
appel de liste — sans relire aucun contenu."""
|
|
before = {e["path"]: (e["mtime"], e["size"]) for e in _paths(client)["results"]}
|
|
assert "note1.md" in before
|
|
|
|
# Modification via l'API d'écriture (même canal que la synchro TikTak).
|
|
save = client.put(
|
|
"/api/file/TestVault/save",
|
|
params={"path": "note1.md"},
|
|
json={"content": "# Introduction à Python\nContenu enrichi pour la synchro.\n"},
|
|
)
|
|
assert save.status_code == 200, save.text
|
|
|
|
after = {e["path"]: (e["mtime"], e["size"]) for e in _paths(client)["results"]}
|
|
changed = [p for p in after if after[p] != before.get(p)]
|
|
assert changed == ["note1.md"]
|
|
# Le fichier non touché n'est pas signalé.
|
|
assert "note2.md" in after and after["note2.md"] == before["note2.md"]
|
|
|
|
|
|
class TestBrowseAndMcpParity:
|
|
def test_browse_files_expose_modified(self, client):
|
|
resp = client.get("/api/browse/TestVault")
|
|
assert resp.status_code == 200, resp.text
|
|
files = [i for i in resp.json()["items"] if i["type"] == "file"]
|
|
assert files
|
|
for item in files:
|
|
assert isinstance(item["modified"], (int, float)) and item["modified"] > 0
|
|
assert item["modified_iso"]
|
|
|
|
def test_mcp_list_directory_exposes_size_and_mtime(self, client):
|
|
entries = call_tool("list_directory", _ctx(), {"vault": "TestVault"}).data
|
|
files = [e for e in entries if e["type"] == "file"]
|
|
assert files
|
|
for entry in files:
|
|
assert isinstance(entry["size"], int) and entry["size"] > 0
|
|
assert isinstance(entry["mtime"], (int, float)) and entry["mtime"] > 0
|
|
assert entry["mtime_iso"]
|
|
|
|
def test_mcp_list_all_files_parity(self, client):
|
|
"""``list_all_files`` (source unique REST + MCP) expose déjà le contrat."""
|
|
result = call_tool("list_all_files", _ctx(), {"vault": "TestVault"}).data
|
|
assert result["files"]
|
|
for entry in result["files"]:
|
|
assert isinstance(entry["size"], int)
|
|
assert isinstance(entry["modified"], (int, float))
|
|
assert entry["modified_iso"]
|