252 lines
10 KiB
Python
252 lines
10 KiB
Python
"""Tests des vaults DIR « Spark » : exclusions par vault et lecture seule.
|
|
|
|
Couvre le contrat DIR_N_READONLY / DIR_N_IGNORED_DIRS / DIR_N_IGNORED_PATHS :
|
|
- parsing des trois variables dans load_vault_config() ;
|
|
- helper chemin_exclu (noms de dossiers et motifs fnmatch à préfixes) ;
|
|
- navigation filtrée, lectures exclues en 404 (vue, raw, téléchargement),
|
|
fichier non exclu lisible, indexation et list_paths filtrés ;
|
|
- mutation refusée en 403 sur un vault déclaré readonly ;
|
|
- /api/vaults expose readonly=true pour un DIR readonly.
|
|
|
|
Le vault de test est un répertoire temporaire monté en DIR (jamais un
|
|
vault de notes), avec une arborescence qui imite la vue Spark visée :
|
|
Console/, Workspaces/ (avec un .home), Executeurs/Hermes/{missions,data,.env}.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
from typing import ClassVar
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
VAULT_NAME = "SparkTest"
|
|
|
|
|
|
def _build_spark_tree(root: Path) -> None:
|
|
"""Crée l'arborescence technique de test sous *root*."""
|
|
(root / "Console").mkdir(parents=True)
|
|
(root / "Console" / "code.py").write_text('print("visible")\n', encoding="utf-8")
|
|
(root / "Console" / "notes.md").write_text("# Notes\n", encoding="utf-8")
|
|
|
|
atelier = root / "Workspaces" / "atelier"
|
|
(atelier / ".home").mkdir(parents=True)
|
|
(atelier / "note.md").write_text("# Atelier\n", encoding="utf-8")
|
|
(atelier / ".home" / "cache.md").write_text("# Cache du sandbox\n", encoding="utf-8")
|
|
|
|
venv = root / ".venv"
|
|
venv.mkdir()
|
|
(venv / "lib.py").write_text("# lib du venv\n", encoding="utf-8")
|
|
|
|
hermes = root / "Executeurs" / "Hermes"
|
|
(hermes / "missions").mkdir(parents=True)
|
|
(hermes / "data").mkdir(parents=True)
|
|
(hermes / "missions" / "mission.md").write_text("# Mission\n", encoding="utf-8")
|
|
(hermes / "data" / "config.md").write_text("# Profil de l'exécuteur\n", encoding="utf-8")
|
|
(hermes / ".env").write_text("CLE_FICTIVE=ne-pas-exposer\n", encoding="utf-8")
|
|
|
|
(root / "racine.md").write_text("# Racine\n", encoding="utf-8")
|
|
|
|
|
|
@pytest.fixture
|
|
def spark_dir(tmp_path: Path) -> Path:
|
|
root = tmp_path / "SparkDir"
|
|
root.mkdir()
|
|
_build_spark_tree(root)
|
|
return root
|
|
|
|
|
|
@pytest.fixture
|
|
def spark_client(spark_dir: Path):
|
|
"""TestClient avec le répertoire de test monté en vault DIR readonly.
|
|
|
|
Même patron que le fixture ``app_with_vault`` de conftest.py, mais en
|
|
DIR_N_* avec les trois nouvelles variables posées.
|
|
"""
|
|
os.environ["DIR_1_NAME"] = VAULT_NAME
|
|
os.environ["DIR_1_PATH"] = str(spark_dir)
|
|
os.environ["DIR_1_READONLY"] = "true"
|
|
os.environ["DIR_1_IGNORED_DIRS"] = ".home,.venv"
|
|
os.environ["DIR_1_IGNORED_PATHS"] = "Executeurs/*/data,Executeurs/*/.env"
|
|
os.environ["OBSIGATE_AUTH_ENABLED"] = "false"
|
|
|
|
import backend.main
|
|
|
|
backend.main._load_config = lambda: {"watcher_enabled": False}
|
|
|
|
from backend.indexer import build_index
|
|
from backend.main import app
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
loop.run_until_complete(build_index())
|
|
|
|
return TestClient(app)
|
|
|
|
|
|
# ── Parsing de la configuration ─────────────────────────────────────
|
|
|
|
|
|
class TestParsingConfig:
|
|
def test_dir_variables_parsees(self, monkeypatch, tmp_path):
|
|
monkeypatch.setenv("DIR_1_NAME", "SparkParse")
|
|
monkeypatch.setenv("DIR_1_PATH", str(tmp_path))
|
|
monkeypatch.setenv("DIR_1_READONLY", "true")
|
|
monkeypatch.setenv("DIR_1_IGNORED_DIRS", ".home, .venv ,__pycache__")
|
|
monkeypatch.setenv("DIR_1_IGNORED_PATHS", "Executeurs/*/data, Executeurs/*/.env")
|
|
|
|
from backend.indexer import load_vault_config
|
|
|
|
cfg = load_vault_config()["SparkParse"]
|
|
assert cfg["type"] == "DIR"
|
|
assert cfg["readonly"] is True
|
|
assert cfg["ignoredDirs"] == {".home", ".venv", "__pycache__"}
|
|
assert cfg["ignoredPaths"] == ["Executeurs/*/data", "Executeurs/*/.env"]
|
|
|
|
def test_dir_sans_variables_defauts(self, monkeypatch, tmp_path):
|
|
monkeypatch.setenv("DIR_1_NAME", "DirNu")
|
|
monkeypatch.setenv("DIR_1_PATH", str(tmp_path))
|
|
|
|
from backend.indexer import load_vault_config
|
|
|
|
cfg = load_vault_config()["DirNu"]
|
|
assert cfg["readonly"] is False
|
|
assert cfg["ignoredDirs"] == set()
|
|
assert cfg["ignoredPaths"] == []
|
|
|
|
|
|
# ── Helper chemin_exclu ─────────────────────────────────────────────
|
|
|
|
|
|
class TestCheminExclu:
|
|
CFG: ClassVar[dict] = {
|
|
"ignoredDirs": {".home", ".venv"},
|
|
"ignoredPaths": ["Executeurs/*/data", "Executeurs/*/.env"],
|
|
}
|
|
|
|
def test_nom_de_dossier_a_toute_profondeur(self):
|
|
from backend.services.vaults import chemin_exclu_dans_config
|
|
|
|
assert chemin_exclu_dans_config(self.CFG, "Workspaces/atelier/.home/cache.md")
|
|
assert chemin_exclu_dans_config(self.CFG, ".venv/lib.py")
|
|
assert chemin_exclu_dans_config(self.CFG, ".home")
|
|
|
|
def test_motif_chemin_et_sous_arbre(self):
|
|
from backend.services.vaults import chemin_exclu_dans_config
|
|
|
|
assert chemin_exclu_dans_config(self.CFG, "Executeurs/Hermes/data")
|
|
assert chemin_exclu_dans_config(self.CFG, "Executeurs/Hermes/data/config.md")
|
|
assert chemin_exclu_dans_config(self.CFG, "Executeurs/Pi/data/x/y.py")
|
|
assert chemin_exclu_dans_config(self.CFG, "Executeurs/Hermes/.env")
|
|
|
|
def test_non_exclus_et_tolerance(self):
|
|
from backend.services.vaults import chemin_exclu_dans_config
|
|
|
|
assert not chemin_exclu_dans_config(self.CFG, "Console/code.py")
|
|
assert not chemin_exclu_dans_config(self.CFG, "Executeurs/Hermes/missions/mission.md")
|
|
assert not chemin_exclu_dans_config({}, "Console/code.py")
|
|
assert not chemin_exclu_dans_config(None, "Console/code.py")
|
|
assert not chemin_exclu_dans_config(self.CFG, "")
|
|
|
|
|
|
# ── Navigation et lectures via l'API ────────────────────────────────
|
|
|
|
|
|
class TestNavigationEtLecture:
|
|
def test_browse_racine_cache_les_dossiers_ignores(self, spark_client):
|
|
resp = spark_client.get(f"/api/browse/{VAULT_NAME}")
|
|
assert resp.status_code == 200
|
|
names = {item["name"] for item in resp.json()["items"]}
|
|
assert {"Console", "Workspaces", "Executeurs", "racine.md"} <= names
|
|
assert ".venv" not in names
|
|
|
|
def test_browse_sous_dossier_cache_data(self, spark_client):
|
|
resp = spark_client.get(f"/api/browse/{VAULT_NAME}", params={"path": "Executeurs/Hermes"})
|
|
assert resp.status_code == 200
|
|
names = {item["name"] for item in resp.json()["items"]}
|
|
assert "missions" in names
|
|
assert "data" not in names
|
|
|
|
def test_browse_chemin_exclu_404(self, spark_client):
|
|
resp = spark_client.get(f"/api/browse/{VAULT_NAME}", params={"path": "Executeurs/Hermes/data"})
|
|
assert resp.status_code == 404
|
|
|
|
def test_vue_fichier_exclu_404(self, spark_client):
|
|
resp = spark_client.get(f"/api/file/{VAULT_NAME}", params={"path": "Executeurs/Hermes/data/config.md"})
|
|
assert resp.status_code == 404
|
|
|
|
def test_raw_exclu_404(self, spark_client):
|
|
resp = spark_client.get(f"/api/file/{VAULT_NAME}/raw", params={"path": "Executeurs/Hermes/data/config.md"})
|
|
assert resp.status_code == 404
|
|
|
|
def test_raw_env_exclu_404(self, spark_client):
|
|
resp = spark_client.get(f"/api/file/{VAULT_NAME}/raw", params={"path": "Executeurs/Hermes/.env"})
|
|
assert resp.status_code == 404
|
|
assert "CLE_FICTIVE" not in resp.text
|
|
|
|
def test_telechargement_env_exclu_404(self, spark_client):
|
|
resp = spark_client.get(f"/api/file/{VAULT_NAME}/download", params={"path": "Executeurs/Hermes/.env"})
|
|
assert resp.status_code == 404
|
|
assert "CLE_FICTIVE" not in resp.text
|
|
|
|
def test_fichier_non_exclu_lisible(self, spark_client):
|
|
resp = spark_client.get(f"/api/file/{VAULT_NAME}/raw", params={"path": "Console/code.py"})
|
|
assert resp.status_code == 200
|
|
assert "visible" in resp.json()["raw"]
|
|
|
|
|
|
# ── Indexation et liste des chemins ─────────────────────────────────
|
|
|
|
|
|
class TestIndexation:
|
|
def test_index_ne_contient_aucun_exclu(self, spark_client):
|
|
from backend.indexer import get_vault_data
|
|
|
|
data = get_vault_data(VAULT_NAME)
|
|
assert data is not None
|
|
file_paths = {f["path"] for f in data.get("files", [])}
|
|
assert "Console/code.py" in file_paths
|
|
for p in file_paths:
|
|
segments = p.split("/")
|
|
assert ".home" not in segments and ".venv" not in segments
|
|
assert "data" not in segments[:-1] or not p.startswith("Executeurs/")
|
|
path_entries = {e["path"] for e in data.get("paths", [])}
|
|
assert "Executeurs/Hermes/data" not in path_entries
|
|
assert "Workspaces/atelier/.home" not in path_entries
|
|
|
|
def test_list_paths_filtre_les_exclus(self, spark_client):
|
|
from backend.services.search import list_paths
|
|
|
|
result = list_paths(VAULT_NAME)
|
|
paths = {r["path"] for r in result["results"]}
|
|
assert "Console/code.py" in paths
|
|
assert not any("/data/" in f"/{p}/" and p.startswith("Executeurs/") for p in paths)
|
|
assert not any(".home" in p.split("/") or ".venv" in p.split("/") for p in paths)
|
|
|
|
|
|
# ── Lecture seule ───────────────────────────────────────────────────
|
|
|
|
|
|
class TestLectureSeule:
|
|
def test_save_refuse_403(self, spark_client):
|
|
resp = spark_client.put(
|
|
f"/api/file/{VAULT_NAME}/save",
|
|
params={"path": "Console/code.py"},
|
|
json={"content": "print('modifié')\n"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
def test_creation_dossier_refusee_403(self, spark_client):
|
|
resp = spark_client.post(f"/api/directory/{VAULT_NAME}", json={"path": "NouveauDossier"})
|
|
assert resp.status_code == 403
|
|
|
|
def test_vaults_expose_readonly(self, spark_client):
|
|
resp = spark_client.get("/api/vaults")
|
|
assert resp.status_code == 200
|
|
entry = next(v for v in resp.json() if v["name"] == VAULT_NAME)
|
|
assert entry["type"] == "DIR"
|
|
assert entry["readonly"] is True
|