ObsiGate/tests/test_modules.py

# 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