test_lint_cache.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_linter/test_lint_cache.py
 
"""
Purpose
-------
FS-510 regression coverage for the incremental lint cache.
 
Responsibilities
----------------
- Validate content_hash and rules_hash are deterministic.
- Validate cache miss on first run and hit on second.
- Validate invalidation on content, rule, or version change.
- Validate corrupt/missing file handling.
- Validate --no-cache bypasses reads.
- Validate thread safety.
 
Diagnostics
-----------
Domain: LINTER-TESTS
"""
 
from __future__ import annotations
 
import json
import threading
from pathlib import Path
 
import pytest
 
from oct.linter.lint_cache import (
    LintCache,
    LintCacheEntry,
    content_hash,
    rules_hash,
)
 
 
# -------------------------------------------------------------------
# Hash determinism
# -------------------------------------------------------------------
 
class TestHashing:
 
    def test_content_hash_deterministic(self):
        assert content_hash("hello") == content_hash("hello")
 
    def test_content_hash_changes_on_different_input(self):
        assert content_hash("hello") != content_hash("world")
 
    def test_rules_hash_deterministic(self):
        rules = {"header": True, "docstring": False}
        assert rules_hash(rules) == rules_hash(rules)
 
    def test_rules_hash_order_independent(self):
        r1 = {"a": True, "b": False}
        r2 = {"b": False, "a": True}
        assert rules_hash(r1) == rules_hash(r2)
 
    def test_rules_hash_changes_on_different_rules(self):
        r1 = {"header": True}
        r2 = {"header": False}
        assert rules_hash(r1) != rules_hash(r2)
 
 
# -------------------------------------------------------------------
# Cache operations
# -------------------------------------------------------------------
 
def _make_entry(file_hash: str = "abc123") -> LintCacheEntry:
    return LintCacheEntry(
        content_hash=file_hash,
        rule_results={"header": True, "docstring": False},
        json_detail={"path": "test.py", "checks": {}},
        oct_version="0.22.0",
        timestamp="2026-04-17T00:00:00",
    )
 
 
class TestCacheOperations:
 
    def test_miss_on_empty_cache(self, tmp_path: Path):
        cache = LintCache(tmp_path / "cache.json", "0.22.0")
        assert cache.get("abc", "def") is None
 
    def test_put_and_get(self, tmp_path: Path):
        cache = LintCache(tmp_path / "cache.json", "0.22.0")
        entry = _make_entry("abc123")
        cache.put("abc123", "rules1", entry)
        result = cache.get("abc123", "rules1")
        assert result is not None
        assert result.content_hash == "abc123"
        assert result.rule_results == {"header": True, "docstring": False}
 
    def test_save_and_reload(self, tmp_path: Path):
        cache_path = tmp_path / "cache.json"
        cache = LintCache(cache_path, "0.22.0")
        cache.put("abc123", "rules1", _make_entry("abc123"))
        cache.save()
 
        cache2 = LintCache(cache_path, "0.22.0")
        result = cache2.get("abc123", "rules1")
        assert result is not None
        assert result.content_hash == "abc123"
 
    def test_invalidated_on_version_change(self, tmp_path: Path):
        cache_path = tmp_path / "cache.json"
        cache = LintCache(cache_path, "0.22.0")
        cache.put("abc123", "rules1", _make_entry("abc123"))
        cache.save()
 
        cache2 = LintCache(cache_path, "0.23.0")
        assert cache2.get("abc123", "rules1") is None
 
    def test_invalidated_on_content_change(self, tmp_path: Path):
        cache = LintCache(tmp_path / "cache.json", "0.22.0")
        cache.put("abc123", "rules1", _make_entry("abc123"))
        assert cache.get("different_hash", "rules1") is None
 
    def test_invalidated_on_rules_change(self, tmp_path: Path):
        cache = LintCache(tmp_path / "cache.json", "0.22.0")
        cache.put("abc123", "rules1", _make_entry("abc123"))
        assert cache.get("abc123", "different_rules") is None
 
    def test_corrupt_file_starts_fresh(self, tmp_path: Path):
        cache_path = tmp_path / "cache.json"
        cache_path.write_text("not json!!!", encoding="utf-8")
        cache = LintCache(cache_path, "0.22.0")
        assert cache.entry_count == 0
 
    def test_missing_file_starts_fresh(self, tmp_path: Path):
        cache = LintCache(tmp_path / "nonexistent.json", "0.22.0")
        assert cache.entry_count == 0
 
    def test_save_creates_parent_dirs(self, tmp_path: Path):
        cache_path = tmp_path / "deep" / "nested" / "cache.json"
        cache = LintCache(cache_path, "0.22.0")
        cache.put("abc", "rules", _make_entry("abc"))
        cache.save()
        assert cache_path.is_file()
 
    def test_entry_count(self, tmp_path: Path):
        cache = LintCache(tmp_path / "cache.json", "0.22.0")
        assert cache.entry_count == 0
        cache.put("a", "r1", _make_entry("a"))
        assert cache.entry_count == 1
        cache.put("b", "r1", _make_entry("b"))
        assert cache.entry_count == 2
 
    def test_thread_safety(self, tmp_path: Path):
        cache = LintCache(tmp_path / "cache.json", "0.22.0")
        errors: list[str] = []
 
        def worker(i: int) -> None:
            try:
                h = f"hash_{i}"
                cache.put(h, "rules", _make_entry(h))
                result = cache.get(h, "rules")
                if result is None:
                    errors.append(f"Worker {i}: put then get returned None")
            except Exception as e:
                errors.append(f"Worker {i}: {e}")
 
        threads = [threading.Thread(target=worker, args=(i,)) for i in range(20)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
 
        assert not errors, f"Thread safety errors: {errors}"
        assert cache.entry_count == 20