feat: Add CLAUDE.md freshness tracking and update reminders

- Add SQLite table and CRUD methods for tracking update history
- Create freshness calculation service based on git file changes
- Add API endpoints for freshness data, marking updates, and history
- Display freshness badges in file tree (green/yellow/red indicators)
- Show freshness gauge and details in metadata panel
- Auto-mark files as updated after CLI sync
- Add English and Chinese i18n translations

Freshness algorithm: 100 - min((changedFilesCount / 20) * 100, 100)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
catlog22
2025-12-20 16:14:46 +08:00
parent 4a3ff82200
commit b27d8a9570
18 changed files with 2260 additions and 18 deletions

View File

@@ -268,6 +268,7 @@ def search(
files_only: bool = typer.Option(False, "--files-only", "-f", help="Return only file paths without content snippets."),
mode: str = typer.Option("auto", "--mode", "-m", help="Search mode: auto, exact, fuzzy, hybrid, vector, pure-vector."),
weights: Optional[str] = typer.Option(None, "--weights", help="Custom RRF weights as 'exact,fuzzy,vector' (e.g., '0.5,0.3,0.2')."),
enrich: bool = typer.Option(False, "--enrich", help="Enrich results with code graph relationships (calls, imports)."),
json_mode: bool = typer.Option(False, "--json", help="Output JSON response."),
verbose: bool = typer.Option(False, "--verbose", "-v", help="Enable debug logging."),
) -> None:
@@ -411,19 +412,42 @@ def search(
console.print(fp)
else:
result = engine.search(query, search_path, options)
results_list = [
{
"path": r.path,
"score": r.score,
"excerpt": r.excerpt,
"source": getattr(r, "search_source", None),
"symbol": getattr(r, "symbol", None),
}
for r in result.results
]
# Enrich results with relationship data if requested
enriched = False
if enrich:
try:
from codexlens.search.enrichment import RelationshipEnricher
# Find index path for the search path
project_record = registry.find_by_source_path(str(search_path))
if project_record:
index_path = Path(project_record["index_root"]) / "_index.db"
if index_path.exists():
with RelationshipEnricher(index_path) as enricher:
results_list = enricher.enrich(results_list, limit=limit)
enriched = True
except Exception as e:
# Enrichment failure should not break search
if verbose:
console.print(f"[yellow]Warning: Enrichment failed: {e}[/yellow]")
payload = {
"query": query,
"mode": actual_mode,
"count": len(result.results),
"results": [
{
"path": r.path,
"score": r.score,
"excerpt": r.excerpt,
"source": getattr(r, "search_source", None),
}
for r in result.results
],
"count": len(results_list),
"enriched": enriched,
"results": results_list,
"stats": {
"dirs_searched": result.stats.dirs_searched,
"files_matched": result.stats.files_matched,
@@ -434,7 +458,8 @@ def search(
print_json(success=True, result=payload)
else:
render_search_results(result.results, verbose=verbose)
console.print(f"[dim]Mode: {actual_mode} | Searched {result.stats.dirs_searched} directories in {result.stats.time_ms:.1f}ms[/dim]")
enrich_status = " | [green]Enriched[/green]" if enriched else ""
console.print(f"[dim]Mode: {actual_mode} | Searched {result.stats.dirs_searched} directories in {result.stats.time_ms:.1f}ms{enrich_status}[/dim]")
except SearchError as exc:
if json_mode:

View File

@@ -0,0 +1,77 @@
# Symbol Extraction and Indexing
This module provides symbol extraction and relationship tracking for code graph enrichment.
## Overview
The `SymbolExtractor` class extracts code symbols (functions, classes) and their relationships (calls, imports) from source files using regex-based pattern matching.
## Supported Languages
- Python (.py)
- TypeScript (.ts, .tsx)
- JavaScript (.js, .jsx)
## Database Schema
### Symbols Table
Stores code symbols with their location information:
- `id`: Primary key
- `qualified_name`: Fully qualified name (e.g., "module.ClassName")
- `name`: Symbol name
- `kind`: Symbol type (function, class)
- `file_path`: Path to source file
- `start_line`: Starting line number
- `end_line`: Ending line number
### Symbol Relationships Table
Stores relationships between symbols:
- `id`: Primary key
- `source_symbol_id`: Foreign key to symbols table
- `target_symbol_fqn`: Fully qualified name of target symbol
- `relationship_type`: Type of relationship (calls, imports)
- `file_path`: Path to source file
- `line`: Line number where relationship occurs
## Usage Example
```python
from pathlib import Path
from codexlens.indexing.symbol_extractor import SymbolExtractor
# Initialize extractor
db_path = Path("./code_index.db")
extractor = SymbolExtractor(db_path)
extractor.connect()
# Extract from file
file_path = Path("src/my_module.py")
with open(file_path) as f:
content = f.read()
symbols, relationships = extractor.extract_from_file(file_path, content)
# Save to database
name_to_id = extractor.save_symbols(symbols)
extractor.save_relationships(relationships, name_to_id)
# Clean up
extractor.close()
```
## Pattern Matching
The extractor uses regex patterns to identify:
- **Functions**: Function definitions (including async, export keywords)
- **Classes**: Class definitions (including export keyword)
- **Imports**: Import/require statements
- **Calls**: Function/method invocations
## Future Enhancements
- Tree-sitter integration for more accurate parsing
- Support for additional languages
- Method and variable extraction
- Enhanced scope tracking
- Relationship type expansion (inherits, implements, etc.)

View File

@@ -0,0 +1,4 @@
"""Code indexing and symbol extraction."""
from codexlens.indexing.symbol_extractor import SymbolExtractor
__all__ = ["SymbolExtractor"]

View File

@@ -0,0 +1,234 @@
"""Symbol and relationship extraction from source code."""
import re
import sqlite3
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
class SymbolExtractor:
"""Extract symbols and relationships from source code using regex patterns."""
# Pattern definitions for different languages
PATTERNS = {
'python': {
'function': r'^(?:async\s+)?def\s+(\w+)\s*\(',
'class': r'^class\s+(\w+)\s*[:\(]',
'import': r'^(?:from\s+([\w.]+)\s+)?import\s+([\w.,\s]+)',
'call': r'(?<![.\w])(\w+)\s*\(',
},
'typescript': {
'function': r'(?:export\s+)?(?:async\s+)?function\s+(\w+)\s*[<\(]',
'class': r'(?:export\s+)?class\s+(\w+)',
'import': r"import\s+.*\s+from\s+['\"]([^'\"]+)['\"]",
'call': r'(?<![.\w])(\w+)\s*[<\(]',
},
'javascript': {
'function': r'(?:export\s+)?(?:async\s+)?function\s+(\w+)\s*\(',
'class': r'(?:export\s+)?class\s+(\w+)',
'import': r"(?:import|require)\s*\(?['\"]([^'\"]+)['\"]",
'call': r'(?<![.\w])(\w+)\s*\(',
}
}
LANGUAGE_MAP = {
'.py': 'python',
'.ts': 'typescript',
'.tsx': 'typescript',
'.js': 'javascript',
'.jsx': 'javascript',
}
def __init__(self, db_path: Path):
self.db_path = db_path
self.db_conn: Optional[sqlite3.Connection] = None
def connect(self) -> None:
"""Connect to database and ensure schema exists."""
self.db_conn = sqlite3.connect(str(self.db_path))
self._ensure_tables()
def _ensure_tables(self) -> None:
"""Create symbols and relationships tables if they don't exist."""
if not self.db_conn:
return
cursor = self.db_conn.cursor()
# Create symbols table with qualified_name
cursor.execute('''
CREATE TABLE IF NOT EXISTS symbols (
id INTEGER PRIMARY KEY AUTOINCREMENT,
qualified_name TEXT NOT NULL,
name TEXT NOT NULL,
kind TEXT NOT NULL,
file_path TEXT NOT NULL,
start_line INTEGER NOT NULL,
end_line INTEGER NOT NULL,
UNIQUE(file_path, name, start_line)
)
''')
# Create relationships table with target_symbol_fqn
cursor.execute('''
CREATE TABLE IF NOT EXISTS symbol_relationships (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_symbol_id INTEGER NOT NULL,
target_symbol_fqn TEXT NOT NULL,
relationship_type TEXT NOT NULL,
file_path TEXT NOT NULL,
line INTEGER,
FOREIGN KEY (source_symbol_id) REFERENCES symbols(id) ON DELETE CASCADE
)
''')
# Create performance indexes
cursor.execute('CREATE INDEX IF NOT EXISTS idx_symbols_name ON symbols(name)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_symbols_file ON symbols(file_path)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_rel_source ON symbol_relationships(source_symbol_id)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_rel_target ON symbol_relationships(target_symbol_fqn)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_rel_type ON symbol_relationships(relationship_type)')
self.db_conn.commit()
def extract_from_file(self, file_path: Path, content: str) -> Tuple[List[Dict], List[Dict]]:
"""Extract symbols and relationships from file content.
Args:
file_path: Path to the source file
content: File content as string
Returns:
Tuple of (symbols, relationships) where:
- symbols: List of symbol dicts with qualified_name, name, kind, file_path, start_line, end_line
- relationships: List of relationship dicts with source_scope, target, type, file_path, line
"""
ext = file_path.suffix.lower()
lang = self.LANGUAGE_MAP.get(ext)
if not lang or lang not in self.PATTERNS:
return [], []
patterns = self.PATTERNS[lang]
symbols = []
relationships = []
lines = content.split('\n')
current_scope = None
for line_num, line in enumerate(lines, 1):
# Extract function/class definitions
for kind in ['function', 'class']:
if kind in patterns:
match = re.search(patterns[kind], line)
if match:
name = match.group(1)
qualified_name = f"{file_path.stem}.{name}"
symbols.append({
'qualified_name': qualified_name,
'name': name,
'kind': kind,
'file_path': str(file_path),
'start_line': line_num,
'end_line': line_num, # Simplified - would need proper parsing for actual end
})
current_scope = name
# Extract imports
if 'import' in patterns:
match = re.search(patterns['import'], line)
if match:
import_target = match.group(1) or match.group(2) if match.lastindex >= 2 else match.group(1)
if import_target and current_scope:
relationships.append({
'source_scope': current_scope,
'target': import_target.strip(),
'type': 'imports',
'file_path': str(file_path),
'line': line_num,
})
# Extract function calls (simplified)
if 'call' in patterns and current_scope:
for match in re.finditer(patterns['call'], line):
call_name = match.group(1)
# Skip common keywords and the current function
if call_name not in ['if', 'for', 'while', 'return', 'print', 'len', 'str', 'int', 'float', 'list', 'dict', 'set', 'tuple', current_scope]:
relationships.append({
'source_scope': current_scope,
'target': call_name,
'type': 'calls',
'file_path': str(file_path),
'line': line_num,
})
return symbols, relationships
def save_symbols(self, symbols: List[Dict]) -> Dict[str, int]:
"""Save symbols to database and return name->id mapping.
Args:
symbols: List of symbol dicts with qualified_name, name, kind, file_path, start_line, end_line
Returns:
Dictionary mapping symbol name to database id
"""
if not self.db_conn or not symbols:
return {}
cursor = self.db_conn.cursor()
name_to_id = {}
for sym in symbols:
try:
cursor.execute('''
INSERT OR IGNORE INTO symbols
(qualified_name, name, kind, file_path, start_line, end_line)
VALUES (?, ?, ?, ?, ?, ?)
''', (sym['qualified_name'], sym['name'], sym['kind'],
sym['file_path'], sym['start_line'], sym['end_line']))
# Get the id
cursor.execute('''
SELECT id FROM symbols
WHERE file_path = ? AND name = ? AND start_line = ?
''', (sym['file_path'], sym['name'], sym['start_line']))
row = cursor.fetchone()
if row:
name_to_id[sym['name']] = row[0]
except sqlite3.Error:
continue
self.db_conn.commit()
return name_to_id
def save_relationships(self, relationships: List[Dict], name_to_id: Dict[str, int]) -> None:
"""Save relationships to database.
Args:
relationships: List of relationship dicts with source_scope, target, type, file_path, line
name_to_id: Dictionary mapping symbol names to database ids
"""
if not self.db_conn or not relationships:
return
cursor = self.db_conn.cursor()
for rel in relationships:
source_id = name_to_id.get(rel['source_scope'])
if source_id:
try:
cursor.execute('''
INSERT INTO symbol_relationships
(source_symbol_id, target_symbol_fqn, relationship_type, file_path, line)
VALUES (?, ?, ?, ?, ?)
''', (source_id, rel['target'], rel['type'], rel['file_path'], rel['line']))
except sqlite3.Error:
continue
self.db_conn.commit()
def close(self) -> None:
"""Close database connection."""
if self.db_conn:
self.db_conn.close()
self.db_conn = None

View File

@@ -0,0 +1,150 @@
# codex-lens/src/codexlens/search/enrichment.py
"""Relationship enrichment for search results."""
import sqlite3
from pathlib import Path
from typing import List, Dict, Any, Optional
class RelationshipEnricher:
"""Enriches search results with code graph relationships."""
def __init__(self, index_path: Path):
"""Initialize with path to index database.
Args:
index_path: Path to _index.db SQLite database
"""
self.index_path = index_path
self.db_conn: Optional[sqlite3.Connection] = None
self._connect()
def _connect(self) -> None:
"""Establish read-only database connection."""
if self.index_path.exists():
self.db_conn = sqlite3.connect(
f"file:{self.index_path}?mode=ro",
uri=True,
check_same_thread=False
)
self.db_conn.row_factory = sqlite3.Row
def enrich(self, results: List[Dict[str, Any]], limit: int = 10) -> List[Dict[str, Any]]:
"""Add relationship data to search results.
Args:
results: List of search result dictionaries
limit: Maximum number of results to enrich
Returns:
Results with relationships field added
"""
if not self.db_conn:
return results
for result in results[:limit]:
file_path = result.get('file') or result.get('path')
symbol_name = result.get('symbol')
result['relationships'] = self._find_relationships(file_path, symbol_name)
return results
def _find_relationships(self, file_path: Optional[str], symbol_name: Optional[str]) -> List[Dict[str, Any]]:
"""Query relationships for a symbol.
Args:
file_path: Path to file containing the symbol
symbol_name: Name of the symbol
Returns:
List of relationship dictionaries with type, direction, target/source, file, line
"""
if not self.db_conn or not symbol_name:
return []
relationships = []
cursor = self.db_conn.cursor()
try:
# Find symbol ID(s) by name and optionally file
if file_path:
cursor.execute(
'SELECT id FROM symbols WHERE name = ? AND file_path = ?',
(symbol_name, file_path)
)
else:
cursor.execute('SELECT id FROM symbols WHERE name = ?', (symbol_name,))
symbol_ids = [row[0] for row in cursor.fetchall()]
if not symbol_ids:
return []
# Query outgoing relationships (symbol is source)
placeholders = ','.join('?' * len(symbol_ids))
cursor.execute(f'''
SELECT sr.relationship_type, sr.target_symbol_fqn, sr.file_path, sr.line
FROM symbol_relationships sr
WHERE sr.source_symbol_id IN ({placeholders})
''', symbol_ids)
for row in cursor.fetchall():
relationships.append({
'type': row[0],
'direction': 'outgoing',
'target': row[1],
'file': row[2],
'line': row[3],
})
# Query incoming relationships (symbol is target)
# Match against symbol name or qualified name patterns
cursor.execute('''
SELECT sr.relationship_type, s.name AS source_name, sr.file_path, sr.line
FROM symbol_relationships sr
JOIN symbols s ON sr.source_symbol_id = s.id
WHERE sr.target_symbol_fqn = ? OR sr.target_symbol_fqn LIKE ?
''', (symbol_name, f'%.{symbol_name}'))
for row in cursor.fetchall():
rel_type = row[0]
# Convert to incoming type
incoming_type = self._to_incoming_type(rel_type)
relationships.append({
'type': incoming_type,
'direction': 'incoming',
'source': row[1],
'file': row[2],
'line': row[3],
})
except sqlite3.Error:
return []
return relationships
def _to_incoming_type(self, outgoing_type: str) -> str:
"""Convert outgoing relationship type to incoming type.
Args:
outgoing_type: The outgoing relationship type (e.g., 'calls', 'imports')
Returns:
Corresponding incoming type (e.g., 'called_by', 'imported_by')
"""
type_map = {
'calls': 'called_by',
'imports': 'imported_by',
'extends': 'extended_by',
}
return type_map.get(outgoing_type, f'{outgoing_type}_by')
def close(self) -> None:
"""Close database connection."""
if self.db_conn:
self.db_conn.close()
self.db_conn = None
def __enter__(self) -> 'RelationshipEnricher':
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
self.close()

View File

@@ -0,0 +1,122 @@
"""Tests for CLI search command with --enrich flag."""
import json
import pytest
from typer.testing import CliRunner
from codexlens.cli.commands import app
class TestCLISearchEnrich:
"""Test CLI search command with --enrich flag integration."""
@pytest.fixture
def runner(self):
"""Create CLI test runner."""
return CliRunner()
def test_search_with_enrich_flag_help(self, runner):
"""Test --enrich flag is documented in help."""
result = runner.invoke(app, ["search", "--help"])
assert result.exit_code == 0
assert "--enrich" in result.output
assert "relationships" in result.output.lower() or "graph" in result.output.lower()
def test_search_with_enrich_flag_accepted(self, runner):
"""Test --enrich flag is accepted by the CLI."""
result = runner.invoke(app, ["search", "test", "--enrich"])
# Should not show 'unknown option' error
assert "No such option" not in result.output
assert "error: unrecognized" not in result.output.lower()
def test_search_without_enrich_flag(self, runner):
"""Test search without --enrich flag has no relationships."""
result = runner.invoke(app, ["search", "test", "--json"])
# Even without an index, JSON should be attempted
if result.exit_code == 0:
try:
data = json.loads(result.output)
# If we get results, they should not have enriched=true
if data.get("success") and "result" in data:
assert data["result"].get("enriched", False) is False
except json.JSONDecodeError:
pass # Not JSON output, that's fine for error cases
def test_search_enrich_json_output_structure(self, runner):
"""Test JSON output structure includes enriched flag."""
result = runner.invoke(app, ["search", "test", "--json", "--enrich"])
# If we get valid JSON output, check structure
if result.exit_code == 0:
try:
data = json.loads(result.output)
if data.get("success") and "result" in data:
# enriched field should exist
assert "enriched" in data["result"]
except json.JSONDecodeError:
pass # Not JSON output
def test_search_enrich_with_mode(self, runner):
"""Test --enrich works with different search modes."""
modes = ["exact", "fuzzy", "hybrid"]
for mode in modes:
result = runner.invoke(
app, ["search", "test", "--mode", mode, "--enrich"]
)
# Should not show validation errors
assert "Invalid" not in result.output
class TestEnrichFlagBehavior:
"""Test behavioral aspects of --enrich flag."""
@pytest.fixture
def runner(self):
"""Create CLI test runner."""
return CliRunner()
def test_enrich_failure_does_not_break_search(self, runner):
"""Test that enrichment failure doesn't prevent search from returning results."""
# Even without proper index, search should not crash due to enrich
result = runner.invoke(app, ["search", "test", "--enrich", "--verbose"])
# Should not have unhandled exception
assert "Traceback" not in result.output
def test_enrich_flag_with_files_only(self, runner):
"""Test --enrich is accepted with --files-only mode."""
result = runner.invoke(app, ["search", "test", "--enrich", "--files-only"])
# Should not show option conflict error
assert "conflict" not in result.output.lower()
def test_enrich_flag_with_limit(self, runner):
"""Test --enrich works with --limit parameter."""
result = runner.invoke(app, ["search", "test", "--enrich", "--limit", "5"])
# Should not show validation error
assert "Invalid" not in result.output
class TestEnrichOutputFormat:
"""Test output format with --enrich flag."""
@pytest.fixture
def runner(self):
"""Create CLI test runner."""
return CliRunner()
def test_enrich_verbose_shows_status(self, runner):
"""Test verbose mode shows enrichment status."""
result = runner.invoke(app, ["search", "test", "--enrich", "--verbose"])
# Verbose mode may show enrichment info or warnings
# Just ensure it doesn't crash
assert result.exit_code in [0, 1] # 0 = success, 1 = no index
def test_json_output_has_enriched_field(self, runner):
"""Test JSON output always has enriched field when --enrich used."""
result = runner.invoke(app, ["search", "test", "--json", "--enrich"])
if result.exit_code == 0:
try:
data = json.loads(result.output)
if data.get("success"):
result_data = data.get("result", {})
assert "enriched" in result_data
assert isinstance(result_data["enriched"], bool)
except json.JSONDecodeError:
pass

View File

@@ -0,0 +1,234 @@
"""Tests for search result enrichment with relationship data."""
import sqlite3
import tempfile
import time
from pathlib import Path
import pytest
from codexlens.search.enrichment import RelationshipEnricher
@pytest.fixture
def mock_db():
"""Create a mock database with symbols and relationships."""
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "_index.db"
conn = sqlite3.connect(str(db_path))
cursor = conn.cursor()
# Create schema
cursor.execute('''
CREATE TABLE symbols (
id INTEGER PRIMARY KEY AUTOINCREMENT,
qualified_name TEXT NOT NULL,
name TEXT NOT NULL,
kind TEXT NOT NULL,
file_path TEXT NOT NULL,
start_line INTEGER NOT NULL,
end_line INTEGER NOT NULL
)
''')
cursor.execute('''
CREATE TABLE symbol_relationships (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_symbol_id INTEGER NOT NULL,
target_symbol_fqn TEXT NOT NULL,
relationship_type TEXT NOT NULL,
file_path TEXT NOT NULL,
line INTEGER,
FOREIGN KEY (source_symbol_id) REFERENCES symbols(id)
)
''')
# Insert test data
cursor.execute('''
INSERT INTO symbols (qualified_name, name, kind, file_path, start_line, end_line)
VALUES ('module.main', 'main', 'function', 'module.py', 1, 10)
''')
main_id = cursor.lastrowid
cursor.execute('''
INSERT INTO symbols (qualified_name, name, kind, file_path, start_line, end_line)
VALUES ('module.helper', 'helper', 'function', 'module.py', 12, 20)
''')
helper_id = cursor.lastrowid
cursor.execute('''
INSERT INTO symbols (qualified_name, name, kind, file_path, start_line, end_line)
VALUES ('utils.fetch', 'fetch', 'function', 'utils.py', 1, 5)
''')
fetch_id = cursor.lastrowid
# main calls helper
cursor.execute('''
INSERT INTO symbol_relationships (source_symbol_id, target_symbol_fqn, relationship_type, file_path, line)
VALUES (?, 'helper', 'calls', 'module.py', 5)
''', (main_id,))
# main calls fetch
cursor.execute('''
INSERT INTO symbol_relationships (source_symbol_id, target_symbol_fqn, relationship_type, file_path, line)
VALUES (?, 'utils.fetch', 'calls', 'module.py', 6)
''', (main_id,))
# helper imports os
cursor.execute('''
INSERT INTO symbol_relationships (source_symbol_id, target_symbol_fqn, relationship_type, file_path, line)
VALUES (?, 'os', 'imports', 'module.py', 13)
''', (helper_id,))
conn.commit()
conn.close()
yield db_path
class TestRelationshipEnricher:
"""Test suite for RelationshipEnricher."""
def test_enrich_with_relationships(self, mock_db):
"""Test enriching results with valid relationships."""
with RelationshipEnricher(mock_db) as enricher:
results = [
{"path": "module.py", "score": 0.9, "excerpt": "def main():", "symbol": "main"},
{"path": "module.py", "score": 0.8, "excerpt": "def helper():", "symbol": "helper"},
]
enriched = enricher.enrich(results, limit=10)
# Check main's relationships
main_result = enriched[0]
assert "relationships" in main_result
main_rels = main_result["relationships"]
assert len(main_rels) >= 2
# Verify outgoing relationships
outgoing = [r for r in main_rels if r["direction"] == "outgoing"]
targets = [r["target"] for r in outgoing]
assert "helper" in targets or any("helper" in t for t in targets)
# Check helper's relationships
helper_result = enriched[1]
assert "relationships" in helper_result
helper_rels = helper_result["relationships"]
assert len(helper_rels) >= 1
# Verify incoming relationships (main calls helper)
incoming = [r for r in helper_rels if r["direction"] == "incoming"]
assert len(incoming) >= 1
assert incoming[0]["type"] == "called_by"
def test_enrich_missing_symbol(self, mock_db):
"""Test graceful handling of missing symbols."""
with RelationshipEnricher(mock_db) as enricher:
results = [
{"path": "unknown.py", "score": 0.9, "excerpt": "code", "symbol": "nonexistent"},
]
enriched = enricher.enrich(results, limit=10)
# Should return empty relationships, not crash
assert "relationships" in enriched[0]
assert enriched[0]["relationships"] == []
def test_enrich_no_symbol_name(self, mock_db):
"""Test handling results without symbol names."""
with RelationshipEnricher(mock_db) as enricher:
results = [
{"path": "module.py", "score": 0.9, "excerpt": "code", "symbol": None},
]
enriched = enricher.enrich(results, limit=10)
assert "relationships" in enriched[0]
assert enriched[0]["relationships"] == []
def test_enrich_performance(self, mock_db):
"""Test that enrichment is fast (<100ms for 10 results)."""
with RelationshipEnricher(mock_db) as enricher:
results = [
{"path": "module.py", "score": 0.9, "excerpt": f"code{i}", "symbol": "main"}
for i in range(10)
]
start = time.perf_counter()
enricher.enrich(results, limit=10)
elapsed_ms = (time.perf_counter() - start) * 1000
assert elapsed_ms < 100, f"Enrichment took {elapsed_ms:.1f}ms, expected < 100ms"
def test_enrich_limit(self, mock_db):
"""Test that limit parameter is respected."""
with RelationshipEnricher(mock_db) as enricher:
results = [
{"path": "module.py", "score": 0.9, "symbol": "main"},
{"path": "module.py", "score": 0.8, "symbol": "helper"},
{"path": "utils.py", "score": 0.7, "symbol": "fetch"},
]
# Only enrich first 2
enriched = enricher.enrich(results, limit=2)
assert "relationships" in enriched[0]
assert "relationships" in enriched[1]
# Third result should NOT have relationships key
assert "relationships" not in enriched[2]
def test_connection_failure_graceful(self):
"""Test graceful handling when database doesn't exist."""
nonexistent = Path("/nonexistent/path/_index.db")
with RelationshipEnricher(nonexistent) as enricher:
results = [{"path": "test.py", "score": 0.9, "symbol": "test"}]
enriched = enricher.enrich(results)
# Should return original results without crashing
assert len(enriched) == 1
def test_incoming_type_conversion(self, mock_db):
"""Test that relationship types are correctly converted for incoming."""
with RelationshipEnricher(mock_db) as enricher:
results = [
{"path": "module.py", "score": 0.9, "symbol": "helper"},
]
enriched = enricher.enrich(results)
rels = enriched[0]["relationships"]
incoming = [r for r in rels if r["direction"] == "incoming"]
if incoming:
# calls should become called_by
assert incoming[0]["type"] == "called_by"
def test_context_manager(self, mock_db):
"""Test that context manager properly opens and closes connections."""
enricher = RelationshipEnricher(mock_db)
assert enricher.db_conn is not None
enricher.close()
assert enricher.db_conn is None
# Using context manager
with RelationshipEnricher(mock_db) as e:
assert e.db_conn is not None
assert e.db_conn is None
def test_relationship_data_structure(self, mock_db):
"""Test that relationship data has correct structure."""
with RelationshipEnricher(mock_db) as enricher:
results = [{"path": "module.py", "score": 0.9, "symbol": "main"}]
enriched = enricher.enrich(results)
rels = enriched[0]["relationships"]
for rel in rels:
# All relationships should have required fields
assert "type" in rel
assert "direction" in rel
assert "file" in rel
assert rel["direction"] in ["outgoing", "incoming"]
# Outgoing should have target, incoming should have source
if rel["direction"] == "outgoing":
assert "target" in rel
else:
assert "source" in rel

View File

@@ -0,0 +1,238 @@
"""Tests for symbol extraction and relationship tracking."""
import tempfile
from pathlib import Path
import pytest
from codexlens.indexing.symbol_extractor import SymbolExtractor
@pytest.fixture
def extractor():
"""Create a temporary symbol extractor for testing."""
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "test.db"
ext = SymbolExtractor(db_path)
ext.connect()
yield ext
ext.close()
class TestSymbolExtractor:
"""Test suite for SymbolExtractor."""
def test_database_schema_creation(self, extractor):
"""Test that database tables and indexes are created correctly."""
cursor = extractor.db_conn.cursor()
# Check symbols table exists
cursor.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='symbols'"
)
assert cursor.fetchone() is not None
# Check symbol_relationships table exists
cursor.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='symbol_relationships'"
)
assert cursor.fetchone() is not None
# Check indexes exist
cursor.execute(
"SELECT COUNT(*) FROM sqlite_master WHERE type='index' AND name LIKE 'idx_%'"
)
idx_count = cursor.fetchone()[0]
assert idx_count == 5
def test_python_function_extraction(self, extractor):
"""Test extracting functions from Python code."""
code = """
def hello():
pass
async def world():
pass
"""
symbols, _ = extractor.extract_from_file(Path("test.py"), code)
assert len(symbols) == 2
assert symbols[0]["name"] == "hello"
assert symbols[0]["kind"] == "function"
assert symbols[1]["name"] == "world"
assert symbols[1]["kind"] == "function"
def test_python_class_extraction(self, extractor):
"""Test extracting classes from Python code."""
code = """
class MyClass:
pass
class AnotherClass(BaseClass):
pass
"""
symbols, _ = extractor.extract_from_file(Path("test.py"), code)
assert len(symbols) == 2
assert symbols[0]["name"] == "MyClass"
assert symbols[0]["kind"] == "class"
assert symbols[1]["name"] == "AnotherClass"
assert symbols[1]["kind"] == "class"
def test_typescript_extraction(self, extractor):
"""Test extracting symbols from TypeScript code."""
code = """
export function calculateSum(a: number, b: number): number {
return a + b;
}
export class Calculator {
multiply(x: number, y: number) {
return x * y;
}
}
"""
symbols, _ = extractor.extract_from_file(Path("test.ts"), code)
assert len(symbols) == 2
assert symbols[0]["name"] == "calculateSum"
assert symbols[0]["kind"] == "function"
assert symbols[1]["name"] == "Calculator"
assert symbols[1]["kind"] == "class"
def test_javascript_extraction(self, extractor):
"""Test extracting symbols from JavaScript code."""
code = """
function processData(data) {
return data;
}
class DataProcessor {
transform(input) {
return input;
}
}
"""
symbols, _ = extractor.extract_from_file(Path("test.js"), code)
assert len(symbols) == 2
assert symbols[0]["name"] == "processData"
assert symbols[1]["name"] == "DataProcessor"
def test_relationship_extraction(self, extractor):
"""Test extracting relationships between symbols."""
code = """
def helper():
pass
def main():
helper()
print("done")
"""
_, relationships = extractor.extract_from_file(Path("test.py"), code)
# Should find calls to helper and print
call_targets = [r["target"] for r in relationships if r["type"] == "calls"]
assert "helper" in call_targets
def test_save_and_query_symbols(self, extractor):
"""Test saving symbols to database and querying them."""
code = """
def test_func():
pass
class TestClass:
pass
"""
symbols, _ = extractor.extract_from_file(Path("test.py"), code)
name_to_id = extractor.save_symbols(symbols)
assert len(name_to_id) == 2
assert "test_func" in name_to_id
assert "TestClass" in name_to_id
# Query database
cursor = extractor.db_conn.cursor()
cursor.execute("SELECT COUNT(*) FROM symbols")
count = cursor.fetchone()[0]
assert count == 2
def test_save_relationships(self, extractor):
"""Test saving relationships to database."""
code = """
def caller():
callee()
def callee():
pass
"""
symbols, relationships = extractor.extract_from_file(Path("test.py"), code)
name_to_id = extractor.save_symbols(symbols)
extractor.save_relationships(relationships, name_to_id)
# Query database
cursor = extractor.db_conn.cursor()
cursor.execute("SELECT COUNT(*) FROM symbol_relationships")
count = cursor.fetchone()[0]
assert count > 0
def test_qualified_name_generation(self, extractor):
"""Test that qualified names are generated correctly."""
code = """
class MyClass:
pass
"""
symbols, _ = extractor.extract_from_file(Path("module.py"), code)
assert symbols[0]["qualified_name"] == "module.MyClass"
def test_unsupported_language(self, extractor):
"""Test that unsupported languages return empty results."""
code = "some random code"
symbols, relationships = extractor.extract_from_file(Path("test.txt"), code)
assert len(symbols) == 0
assert len(relationships) == 0
def test_empty_file(self, extractor):
"""Test handling empty files."""
symbols, relationships = extractor.extract_from_file(Path("test.py"), "")
assert len(symbols) == 0
assert len(relationships) == 0
def test_complete_workflow(self, extractor):
"""Test complete workflow: extract, save, and verify."""
code = """
class UserService:
def get_user(self, user_id):
return fetch_user(user_id)
def main():
service = UserService()
service.get_user(1)
"""
file_path = Path("service.py")
symbols, relationships = extractor.extract_from_file(file_path, code)
# Save to database
name_to_id = extractor.save_symbols(symbols)
extractor.save_relationships(relationships, name_to_id)
# Verify symbols
cursor = extractor.db_conn.cursor()
cursor.execute("SELECT name, kind FROM symbols ORDER BY start_line")
db_symbols = cursor.fetchall()
assert len(db_symbols) == 2
assert db_symbols[0][0] == "UserService"
assert db_symbols[1][0] == "main"
# Verify relationships
cursor.execute(
"""
SELECT s.name, r.target_symbol_fqn, r.relationship_type
FROM symbol_relationships r
JOIN symbols s ON r.source_symbol_id = s.id
"""
)
db_rels = cursor.fetchall()
assert len(db_rels) > 0