bruno
ef7a3ff1ed
CI / lint (push) Failing after 25s
CI / test (push) Skipped
CI / build (push) Skipped
CI / e2e (push) Skipped
CI / security (push) Successful in 28s
Desktop Build / build-windows (push) Canceled after 0s
Desktop Build / build-linux (push) Canceled after 0s
inotify ne voit pas les ecritures faites depuis d'autres clients NFS (Obsidian sur Windows), donc index_updated n'etait jamais emis et la section Recent du sidebar ne se rafraichissait pas. Detection du fstype par vault via /proc/mounts -> PollingObserver (stat-based) sur mounts reseau, Observer natif conserve sur les mounts locaux.
264 lines
11 KiB
Python
264 lines
11 KiB
Python
# tests/test_watcher.py — Tests for the file watcher (mocked)
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
|
|
import pytest
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# IGNORED_DIRS
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestIgnoredDirs:
|
|
def test_default_ignored_dirs(self):
|
|
old = os.environ.pop("OBSIGATE_IGNORED_DIRS", None)
|
|
try:
|
|
import importlib
|
|
import backend.indexer
|
|
import backend.watcher
|
|
importlib.reload(backend.indexer)
|
|
importlib.reload(backend.watcher)
|
|
dirs = backend.watcher.IGNORED_DIRS
|
|
assert ".obsidian" in dirs
|
|
assert ".trash" in dirs
|
|
assert ".git" in dirs
|
|
finally:
|
|
if old is not None:
|
|
os.environ["OBSIGATE_IGNORED_DIRS"] = old
|
|
|
|
def test_custom_ignored_dirs(self):
|
|
os.environ["OBSIGATE_IGNORED_DIRS"] = ".custom1,.custom2"
|
|
try:
|
|
import importlib
|
|
import backend.indexer
|
|
import backend.watcher
|
|
importlib.reload(backend.indexer)
|
|
importlib.reload(backend.watcher)
|
|
dirs = backend.watcher.IGNORED_DIRS
|
|
assert ".custom1" in dirs
|
|
assert ".custom2" in dirs
|
|
finally:
|
|
os.environ.pop("OBSIGATE_IGNORED_DIRS", None)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# VaultEventHandler (_is_relevant only — no event loop needed)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestIsRelevant:
|
|
@pytest.fixture
|
|
def handler(self):
|
|
from backend.watcher import VaultEventHandler
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
q = asyncio.Queue()
|
|
return VaultEventHandler("TestVault", q, loop)
|
|
|
|
def test_markdown(self, handler):
|
|
assert handler._is_relevant("/vault/test.md") is True
|
|
|
|
def test_supported_ext(self, handler):
|
|
assert handler._is_relevant("/vault/file.py") is True
|
|
|
|
def test_ignored_dir(self, handler):
|
|
assert handler._is_relevant("/vault/.git/config") is False
|
|
|
|
def test_unsupported(self, handler):
|
|
assert handler._is_relevant("/vault/file.bin") is False
|
|
|
|
def test_special_names(self, handler):
|
|
assert handler._is_relevant("/vault/Dockerfile") is True
|
|
assert handler._is_relevant("/vault/Makefile") is True
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# Network-mount detection (NFS/SMB → polling watcher required)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestNetworkMountDetection:
|
|
MOUNTS = [
|
|
"overlay / overlay rw 0 0",
|
|
"server:/export/NFS/OBSIDIAN_DOC /vaults/Obsidian_IT nfs4 rw 0 0",
|
|
"server:/export/home/bruno /vaults/bruno nfs4 rw 0 0",
|
|
"dev/sda1 /data ext4 rw 0 0",
|
|
"srv:/share /mnt/smb cifs rw 0 0",
|
|
]
|
|
|
|
def test_nfs_mount_detected(self):
|
|
from backend.watcher import find_mount_fstype
|
|
ft = find_mount_fstype("/vaults/Obsidian_IT/note.md", self.MOUNTS)
|
|
assert ft == "nfs4"
|
|
|
|
def test_nested_nfs_mount(self):
|
|
from backend.watcher import find_mount_fstype
|
|
ft = find_mount_fstype("/vaults/bruno/sub/dir", self.MOUNTS)
|
|
assert ft == "nfs4"
|
|
|
|
def test_smb_mount_detected(self):
|
|
from backend.watcher import find_mount_fstype
|
|
ft = find_mount_fstype("/mnt/smb/file.md", self.MOUNTS)
|
|
assert ft == "cifs"
|
|
|
|
def test_local_fs_not_network(self):
|
|
from backend.watcher import find_mount_fstype
|
|
ft = find_mount_fstype("/data/file.md", self.MOUNTS)
|
|
assert ft == "ext4"
|
|
|
|
def test_unmatched_path_returns_empty(self):
|
|
from backend.watcher import find_mount_fstype
|
|
assert find_mount_fstype("/nowhere/file.md", self.MOUNTS) == ""
|
|
|
|
def test_longest_prefix_wins(self):
|
|
from backend.watcher import find_mount_fstype
|
|
lines = self.MOUNTS + ["tmpfs /vaults/Obsidian_IT/tmp tmpfs rw 0 0"]
|
|
assert find_mount_fstype("/vaults/Obsidian_IT/tmp/x", lines) == "tmpfs"
|
|
assert find_mount_fstype("/vaults/Obsidian_IT/other", lines) == "nfs4"
|
|
|
|
def test_escaped_space_in_mount_point(self):
|
|
from backend.watcher import find_mount_fstype
|
|
lines = ["srv:/x /vaults/My\\040Vault nfs rw 0 0"]
|
|
assert find_mount_fstype("/vaults/My Vault/a.md", lines) == "nfs"
|
|
|
|
def test_is_network_mount_false_on_windows(self, monkeypatch):
|
|
import backend.watcher as w
|
|
monkeypatch.setattr(w.os, "name", "nt")
|
|
assert w.is_network_mount("/vaults/x") is False
|
|
|
|
def test_watch_vault_uses_polling_for_network_mount(self, monkeypatch):
|
|
"""_watch_vault must pick PollingObserver when is_network_mount is True."""
|
|
import asyncio
|
|
from unittest.mock import MagicMock
|
|
import backend.watcher as w
|
|
|
|
watcher = w.VaultWatcher(on_file_change=MagicMock())
|
|
monkeypatch.setattr(w, "is_network_mount", lambda p: True)
|
|
mock_poll = MagicMock()
|
|
monkeypatch.setattr(w, "PollingObserver", mock_poll)
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
loop.run_until_complete(
|
|
watcher._watch_vault("NetVault", tmpdir, loop)
|
|
)
|
|
mock_poll.assert_called_once()
|
|
assert "NetVault" in watcher.observers
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# VaultWatcher (unit tests with mocks)
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestVaultWatcher:
|
|
@pytest.fixture
|
|
def watcher(self):
|
|
from backend.watcher import VaultWatcher
|
|
return VaultWatcher(on_file_change=MagicMock())
|
|
|
|
def test_initial_state(self, watcher):
|
|
assert watcher._running is False
|
|
assert len(watcher.observers) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_without_start(self, watcher):
|
|
await watcher.stop()
|
|
assert watcher._running is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_calls_callback(self, watcher):
|
|
"""_dispatch should call on_file_change with events."""
|
|
events = [{'type': 'modified', 'vault': 'TestVault', 'src': '/tmp/test.md'}]
|
|
await watcher._dispatch(events)
|
|
watcher.on_file_change.assert_called_once_with(events)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispatch_error_doesnt_crash(self, watcher):
|
|
"""If on_file_change raises, _dispatch should catch the error."""
|
|
watcher.on_file_change = AsyncMock(side_effect=Exception("Boom"))
|
|
events = [{'type': 'modified', 'vault': 'TestVault', 'src': '/tmp/test.md'}]
|
|
# Should not raise
|
|
await watcher._dispatch(events)
|
|
|
|
def test_watch_vault_nonexistent_path(self, watcher):
|
|
"""Watching a nonexistent path should log a warning, not crash."""
|
|
import asyncio
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
coro = watcher._watch_vault("Fake", "/nonexistent/path/xyz", loop)
|
|
loop.run_until_complete(coro)
|
|
# No observer should be created
|
|
assert "Fake" not in watcher.observers
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_stop_with_mock_observer(self, watcher):
|
|
"""Test start and stop with a patch on the Observer."""
|
|
with patch('backend.watcher.Observer') as mock_obs_cls:
|
|
mock_obs = MagicMock()
|
|
mock_obs_cls.return_value = mock_obs
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
vault_path = str(Path(tmpdir) / "vault")
|
|
Path(vault_path).mkdir()
|
|
await watcher.start({"TestVault": vault_path})
|
|
assert watcher._running is True
|
|
assert "TestVault" in watcher.observers
|
|
|
|
await watcher.stop()
|
|
assert watcher._running is False
|
|
mock_obs.stop.assert_called_once()
|
|
mock_obs.join.assert_called_once()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# _on_vault_change integration
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
class TestOnVaultChange:
|
|
@pytest.fixture
|
|
def setup_vault(self):
|
|
import tempfile
|
|
import shutil
|
|
from pathlib import Path
|
|
from backend.indexer import build_index, index
|
|
import asyncio
|
|
|
|
tmp = Path(tempfile.mkdtemp())
|
|
vault = tmp / "TestVault"
|
|
vault.mkdir()
|
|
(vault / "test.md").write_text("# Watcher Test", encoding="utf-8")
|
|
|
|
orig_v1 = os.environ.get("VAULT_1_NAME")
|
|
orig_v2 = os.environ.get("VAULT_1_PATH")
|
|
|
|
os.environ["VAULT_1_NAME"] = "TestVault"
|
|
os.environ["VAULT_1_PATH"] = str(vault)
|
|
os.environ["OBSIGATE_AUTH_ENABLED"] = "false"
|
|
|
|
for k in list(index.keys()):
|
|
del index[k]
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
loop.run_until_complete(build_index())
|
|
|
|
yield vault
|
|
|
|
shutil.rmtree(str(tmp), ignore_errors=True)
|
|
for k in (orig_v1, orig_v2):
|
|
if k:
|
|
os.environ.pop(k, None)
|
|
|
|
def test_vault_change_modified(self, setup_vault):
|
|
from backend.main import _on_vault_change
|
|
vault = setup_vault
|
|
events = [{
|
|
'type': 'modified', 'vault': 'TestVault',
|
|
'src': str(vault / "test.md"),
|
|
'dest': None, 'timestamp': 1234.0,
|
|
}]
|
|
asyncio.run(_on_vault_change(events))
|
|
# Should not crash
|