Bruno Charest
8d1b766947
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%
383 lines
17 KiB
Python
383 lines
17 KiB
Python
# 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
|