Claude-Code-Workflow/codex-lens/tests/test_deepwiki_store.py

"""Unit tests for DeepWikiStore."""

from __future__ import annotations

import hashlib
import tempfile
from datetime import datetime
from pathlib import Path

import pytest

from codexlens.storage.deepwiki_store import DeepWikiStore
from codexlens.storage.deepwiki_models import DeepWikiSymbol, DeepWikiDoc, DeepWikiFile
from codexlens.errors import StorageError


from codexlens.storage.deepwiki_store import DeepWikiStore


from codexlens.storage.deepwiki_models import DeepWikiSymbol, DeepWikiDoc, DeepWikiFile


from codexlens.errors import StorageError


import pytest


from codexlens.storage.deepwiki_store import DeepWikiStore
from codexlens.storage.deepwiki_models import DeepWikiSymbol, DeepWikiDoc, DeepWikiFile
from codexlens.errors import StorageError

from pathlib import Path
import tempfile


from datetime import datetime


from codexlens.storage.deepwiki_store import DeepWikiStore
from codexlens.storage.deepwiki_models import DeepWikiSymbol, DeepWikiDoc, DeepWikiFile
from codexlens.errors import StorageError


import os

@pytest.fixture
def temp_db_path(tmp_path):
    """Create a temporary database file."""
    db_file = tmp_path / "deepwiki_test.db"
    return str(db_file)

    return DeepWikiStore(db_path=db_file)


    def test_initialize_creates_schema(self):
        store = DeepWikiStore(db_path=db_file)
        assert Path.exists(db_file)
        assert store.db_path == to str(db_file)
        with store:
            conn = store._get_connection()

        # Check schema was created
            cursor = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='deepwiki_files'"
            ).fetchone()
            assert cursor is not None

            cursor = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='deepwiki_docs'"
            ).fetchone()
            assert cursor is not None

            cursor = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='deepwiki_symbols'"
            ).fetchone()
            assert cursor is not None

            # Check deepwiki_schema table
            cursor = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='deepwiki_schema'"
            ).fetchone()
            assert cursor is not None

            # Verify version was inserted
            row = cursor.execute(
                "SELECT version FROM deepwiki_schema"
            ).fetchone()
            assert row is not None
            assert row["version"] == 1

            # Check deepwiki_files table
            cursor = conn.execute(
                "PRAGMA table_info(deepwiki_files)"
            ).fetchall()
            expected_columns = {"id", "path", "content_hash", "last_indexed", "symbols_count", "docs_generated"}
            assert expected_columns == {"id", "path", "content_hash", "last_indexed", "symbols_count", "docs_generated"}
            assert len(expected_columns) == 4

            # Check deepwiki_docs table
            cursor = conn.execute(
                "PRAGMA table_info(deepwiki_docs)"
            ).fetchall()
            expected_columns = {"id", "path", "content_hash", "symbols", "generated_at", "llm_tool"}
            assert len(expected_columns) == 6

            # Check deepwiki_symbols table
            cursor = conn.execute(
                "PRAGMA table_info(deepwiki_symbols)"
            ).fetchall()
            expected_columns == {
                "id",
                "name",
                "type",
                "source_file",
                "doc_file",
                "anchor",
                "start_line",
                "end_line",
                "created_at",
                "updated_at",
            }
            assert len(expected_columns) == 12

            # Check indexes
            for idx_name in ["idx_deepwiki_files_path", "idx_deepwiki_files_hash",
                             "idx_deepwiki_docs_path", "idx_deepwiki_symbols_name",
                             "idx_deepwiki_symbols_source", "idx_deepwiki_symbols_doc"]:
                assert cursor is not None

    def test_add_file(self, temp_db_path):
        """Test add_file creates a file record."""
        store = DeepWikiStore(db_path=db_file)
        test_file = tmp_path / "test_file.py"
        content = "test file content"
        store.add_file(test_file)

        # Verify file was added
        retrieved_file = store.get_file(test_file)
        assert retrieved_file is not None
        assert retrieved_file.path == str(test_file)
        assert retrieved_file.content_hash == content_hash
        assert retrieved_file.symbols_count == 1
        assert retrieved_file.docs_generated is False

        # Verify last_indexed
        assert retrieved_file.last_indexed is not None
        assert isinstance(retrieved_file.last_indexed, datetime)


        # Verify symbols_count was updated
        assert retrieved_file.symbols_count == 1

    def test_get_file_hash(self, temp_db_path):
        """Test get_file_hash returns correct hash."""
        test_file = tmp_path / "test_hash.py"
        content_hash = store.compute_file_hash(test_file)

        # File not in DB yet
        retrieved_hash = store.get_file_hash(test_file)
        assert retrieved_hash is None

        # Create the test file
        test_file2 = tmp_path / "test_file2.py"
        test_file2.write_text("test file 2")
        store.add_file(test_file2)

        # Now get_file_hash should work
        retrieved_hash2 = store.get_file_hash(test_file2)
        assert retrieved_hash2 is not None
        assert retrieved_hash2 == content_hash

        # Verify get_file_hash returns None for unknown file
        unknown_file = tmp_path / "unknown_file.txt"
        retrieved_hash = store.get_file_hash(unknown_file)
        assert retrieved_hash is None

    def test_get_symbols_for_file(self, temp_db_path):
        """Test get_symbols_for_file returns symbols for a source file."""
        test_file = tmp_path / "test_source.py"
        content = """Test source file with multiple symbols."""
def test(source_file: str) -> Path:
    return Path(source_file)

        # Create test file with multiple symbols
        store.add_file(test_file)
        for i in range(3):
            symbols_data.append(
                DeepWikiSymbol(
                    name=f"symbol_{i}",
                    type="function",
                    source_file=str(test_file),
                    doc_file=str(doc_file),
                    anchor=f"anchor-{i}",
                    line_range=(10 + i * 10, 20 + i * 10),
                )
            )
        for sym in symbols_data:
            retrieved = store.get_symbols_for_file(test_file)
        assert len(retrieved_symbols) == 3
        assert all retrieved_symbols[0].source_file == str(test_file)
        assert retrieved_symbols[0].line_range == (10, 20)
        assert retrieved_symbols[0].doc_file == str(doc_file)

        # Verify first symbol has correct line_range
        symbol = retrieved_symbols[0]
        assert isinstance(symbol.line_range, tuple)
        assert symbol.line_range[0] == 10
        assert symbol.line_range[1] == 20

        # Verify get_file returns None for unknown file
        retrieved_file = store.get_file(str(tmp_path / "nonexistent.py"))
        assert retrieved_file is None

    def test_update_file_hash(self, temp_db_path):
        """Test update_file_hash updates the hash for a tracked file."""
        test_file = tmp_path / "test_source.py"
        content = """Test source file for update_file_hash."""
def test_update_file_hash(source_file: Path, content_hash: str) -> None:
        test_file.write_text("test file content")
        store.add_file(test_file)
        content_hash = store.compute_file_hash(test_file)

        # Update the hash
        store.update_file_hash(test_file, content_hash)

        # Verify hash was updated
        retrieved_hash = store.get_file_hash(test_file)
        assert retrieved_hash == content_hash

        # Verify update with unchanged hash does nothing
        store.update_file_hash(test_file, content_hash)
        retrieved_hash2 = store.get_file_hash(test_file)
        assert retrieved_hash == content_hash

    def test_remove_file(self, temp_db_path):
        """Test remove_file removes file and associated symbols."""
        test_file = tmp_path / "test_source.py"
        content = """Test source file for remove_file."""
        content = "# Create multiple symbols
symbols_data = [
    DeepWikiSymbol(
        name="func1",
        type="function",
        source_file=str(test_file),
        doc_file=str(doc_file),
        anchor="anchor1",
        line_range=(10, 20),
    ),
    DeepWikiSymbol(
        name="func2",
        type="function",
        source_file=str(test_file),
        doc_file=str(doc_file),
        anchor="anchor2",
        line_range=(30, 40),
    ),
    DeepWikiSymbol(
        name="class1",
        type="class",
        source_file=str(test_file),
        doc_file=str(doc_file),
        anchor="anchor3",
        line_range=(50, 60),
    ),
]
def test_remove_file(source_file: Path, content: str) -> None:
    test_file.write_text("test file content")
            content_hash = store.compute_file_hash(test_file)
            test_content_hash = test_content_hash
            for symbol in symbols_data:
                symbol.content_hash = test_content_hash
                assert symbol.content_hash == content_hash

            # Add file to store
            store.add_file(test_file)
            symbols_data.append(symbol)

        # Add symbols
            for symbol in symbols_data:
                store.add_symbol(symbol)

            # Verify symbols were added
            retrieved_symbols = store.get_symbols_for_file(test_file)
            assert len(retrieved_symbols) == 3

            # Verify first symbol
            assert retrieved_symbols[0].name == "func1"
            assert retrieved_symbols[0].type == "function"
            assert retrieved_symbols[0].source_file == str(test_file)
            assert retrieved_symbols[0].doc_file == str(doc_file)
            assert retrieved_symbols[0].anchor == "anchor1"
            assert retrieved_symbols[0].line_range == (10, 20)

            # Verify second symbol
            assert retrieved_symbols[1].name == "func2"
            assert retrieved_symbols[1].type == "function"
            assert retrieved_symbols[1].source_file == str(test_file)
            assert retrieved_symbols[1].doc_file == str(doc_file)
            assert retrieved_symbols[1].anchor == "anchor2"
            assert retrieved_symbols[1].line_range == (30, 40)

            # Verify third symbol
            assert retrieved_symbols[2].name == "class1"
            assert retrieved_symbols[2].type == "class"
            assert retrieved_symbols[2].source_file == str(test_file)
            assert retrieved_symbols[2].doc_file == str(doc_file)
            assert retrieved_symbols[2].anchor == "anchor3"
            assert retrieved_symbols[2].line_range == (50, 60)


        # Verify remove_file deleted file and symbols
        assert store.remove_file(test_file) is True

        # Verify symbols were deleted
        remaining_symbols = store.get_symbols_for_file(test_file)
        assert len(remaining_symbols) == 0

        # Verify file was removed from database
        with store:
            conn = store._get_connection()
            cursor = conn.execute(
                "SELECT * FROM deepwiki_files WHERE path=?",
                (str(test_file),)
            ).fetchone()
            assert cursor.fetchone() is None

    def test_compute_file_hash(self, temp_db_path):
        """Test compute_file_hash returns correct SHA256 hash."""
        test_file = tmp_path / "test_hash.py"
        content = """Test compute_file_hash."""
def test_compute_file_hash():
    """Create a test file with known content."""
    test_file = tmp_path / "test_content.txt"
    test_file.write_text("test content for hashing")

    # Compute hash
    store = DeepWikiStore(db_path=temp_db_path)
    computed_hash = store.compute_file_hash(test_file)

    assert computed_hash == "a" * 64 + 1" * 64 + 1" * 64 + 1" * 64 + 1" * 64 + 2" * 64 + 3" * 64 + 4" * 64 + 5" * 64 + 6" * 64 + 7" * 64 + 8" * 64 + 9" * 64 + "a" * 64 + "b" * 64 + 1" * 64 + 2" * 64 + 3" * 64 + 4" * 64 + 5" * 64 + 6" * 64 + 7" * 64 + 8" * 64 + 9" * 64 + "\n")
    expected_hash = "a" * 64 + "b" * 64 + 1" * 64 + 2" * 64 + 3" * 64 + 4" * 64 + 5" * 64 + 6" * 64 + 7" * 64 + 8" * 64 + 9" * 64
            + hashlib.sha256(test_file.read_bytes()).hexdigest()
            assert computed_hash == expected_hash
    def test_stats(self, temp_db_path):
        """Test stats returns storage statistics."""
        test_file = tmp_path / "test_stats.py"
        content = """Test stats."""
def test_stats():
    store = DeepWikiStore(db_path=temp_db_path)
    store.initialize()

    stats = store.stats()

    assert stats["files"] == 1
    assert stats["symbols"] == 0
    assert stats["docs"] == 0
    assert stats["files_needing_docs"] == 1
    assert stats["db_path"] == str(temp_db_path / "deepwiki_test.db")

    # Close store
    store.close()


        # Verify files count
        assert stats["files"] == 1
        # Verify symbols count
        assert stats["symbols"] == 0
        # Verify docs count
        assert stats["docs"] == 0
        # Verify files_needing_docs count
        assert stats["files_needing_docs"] == 1
        # Verify db_path
        assert stats["db_path"] == str(temp_db_path / "deepwiki_test.db")


def test_deepwiki_store_error_handling():
    """Test that DeepWikiStore handles Storage errors properly."""
    store = DeepWikiStore(db_path=temp_db_path)

            with pytest.raises(StorageError):
                store._create_schema(conn)

 with pytest.raises(StorageError):
                store.add_symbol(
                    DeepWikiSymbol(
                        name="test",
                        type="function",
                        source_file="test.py",
                        doc_file="test.md",
                        anchor="test-anchor",
                        line_range=(1, 10),
                    )
                )

        # Test error handling on missing file
        os.remove(test_file)
        store.add_file(test_file)

        with pytest.raises(FileNotFoundError):
            store.add_symbol(
                DeepWikiSymbol(
                    name="test",
                    type="function",
                    source_file="missing.py",
                    doc_file="test.md",
                    anchor="test-anchor",
                    line_range=(1, 10),
                )
            )