test: expand coverage to 49% (+78 new tests, 175 total)

Add 78 new tests targeting high-impact uncovered modules: - tests/test_search_advanced.py (23 tests): InvertedIndex CRUD, search/advanced_search/suggest functions, tag/title indexing - tests/test_indexer_advanced.py (15 tests): hooks, file CRUD, path index, lookup, generation counter - tests/test_modules.py (40 tests): audit, history, rate limit, saved searches, vault settings, webhooks, share Coverage improvements: ratelimit.py: 80% → 100% share.py: 24% → 97% saved_searches: 37% → 95% history.py: 26% → 86% audit.py: 0% → 85% search.py: 44% → 82% webhooks.py: 31% → 67% vault_settings: 31% → 69% indexer.py: 47% → 65% Overall: 35% → 49%
2026-05-27 22:32:10 -04:00
parent edb9e98f81
commit 8d1b766947
3 changed files with 774 additions and 0 deletions
@@ -0,0 +1,382 @@
+# tests/test_modules.py — Tests for utility modules: audit, history, ratelimit, settings, webhooks, share
+import os
+import json
+import time
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Audit
+# ═══════════════════════════════════════════════════════════════════
+
+class TestAudit:
+    @pytest.fixture(autouse=True)
+    def _setup(self, tmp_path, monkeypatch):
+        from backend import audit
+        self.audit_file = tmp_path / "audit.log"
+        monkeypatch.setattr(audit, "AUDIT_LOG_FILE", self.audit_file)
+        yield
+
+    def test_log_file_save(self):
+        from backend.audit import log_file_save
+        log_file_save("alice", "TestVault", "notes/test.md", 1024, ip="192.168.1.1")
+        assert self.audit_file.exists()
+        entries = []
+        with open(self.audit_file, "r", encoding="utf-8") as f:
+            for line in f:
+                if line.strip():
+                    entries.append(json.loads(line))
+        assert len(entries) == 1
+        assert entries[0]["action"] == "file_save"
+        assert entries[0]["username"] == "alice"
+        assert entries[0]["vault"] == "TestVault"
+        assert entries[0]["size"] == 1024
+
+    def test_log_file_delete(self):
+        from backend.audit import log_file_delete
+        log_file_delete("bob", "TestVault", "notes/old.md", ip="10.0.0.1")
+        assert self.audit_file.exists()
+
+    def test_log_config_change(self):
+        from backend.audit import log_config_change
+        log_config_change("admin", {"theme": "dark"}, ip="127.0.0.1")
+        assert self.audit_file.exists()
+
+    def test_log_vault_add(self):
+        from backend.audit import log_vault_add
+        log_vault_add("admin", "NewVault", "/vaults/New", ip="192.168.1.1")
+        assert self.audit_file.exists()
+
+    def test_log_vault_remove(self):
+        from backend.audit import log_vault_remove
+        log_vault_remove("admin", "OldVault", ip="192.168.1.1")
+        assert self.audit_file.exists()
+
+    def test_get_recent_entries(self):
+        from backend.audit import log_file_save, get_recent_entries
+        log_file_save("alice", "V1", "a.md", 100)
+        log_file_save("bob", "V2", "b.md", 200)
+        log_file_save("alice", "V3", "c.md", 300)
+        entries = get_recent_entries(limit=2)
+        assert len(entries) == 2
+        # Most recent first
+        assert entries[0]["path"] == "c.md"
+
+    def test_get_recent_entries_filtered(self):
+        from backend.audit import log_file_save, log_file_delete, get_recent_entries
+        log_file_save("alice", "V1", "a.md", 100)
+        log_file_delete("bob", "V1", "b.md")
+        entries = get_recent_entries(action="file_delete")
+        assert len(entries) >= 1
+        assert all(e["action"] == "file_delete" for e in entries)
+
+    def test_rotate_if_needed(self, monkeypatch):
+        from backend import audit
+        # Force rotation by setting max size very low
+        monkeypatch.setattr(audit, "MAX_LOG_SIZE", 100)
+        from backend.audit import log_file_save
+        # Write enough entries to trigger rotation
+        for i in range(20):
+            log_file_save("user", "vault", f"file{i}.md", 500)
+        # Should have created a backup
+        backup = self.audit_file.with_suffix(".log.1")
+        assert backup.exists() or self.audit_file.exists()
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Rate Limiting — full coverage
+# ═══════════════════════════════════════════════════════════════════
+
+class TestRateLimiterFull:
+    def test_cleanup_expired(self, monkeypatch):
+        """Test that expired entries are cleaned up."""
+        from backend import ratelimit
+        # Force cleanup interval to 0
+        monkeypatch.setattr(ratelimit, "CLEANUP_INTERVAL", 0)
+        # Set a very short window
+        monkeypatch.setattr(ratelimit, "WINDOW_SECONDS", 1)
+        monkeypatch.setattr(ratelimit, "_last_cleanup", 0)
+        from backend.ratelimit import record_failure, is_rate_limited
+        ip = "192.168.100.1"
+        record_failure(ip)
+        assert not is_rate_limited(ip)
+        # Wait for window to expire
+        time.sleep(1.1)
+        # Next call should trigger cleanup
+        assert not is_rate_limited(ip)
+
+    def test_get_status_with_ip(self):
+        from backend.ratelimit import get_status
+        status = get_status("10.0.0.99")
+        assert status["ip"] == "10.0.0.99"
+        assert "failures" in status
+        assert "limited" in status
+        assert status["limited"] is False
+
+
+# ═══════════════════════════════════════════════════════════════════
+# History
+# ═══════════════════════════════════════════════════════════════════
+
+class TestHistory:
+    @pytest.fixture(autouse=True)
+    def _setup(self, tmp_path, monkeypatch):
+        from backend import history
+        self.history_dir = tmp_path / "data" / "history"
+        self.history_dir.mkdir(parents=True, exist_ok=True)
+        monkeypatch.setattr(history, "HISTORY_DIR", self.history_dir)
+        yield
+
+    def test_record_open_creates_file(self):
+        from backend.history import record_open, get_recent_opened
+        record_open("alice", "TestVault", "notes/test.md", "Test Note")
+        user_file = self.history_dir / "alice.json"
+        assert user_file.exists()
+        recent = get_recent_opened("alice")
+        assert len(recent) == 1
+        assert recent[0]["vault"] == "TestVault"
+        assert recent[0]["path"] == "notes/test.md"
+
+    def test_record_open_dedup(self):
+        from backend.history import record_open, get_recent_opened
+        record_open("alice", "V1", "a.md", "A")
+        record_open("alice", "V1", "a.md", "A")
+        recent = get_recent_opened("alice")
+        assert len(recent) == 1  # Deduplicated
+
+    def test_get_recent_limit(self):
+        from backend.history import record_open, get_recent_opened
+        for i in range(25):
+            record_open("alice", "V1", f"file{i}.md", f"File {i}")
+        recent = get_recent_opened("alice", limit=5)
+        assert len(recent) <= 5
+
+    def test_get_recent_vault_filter(self):
+        from backend.history import record_open, get_recent_opened
+        record_open("alice", "V1", "a.md", "A")
+        record_open("alice", "V2", "b.md", "B")
+        recent = get_recent_opened("alice", vault_filter="V1")
+        assert len(recent) >= 1
+        assert all(r["vault"] == "V1" for r in recent)
+
+    def test_toggle_bookmark(self):
+        from backend.history import toggle_bookmark, get_bookmarks, is_bookmarked
+        toggle_bookmark("alice", "V1", "a.md", "Note A")
+        assert is_bookmarked("alice", "V1", "a.md") is True
+        bookmarks = get_bookmarks("alice")
+        assert len(bookmarks) == 1
+        # Toggle off
+        toggle_bookmark("alice", "V1", "a.md", "Note A")
+        assert is_bookmarked("alice", "V1", "a.md") is False
+
+    def test_bookmark_vault_filter(self):
+        from backend.history import toggle_bookmark, get_bookmarks
+        toggle_bookmark("alice", "V1", "a.md", "A")
+        toggle_bookmark("alice", "V2", "b.md", "B")
+        bookmarks = get_bookmarks("alice", vault_filter="V1")
+        assert len(bookmarks) >= 1
+        assert all(r["vault"] == "V1" for r in bookmarks)
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Saved Searches
+# ═══════════════════════════════════════════════════════════════════
+
+class TestSavedSearches:
+    @pytest.fixture(autouse=True)
+    def _setup(self, tmp_path, monkeypatch):
+        from backend import saved_searches
+        self.data_dir = tmp_path / "data"
+        self.data_dir.mkdir(parents=True, exist_ok=True)
+        monkeypatch.setattr(saved_searches, "DATA_DIR", self.data_dir)
+        yield
+
+    def test_get_empty(self):
+        from backend.saved_searches import get_saved
+        assert get_saved("alice") == []
+
+    def test_save_and_get(self):
+        from backend.saved_searches import save_search, get_saved
+        save_search("alice", {"query": "python tutorial", "vault": "all"})
+        searches = get_saved("alice")
+        assert len(searches) == 1
+        assert searches[0]["query"] == "python tutorial"
+
+    def test_delete_saved(self):
+        from backend.saved_searches import save_search, delete_saved, get_saved
+        save_search("alice", {"query": "test query"})
+        searches = get_saved("alice")
+        assert len(searches) == 1
+        sid = searches[0]["id"]
+        delete_saved("alice", sid)
+        assert get_saved("alice") == []
+
+    def test_delete_nonexistent(self):
+        from backend.saved_searches import delete_saved
+        # Should not raise
+        delete_saved("alice", "nonexistent-id")
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Vault Settings
+# ═══════════════════════════════════════════════════════════════════
+
+class TestVaultSettings:
+    @pytest.fixture(autouse=True)
+    def _setup(self, tmp_path, monkeypatch):
+        from backend import vault_settings
+        self.settings_dir = tmp_path / "data"
+        self.settings_dir.mkdir(parents=True, exist_ok=True)
+        self.settings_file = self.settings_dir / "vault_settings.json"
+        monkeypatch.setattr(vault_settings, "_SETTINGS_PATH", self.settings_file)
+        # Reset in-memory cache between tests
+        monkeypatch.setattr(vault_settings, "_vault_settings", {})
+        yield
+        monkeypatch.setattr(vault_settings, "_vault_settings", {})
+
+    def test_get_default(self):
+        from backend.vault_settings import get_vault_setting
+        # get_vault_setting returns settings dict or None if no settings exist
+        result = get_vault_setting("TestVault")
+        # Fresh vault has no settings — result may be None or empty dict
+        if result is not None:
+            assert isinstance(result, dict)
+
+    def test_update_and_get(self):
+        from backend.vault_settings import update_vault_setting, get_vault_setting
+        update_vault_setting("TestVault", {"hideHiddenFiles": True})
+        result = get_vault_setting("TestVault")
+        assert result is not None
+        assert result.get("hideHiddenFiles") is True
+
+    def test_get_all(self):
+        from backend.vault_settings import update_vault_setting, get_all_vault_settings
+        update_vault_setting("V1", {"hideHiddenFiles": True})
+        update_vault_setting("V2", {"hideHiddenFiles": False})
+        settings = get_all_vault_settings()
+        assert settings.get("V1", {}).get("hideHiddenFiles") is True
+        assert settings.get("V2", {}).get("hideHiddenFiles") is False
+
+    def test_delete(self):
+        from backend.vault_settings import update_vault_setting, delete_vault_setting, get_vault_setting
+        update_vault_setting("TestVault", {"hideHiddenFiles": True})
+        assert delete_vault_setting("TestVault") is True
+        result = get_vault_setting("TestVault")
+        # After deletion, settings should be None or empty
+        if result is not None:
+            assert result.get("hideHiddenFiles", False) is False
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Webhooks
+# ═══════════════════════════════════════════════════════════════════
+
+class TestWebhooks:
+    @pytest.fixture(autouse=True)
+    def _setup(self, tmp_path, monkeypatch):
+        from backend import webhooks
+        self.wh_file = tmp_path / "webhooks.json"
+        monkeypatch.setattr(webhooks, "WEBHOOKS_FILE", self.wh_file)
+        yield
+
+    def test_get_empty(self):
+        from backend.webhooks import get_webhooks
+        assert get_webhooks() == []
+
+    def test_create_webhook(self):
+        from backend.webhooks import create_webhook, get_webhooks
+        wh = create_webhook("My Hook", "https://example.com/hook", ["file_created", "file_modified"])
+        assert wh["name"] == "My Hook"
+        assert wh["enabled"] is True
+        assert len(get_webhooks()) == 1
+
+    def test_create_invalid_event_filtered(self):
+        from backend.webhooks import create_webhook
+        wh = create_webhook("Test", "https://example.com", ["file_created", "invalid_event"])
+        assert wh["events"] == ["file_created"]
+
+    def test_update_webhook(self):
+        from backend.webhooks import create_webhook, update_webhook, get_webhooks
+        wh = create_webhook("Test", "https://example.com", ["file_created"])
+        update_webhook(wh["id"], {"name": "Updated", "enabled": False})
+        updated = get_webhooks()[0]
+        assert updated["name"] == "Updated"
+        assert updated["enabled"] is False
+
+    def test_update_nonexistent(self):
+        from backend.webhooks import update_webhook
+        result = update_webhook("fake-id", {"name": "X"})
+        assert result is None
+
+    def test_delete_webhook(self):
+        from backend.webhooks import create_webhook, delete_webhook, get_webhooks
+        wh = create_webhook("Test", "https://example.com", ["file_created"])
+        assert delete_webhook(wh["id"]) is True
+        assert get_webhooks() == []
+
+    def test_delete_nonexistent(self):
+        from backend.webhooks import delete_webhook
+        assert delete_webhook("fake-id") is False
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Share
+# ═══════════════════════════════════════════════════════════════════
+
+class TestShare:
+    @pytest.fixture(autouse=True)
+    def _setup(self, tmp_path, monkeypatch):
+        from backend import share
+        self.share_file = tmp_path / "shares.json"
+        monkeypatch.setattr(share, "SHARES_FILE", self.share_file)
+        yield
+
+    def test_create_share(self):
+        from backend.share import create_share, get_share_by_token
+        share = create_share("TestVault", "notes/test.md", "alice")
+        assert len(share["token"]) == 64  # 32-byte hex
+        retrieved = get_share_by_token(share["token"])
+        assert retrieved is not None
+        assert retrieved["vault"] == "TestVault"
+
+    def test_share_expiry(self):
+        from backend.share import create_share, get_share_by_token
+        # Share that expires in the past (negative hours = already expired)
+        share = create_share("TestVault", "notes/test.md", "alice", expires_in_hours=-1)
+        retrieved = get_share_by_token(share["token"])
+        assert retrieved is None
+
+    def test_record_access(self):
+        from backend.share import create_share, record_access, get_share_by_token
+        share = create_share("TestVault", "notes/test.md", "alice")
+        record_access(share["token"])
+        retrieved = get_share_by_token(share["token"])
+        assert retrieved["access_count"] == 1
+        assert retrieved["last_accessed"] is not None
+
+    def test_revoke_share(self):
+        from backend.share import create_share, revoke_share, get_share_by_token
+        share = create_share("TestVault", "notes/test.md", "alice")
+        assert revoke_share(share["token"]) is True
+        assert get_share_by_token(share["token"]) is None
+
+    def test_revoke_nonexistent(self):
+        from backend.share import revoke_share
+        assert revoke_share("fake-token") is False
+
+    def test_list_shares(self):
+        from backend.share import create_share, list_shares
+        create_share("V1", "a.md", "alice")
+        create_share("V2", "b.md", "bob")
+        all_shares = list_shares()
+        assert len(all_shares) == 2
+        v1_shares = list_shares(vault_filter="V1")
+        assert len(v1_shares) == 1
+
+    def test_invalid_token(self):
+        from backend.share import get_share_by_token
+        assert get_share_by_token("invalid-token-12345") is None