ObsiGate/tests/test_watcher.py

# tests/test_watcher.py — Tests for the file watcher (mocked)
import asyncio
import os
from pathlib import Path
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.watcher
            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.watcher
            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


# ═══════════════════════════════════════════════════════════════════
# 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, 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


import tempfile