learningcircuit--local-deep-research/tests/security/file_integrity/test_integrity_manager_edge_cases.py

"""High-value edge case tests for security/file_integrity/integrity_manager.py.

Tests internal methods: _normalize_path, _get_verifier_for_file,
_needs_verification, _do_verification, _update_stats, register_verifier,
and cleanup logic.
"""

from datetime import datetime, UTC
from pathlib import Path
from unittest.mock import MagicMock


from local_deep_research.security.file_integrity.integrity_manager import (
    FileIntegrityManager,
)


def _make_manager():
    """Create a FileIntegrityManager instance bypassing __init__."""
    mgr = FileIntegrityManager.__new__(FileIntegrityManager)
    mgr.username = "test"
    mgr.password = None
    mgr.verifiers = []
    return mgr


class TestNormalizePath:
    """Test _normalize_path."""

    def test_returns_string(self, tmp_path):
        """Returns a string, not a Path object."""
        mgr = _make_manager()
        result = mgr._normalize_path(tmp_path / "file.txt")
        assert isinstance(result, str)

    def test_resolves_to_absolute(self, tmp_path):
        """Result is an absolute path."""
        mgr = _make_manager()
        result = mgr._normalize_path(tmp_path / "file.txt")
        assert Path(result).is_absolute()

    def test_resolves_symlink(self, tmp_path):
        """Symlink and target resolve to the same normalized path."""
        mgr = _make_manager()
        real_file = tmp_path / "real.txt"
        real_file.write_text("content")
        symlink = tmp_path / "link.txt"
        symlink.symlink_to(real_file)

        assert mgr._normalize_path(symlink) == mgr._normalize_path(real_file)


class TestGetVerifierForFile:
    """Test _get_verifier_for_file."""

    def test_returns_none_with_no_verifiers(self, tmp_path):
        """No registered verifiers returns None."""
        mgr = _make_manager()
        assert mgr._get_verifier_for_file(tmp_path / "test.txt") is None

    def test_returns_matching_verifier(self, tmp_path):
        """Returns the first verifier that matches."""
        mgr = _make_manager()
        v1 = MagicMock()
        v1.should_verify.return_value = False
        v2 = MagicMock()
        v2.should_verify.return_value = True
        mgr.verifiers = [v1, v2]
        assert mgr._get_verifier_for_file(tmp_path / "test.txt") is v2

    def test_first_match_wins(self, tmp_path):
        """When multiple verifiers match, the first one wins."""
        mgr = _make_manager()
        v1 = MagicMock()
        v1.should_verify.return_value = True
        v2 = MagicMock()
        v2.should_verify.return_value = True
        mgr.verifiers = [v1, v2]
        assert mgr._get_verifier_for_file(tmp_path / "test.txt") is v1


class TestNeedsVerification:
    """Test _needs_verification mtime logic."""

    def test_returns_true_for_missing_file(self, tmp_path):
        """Missing file needs verification."""
        mgr = _make_manager()
        record = MagicMock()
        record.last_verified_at = datetime.now(UTC)
        result = mgr._needs_verification(record, tmp_path / "nonexistent.txt")
        assert result is True

    def test_returns_true_when_never_verified(self, tmp_path):
        """File never verified (last_verified_at=None) needs verification."""
        mgr = _make_manager()
        f = tmp_path / "test.txt"
        f.write_text("content")
        record = MagicMock()
        record.last_verified_at = None
        assert mgr._needs_verification(record, f) is True

    def test_returns_true_when_file_mtime_is_none(self, tmp_path):
        """Stored mtime=None means verification needed."""
        mgr = _make_manager()
        f = tmp_path / "test.txt"
        f.write_text("content")
        record = MagicMock()
        record.last_verified_at = datetime.now(UTC)
        record.file_mtime = None
        assert mgr._needs_verification(record, f) is True

    def test_returns_false_when_mtime_unchanged(self, tmp_path):
        """Unchanged mtime means no verification needed."""
        mgr = _make_manager()
        f = tmp_path / "test.txt"
        f.write_text("content")
        record = MagicMock()
        record.last_verified_at = datetime.now(UTC)
        record.file_mtime = f.stat().st_mtime
        assert mgr._needs_verification(record, f) is False

    def test_returns_true_when_mtime_changed(self, tmp_path):
        """Changed mtime triggers verification."""
        mgr = _make_manager()
        f = tmp_path / "test.txt"
        f.write_text("content")
        record = MagicMock()
        record.last_verified_at = datetime.now(UTC)
        record.file_mtime = f.stat().st_mtime - 1.0
        assert mgr._needs_verification(record, f) is True

    def test_mtime_tolerance_below_threshold(self, tmp_path):
        """Difference of 0.0005 is within tolerance (no verification)."""
        mgr = _make_manager()
        f = tmp_path / "test.txt"
        f.write_text("content")
        actual_mtime = f.stat().st_mtime
        record = MagicMock()
        record.last_verified_at = datetime.now(UTC)
        record.file_mtime = actual_mtime - 0.0005
        assert mgr._needs_verification(record, f) is False

    def test_mtime_tolerance_above_threshold(self, tmp_path):
        """Difference of 0.0015 exceeds tolerance (verification needed)."""
        mgr = _make_manager()
        f = tmp_path / "test.txt"
        f.write_text("content")
        actual_mtime = f.stat().st_mtime
        record = MagicMock()
        record.last_verified_at = datetime.now(UTC)
        record.file_mtime = actual_mtime - 0.0015
        assert mgr._needs_verification(record, f) is True


class TestDoVerification:
    """Test _do_verification."""

    def test_file_missing_returns_false(self, tmp_path):
        """Missing file returns (False, 'file_missing')."""
        mgr = _make_manager()
        record = MagicMock()
        session = MagicMock()
        result, reason = mgr._do_verification(
            record, tmp_path / "missing.txt", session
        )
        assert result is False
        assert reason == "file_missing"

    def test_no_verifier_returns_false(self, tmp_path):
        """No matching verifier returns (False, 'no_verifier')."""
        mgr = _make_manager()
        f = tmp_path / "test.txt"
        f.write_text("content")
        record = MagicMock()
        session = MagicMock()
        result, reason = mgr._do_verification(record, f, session)
        assert result is False
        assert reason == "no_verifier"

    def test_checksum_match_returns_true(self, tmp_path):
        """Matching checksum returns (True, None)."""
        mgr = _make_manager()
        f = tmp_path / "test.txt"
        f.write_text("content")
        verifier = MagicMock()
        verifier.should_verify.return_value = True
        verifier.calculate_checksum.return_value = "abc123"
        mgr.verifiers = [verifier]
        record = MagicMock()
        record.checksum = "abc123"
        session = MagicMock()
        result, reason = mgr._do_verification(record, f, session)
        assert result is True
        assert reason is None

    def test_checksum_mismatch_returns_false(self, tmp_path):
        """Mismatching checksum returns (False, 'checksum_mismatch')."""
        mgr = _make_manager()
        f = tmp_path / "test.txt"
        f.write_text("content")
        verifier = MagicMock()
        verifier.should_verify.return_value = True
        verifier.calculate_checksum.return_value = "different"
        mgr.verifiers = [verifier]
        record = MagicMock()
        record.checksum = "abc123"
        session = MagicMock()
        result, reason = mgr._do_verification(record, f, session)
        assert result is False
        assert reason == "checksum_mismatch"

    def test_mtime_not_updated_on_mismatch(self, tmp_path):
        """On checksum mismatch, record.file_mtime is not changed."""
        mgr = _make_manager()
        f = tmp_path / "test.txt"
        f.write_text("content")
        verifier = MagicMock()
        verifier.should_verify.return_value = True
        verifier.calculate_checksum.return_value = "different"
        mgr.verifiers = [verifier]
        record = MagicMock()
        record.checksum = "abc123"
        sentinel = 12345.0
        record.file_mtime = sentinel
        session = MagicMock()
        mgr._do_verification(record, f, session)
        assert record.file_mtime == sentinel


class TestUpdateStats:
    """Test _update_stats."""

    def test_pass_increments_successes_resets_failures(self):
        """Passing verification increments successes and resets failures."""
        mgr = _make_manager()
        record = MagicMock()
        record.total_verifications = 0
        record.consecutive_successes = 0
        record.consecutive_failures = 3
        mgr._update_stats(record, True, MagicMock())
        assert record.total_verifications == 1
        assert record.consecutive_successes == 1
        assert record.consecutive_failures == 0

    def test_fail_increments_failures_resets_successes(self):
        """Failing verification increments failures and resets successes."""
        mgr = _make_manager()
        record = MagicMock()
        record.total_verifications = 0
        record.consecutive_successes = 5
        record.consecutive_failures = 0
        mgr._update_stats(record, False, MagicMock())
        assert record.total_verifications == 1
        assert record.consecutive_failures == 1
        assert record.consecutive_successes == 0

    def test_last_verified_at_set(self):
        """last_verified_at is set to a datetime on every call."""
        mgr = _make_manager()
        record = MagicMock()
        record.total_verifications = 0
        record.consecutive_successes = 0
        record.consecutive_failures = 0
        record.last_verified_at = None
        mgr._update_stats(record, True, MagicMock())
        assert record.last_verified_at is not None

    def test_last_verification_passed_reflects_outcome(self):
        """last_verification_passed matches the passed parameter."""
        mgr = _make_manager()
        record = MagicMock()
        record.total_verifications = 0
        record.consecutive_successes = 0
        record.consecutive_failures = 0
        mgr._update_stats(record, False, MagicMock())
        assert record.last_verification_passed is False


class TestRegisterVerifier:
    """Test register_verifier."""

    def test_adds_to_verifiers_list(self):
        """Verifier is added to the list."""
        mgr = _make_manager()
        v = MagicMock()
        v.get_file_type.return_value = "test"
        mgr.register_verifier(v)
        assert v in mgr.verifiers

    def test_preserves_insertion_order(self):
        """Verifiers appear in registration order."""
        mgr = _make_manager()
        v1 = MagicMock()
        v1.get_file_type.return_value = "type1"
        v2 = MagicMock()
        v2.get_file_type.return_value = "type2"
        v3 = MagicMock()
        v3.get_file_type.return_value = "type3"
        mgr.register_verifier(v1)
        mgr.register_verifier(v2)
        mgr.register_verifier(v3)
        assert mgr.verifiers == [v1, v2, v3]


class TestClassConstants:
    """Test FileIntegrityManager class constants."""

    def test_max_failures_per_file(self):
        assert FileIntegrityManager.MAX_FAILURES_PER_FILE == 100

    def test_max_total_failures(self):
        assert FileIntegrityManager.MAX_TOTAL_FAILURES == 10000

    def test_per_file_less_than_total(self):
        """Per-file limit must be less than total limit."""
        assert (
            FileIntegrityManager.MAX_FAILURES_PER_FILE
            < FileIntegrityManager.MAX_TOTAL_FAILURES
        )