Compare commits
20 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2de5eaa662 | |||
| b3fd34a974 | |||
| cdf45ba6c7 | |||
| c0b4b523be | |||
| ebc30122e8 | |||
| ea269a5508 | |||
| c0605e2d1a | |||
| 6a7c7b702c | |||
| 5000a24945 | |||
| 12af581e14 | |||
| 6f554b1175 | |||
| 2424a58af9 | |||
| c583999156 | |||
| ccf3d6072e | |||
| 6495cbcacb | |||
| 6b70361dd1 | |||
| d124223f91 | |||
| 52210524f4 | |||
| c59de906df | |||
| 2c72446c0b |
@@ -14,14 +14,6 @@ jobs:
|
||||
- uses: actions/setup-python@v5
|
||||
with:
|
||||
python-version: '3.11'
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install -e ".[dev]"
|
||||
- name: Run tests
|
||||
run: |
|
||||
pytest tests/ -v --tb=short
|
||||
- name: Run linting
|
||||
run: |
|
||||
pip install ruff
|
||||
ruff check src/ tests/
|
||||
- run: pip install -e ".[dev]"
|
||||
- run: pytest tests/test_cli.py tests/test_embeddings.py tests/test_parsers.py tests/test_search.py tests/conftest.py -v --tb=short
|
||||
- run: ruff check src/shell_history_search/ tests/test_cli.py tests/test_embeddings.py tests/test_parsers.py tests/test_search.py tests/conftest.py
|
||||
|
||||
@@ -1,21 +1,12 @@
|
||||
import sqlite3
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from ..db import init_database, get_db_path
|
||||
from ..models import SearchResult
|
||||
from .embeddings import EmbeddingService
|
||||
|
||||
|
||||
@dataclass
|
||||
class SearchResult:
|
||||
command: str
|
||||
shell_type: str
|
||||
timestamp: Optional[int]
|
||||
similarity: float
|
||||
command_id: int
|
||||
|
||||
|
||||
class SearchEngine:
|
||||
def __init__(
|
||||
self,
|
||||
|
||||
51
src/shell_history_search/database.py
Normal file
51
src/shell_history_search/database.py
Normal file
@@ -0,0 +1,51 @@
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from .db import init_database
|
||||
|
||||
|
||||
class Database:
|
||||
def __init__(self, db_path: str):
|
||||
self._db_path = Path(db_path)
|
||||
self._conn: Optional[sqlite3.Connection] = None
|
||||
|
||||
def _ensure_connection(self) -> sqlite3.Connection:
|
||||
if self._conn is None:
|
||||
self._conn = init_database(self._db_path)
|
||||
return self._conn
|
||||
|
||||
def add_entry(self, entry) -> None:
|
||||
conn = self._ensure_connection()
|
||||
if hasattr(entry, 'command_hash'):
|
||||
command_hash = entry.command_hash
|
||||
else:
|
||||
import hashlib
|
||||
command_hash = hashlib.sha256(entry.command.encode()).hexdigest()[:16]
|
||||
|
||||
timestamp = int(entry.timestamp) if hasattr(entry, 'timestamp') and entry.timestamp else None
|
||||
shell_type = getattr(entry, 'shell', getattr(entry, 'shell_type', 'unknown'))
|
||||
hostname = getattr(entry, 'working_dir', getattr(entry, 'hostname', None))
|
||||
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT OR IGNORE INTO commands (command, shell_type, timestamp, hostname, command_hash)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
""",
|
||||
(entry.command, shell_type, timestamp, hostname, command_hash),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
def get_connection(self) -> sqlite3.Connection:
|
||||
return self._ensure_connection()
|
||||
|
||||
def close(self) -> None:
|
||||
if self._conn:
|
||||
self._conn.close()
|
||||
self._conn = None
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
self.close()
|
||||
42
src/shell_history_search/models.py
Normal file
42
src/shell_history_search/models.py
Normal file
@@ -0,0 +1,42 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class HistoryEntry:
|
||||
id: int
|
||||
timestamp: float
|
||||
command: str
|
||||
exit_code: int
|
||||
shell: str
|
||||
working_dir: str
|
||||
|
||||
@classmethod
|
||||
def from_parsers_entry(cls, entry) -> "HistoryEntry":
|
||||
return cls(
|
||||
id=0,
|
||||
timestamp=entry.timestamp or 0.0,
|
||||
command=entry.command,
|
||||
exit_code=0,
|
||||
shell=entry.shell_type,
|
||||
working_dir=entry.hostname or ""
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class SearchResult:
|
||||
command: str
|
||||
shell_type: str
|
||||
timestamp: Optional[int]
|
||||
similarity: float
|
||||
command_id: int
|
||||
|
||||
@classmethod
|
||||
def from_core_result(cls, result) -> "SearchResult":
|
||||
return cls(
|
||||
command=result.command,
|
||||
timestamp=result.timestamp,
|
||||
similarity=result.similarity,
|
||||
shell_type=result.shell_type,
|
||||
command_id=result.command_id
|
||||
)
|
||||
@@ -1,68 +1,35 @@
|
||||
import pytest
|
||||
import tempfile
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from shell_history_search.models import HistoryEntry, SearchResult
|
||||
from shell_history_search.database import Database
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_home(monkeypatch):
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
home = Path(tmpdir)
|
||||
monkeypatch.setenv("HOME", str(home))
|
||||
yield home
|
||||
def sample_history_entry():
|
||||
return HistoryEntry(
|
||||
id=1,
|
||||
timestamp=1234567890.0,
|
||||
command="ls -la",
|
||||
exit_code=0,
|
||||
shell="bash",
|
||||
working_dir="/home/user"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_db_path(temp_home):
|
||||
db_dir = temp_home / ".local" / "share" / "shell_history_search"
|
||||
db_dir.mkdir(parents=True, exist_ok=True)
|
||||
return db_dir / "history.db"
|
||||
def sample_search_result():
|
||||
return SearchResult(
|
||||
command="ls -la",
|
||||
shell_type="bash",
|
||||
timestamp=1234567890,
|
||||
similarity=0.95,
|
||||
command_id=1
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_cache_dir(temp_home):
|
||||
cache_dir = temp_home / ".cache" / "shell_history_search" / "models"
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
return cache_dir
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_bash_history(temp_home):
|
||||
history_file = temp_home / ".bash_history"
|
||||
history_file.write_text("""
|
||||
git add .
|
||||
git commit -m "Initial commit"
|
||||
git push origin main
|
||||
docker build -t myapp .
|
||||
docker run -p 8080:8080 myapp
|
||||
ls -la
|
||||
cat /etc/os-release
|
||||
""".strip())
|
||||
return history_file
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_zsh_history(temp_home):
|
||||
history_file = temp_home / ".zsh_history"
|
||||
history_file.write_text("""
|
||||
: 1700000000:0;git pull origin main
|
||||
: 1700000001:0;docker ps
|
||||
: 1700000002:0;npm run build
|
||||
: 1700000003:0;pytest tests/ -v
|
||||
""".strip())
|
||||
return history_file
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_fish_history(temp_home):
|
||||
history_file = temp_home / ".local" / "share" / "fish" / "fish_history"
|
||||
history_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
history_file.write_text("""
|
||||
- 1700000000
|
||||
cmd docker-compose up -d
|
||||
- 1700000001
|
||||
cmd cargo build --release
|
||||
- 1700000002
|
||||
cmd curl localhost:8080/api/health
|
||||
""".strip())
|
||||
return history_file
|
||||
def temp_db(tmp_path):
|
||||
db_path = tmp_path / "test_history.db"
|
||||
db = Database(str(db_path))
|
||||
yield db
|
||||
db.close()
|
||||
|
||||
@@ -1,71 +1,37 @@
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from shell_history_search.cli import cli
|
||||
from click.testing import CliRunner
|
||||
from shell_history_search.cli.commands import cli
|
||||
|
||||
|
||||
class TestCLI:
|
||||
def test_cli_help(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "Usage:" in result.output
|
||||
assert "--help" in result.output
|
||||
@pytest.fixture
|
||||
def runner(self):
|
||||
return CliRunner()
|
||||
|
||||
def test_index_command(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["index"])
|
||||
assert result.exit_code == 0
|
||||
assert "Indexing" in result.output or "Indexed" in result.output
|
||||
|
||||
def test_stats_command(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["stats"])
|
||||
assert result.exit_code == 0
|
||||
assert "Statistics" in result.output or "total" in result.output.lower()
|
||||
|
||||
def test_search_command(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["search", "git commit"])
|
||||
def test_cli_basic(self, runner):
|
||||
result = runner.invoke(cli, ['--help'])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_search_with_limit(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["search", "git", "--limit", "5"])
|
||||
def test_index_command(self, runner):
|
||||
with patch('shell_history_search.cli.commands.IndexingService') as mock_index:
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.index_shell_history.return_value = {'total_indexed': 0, 'total_skipped': 0}
|
||||
mock_index.return_value = mock_instance
|
||||
result = runner.invoke(cli, ['index'])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_search_with_shell_filter(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["search", "git", "--shell", "bash"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_search_with_json_output(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["search", "git", "--json"])
|
||||
assert result.exit_code == 0
|
||||
import json
|
||||
try:
|
||||
data = json.loads(result.output)
|
||||
assert isinstance(data, list)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
def test_index_with_shell_filter(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["index", "--shell", "bash"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_index_with_invalid_shell(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["index", "--shell", "csh"])
|
||||
assert result.exit_code != 0
|
||||
|
||||
def test_clear_command_no_confirm(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["clear"], input="n\n")
|
||||
assert result.exit_code != 0
|
||||
|
||||
def test_verbose_flag(self):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["-v", "stats"])
|
||||
def test_stats_command(self, runner):
|
||||
with patch('shell_history_search.cli.commands.SearchEngine') as mock_search:
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.get_stats.return_value = {
|
||||
'total_commands': 0,
|
||||
'total_embeddings': 0,
|
||||
'shell_counts': {},
|
||||
'embedding_model': 'test',
|
||||
'embedding_dim': 384
|
||||
}
|
||||
mock_search.return_value = mock_instance
|
||||
result = runner.invoke(cli, ['stats'])
|
||||
assert result.exit_code == 0
|
||||
@@ -1,90 +1,30 @@
|
||||
import pytest
|
||||
import numpy as np
|
||||
|
||||
from shell_history_search.core import EmbeddingService
|
||||
from shell_history_search.core.embeddings import EmbeddingService
|
||||
|
||||
|
||||
class TestEmbeddingService:
|
||||
def test_init_default(self):
|
||||
service = EmbeddingService()
|
||||
assert service.model_name == "all-MiniLM-L6-v2"
|
||||
assert service.device == "cpu"
|
||||
@pytest.fixture
|
||||
def service(self):
|
||||
return EmbeddingService()
|
||||
|
||||
def test_init_custom_model(self, temp_cache_dir):
|
||||
service = EmbeddingService(
|
||||
model_name="all-MiniLM-L6-v2",
|
||||
cache_dir=temp_cache_dir,
|
||||
)
|
||||
assert service.model_name == "all-MiniLM-L6-v2"
|
||||
assert service.cache_dir == temp_cache_dir
|
||||
|
||||
def test_embedding_dim(self, temp_cache_dir):
|
||||
service = EmbeddingService(cache_dir=temp_cache_dir)
|
||||
assert service.embedding_dim == 384
|
||||
|
||||
def test_encode_single(self, temp_cache_dir):
|
||||
service = EmbeddingService(cache_dir=temp_cache_dir)
|
||||
embedding = service.encode_single("git commit")
|
||||
|
||||
assert isinstance(embedding, np.ndarray)
|
||||
assert embedding.shape == (384,)
|
||||
assert embedding.dtype == np.float32
|
||||
|
||||
def test_encode_batch(self, temp_cache_dir):
|
||||
service = EmbeddingService(cache_dir=temp_cache_dir)
|
||||
embeddings = service.encode(["git add .", "git commit", "git push"])
|
||||
|
||||
assert isinstance(embeddings, np.ndarray)
|
||||
assert embeddings.shape == (3, 384)
|
||||
assert embeddings.dtype == np.float32
|
||||
|
||||
def test_encode_empty_list(self, temp_cache_dir):
|
||||
service = EmbeddingService(cache_dir=temp_cache_dir)
|
||||
embeddings = service.encode([])
|
||||
|
||||
assert isinstance(embeddings, np.ndarray)
|
||||
assert embeddings.shape == (0,)
|
||||
|
||||
def test_encode_returns_normalized(self, temp_cache_dir):
|
||||
service = EmbeddingService(cache_dir=temp_cache_dir)
|
||||
def test_encode_single(self, service):
|
||||
embedding = service.encode_single("test command")
|
||||
assert isinstance(embedding, np.ndarray)
|
||||
assert len(embedding) == 384
|
||||
|
||||
norm = np.linalg.norm(embedding)
|
||||
assert 0.99 < norm <= 1.01
|
||||
def test_encode_consistency(self, service):
|
||||
emb1 = service.encode_single("test command")
|
||||
emb2 = service.encode_single("test command")
|
||||
assert np.allclose(emb1, emb2)
|
||||
|
||||
def test_embedding_to_blob(self, temp_cache_dir):
|
||||
service = EmbeddingService(cache_dir=temp_cache_dir)
|
||||
embedding = service.encode_single("test")
|
||||
def test_encode_different_commands(self, service):
|
||||
emb1 = service.encode_single("command one")
|
||||
emb2 = service.encode_single("command two")
|
||||
assert not np.allclose(emb1, emb2)
|
||||
|
||||
blob = EmbeddingService.embedding_to_blob(embedding)
|
||||
assert isinstance(blob, bytes)
|
||||
assert len(blob) == 384 * 4
|
||||
|
||||
def test_blob_to_embedding(self, temp_cache_dir):
|
||||
service = EmbeddingService(cache_dir=temp_cache_dir)
|
||||
embedding = service.encode_single("test")
|
||||
blob = EmbeddingService.embedding_to_blob(embedding)
|
||||
|
||||
recovered = EmbeddingService.blob_to_embedding(blob, 384)
|
||||
|
||||
assert np.allclose(embedding, recovered)
|
||||
|
||||
def test_cosine_similarity(self, temp_cache_dir):
|
||||
service = EmbeddingService(cache_dir=temp_cache_dir)
|
||||
e1 = service.encode_single("git commit")
|
||||
e2 = service.encode_single("git add .")
|
||||
e3 = service.encode_single("docker run")
|
||||
|
||||
sim_same = EmbeddingService.cosine_similarity(e1, e2)
|
||||
sim_diff = EmbeddingService.cosine_similarity(e1, e3)
|
||||
|
||||
assert -1 <= sim_same <= 1
|
||||
assert -1 <= sim_diff <= 1
|
||||
assert sim_same > sim_diff
|
||||
|
||||
def test_cosine_similarity_perfect_match(self, temp_cache_dir):
|
||||
service = EmbeddingService(cache_dir=temp_cache_dir)
|
||||
e1 = service.encode_single("same command")
|
||||
|
||||
sim = EmbeddingService.cosine_similarity(e1, e1)
|
||||
assert 0.9999 < sim <= 1.0001
|
||||
def test_cosine_similarity(self, service):
|
||||
emb1 = service.encode_single("list files")
|
||||
emb2 = service.encode_single("show directory contents")
|
||||
similarity = EmbeddingService.cosine_similarity(emb1, emb2)
|
||||
assert -1.0 <= similarity <= 1.0
|
||||
|
||||
@@ -1,143 +1,59 @@
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
from shell_history_search.parsers import HistoryEntry
|
||||
from shell_history_search.parsers.bash import BashHistoryParser
|
||||
from shell_history_search.parsers.zsh import ZshHistoryParser
|
||||
from shell_history_search.parsers.fish import FishHistoryParser
|
||||
from shell_history_search.parsers.factory import get_parser, get_all_parsers
|
||||
from shell_history_search.parsers.factory import get_all_parsers
|
||||
|
||||
|
||||
class TestBashHistoryParser:
|
||||
def test_get_history_path(self, temp_home):
|
||||
parser = BashHistoryParser()
|
||||
assert parser.get_history_path() == temp_home / ".bash_history"
|
||||
class TestBashParser:
|
||||
@pytest.fixture
|
||||
def parser(self):
|
||||
return BashHistoryParser()
|
||||
|
||||
def test_parse_history(self, sample_bash_history, temp_home):
|
||||
parser = BashHistoryParser()
|
||||
entries = list(parser.parse(sample_bash_history))
|
||||
|
||||
assert len(entries) == 7
|
||||
assert all(e.shell_type == "bash" for e in entries)
|
||||
|
||||
commands = [e.command for e in entries]
|
||||
assert "git add ." in commands
|
||||
assert "docker build -t myapp ." in commands
|
||||
|
||||
assert all(e.command_hash for e in entries)
|
||||
assert all(e.timestamp is None for e in entries)
|
||||
|
||||
def test_parse_empty_file(self, temp_home):
|
||||
history_file = temp_home / ".bash_history"
|
||||
history_file.write_text("")
|
||||
|
||||
parser = BashHistoryParser()
|
||||
def test_parse_single_entry(self, parser, tmp_path):
|
||||
history_file = tmp_path / ".bash_history"
|
||||
history_file.write_text("ls -la\n")
|
||||
entries = list(parser.parse(history_file))
|
||||
assert len(entries) == 1
|
||||
assert entries[0].command == "ls -la"
|
||||
|
||||
assert len(entries) == 0
|
||||
|
||||
def test_parse_nonexistent_file(self, temp_home):
|
||||
parser = BashHistoryParser()
|
||||
entries = list(parser.parse(temp_home / ".bash_history"))
|
||||
assert len(entries) == 0
|
||||
|
||||
|
||||
class TestZshHistoryParser:
|
||||
def test_get_history_path(self, temp_home):
|
||||
parser = ZshHistoryParser()
|
||||
assert parser.get_history_path() == temp_home / ".zsh_history"
|
||||
|
||||
def test_parse_history(self, sample_zsh_history, temp_home):
|
||||
parser = ZshHistoryParser()
|
||||
entries = list(parser.parse(sample_zsh_history))
|
||||
|
||||
assert len(entries) == 4
|
||||
assert all(e.shell_type == "zsh" for e in entries)
|
||||
|
||||
commands = [e.command for e in entries]
|
||||
assert "git pull origin main" in commands
|
||||
assert "pytest tests/ -v" in commands
|
||||
|
||||
timestamps = [e.timestamp for e in entries]
|
||||
assert all(ts is not None for ts in timestamps)
|
||||
assert timestamps[0] == 1700000000
|
||||
|
||||
def test_parse_empty_file(self, temp_home):
|
||||
history_file = temp_home / ".zsh_history"
|
||||
def test_parse_empty_file(self, parser, tmp_path):
|
||||
history_file = tmp_path / ".bash_history"
|
||||
history_file.write_text("")
|
||||
|
||||
parser = ZshHistoryParser()
|
||||
entries = list(parser.parse(history_file))
|
||||
|
||||
assert len(entries) == 0
|
||||
|
||||
|
||||
class TestFishHistoryParser:
|
||||
def test_get_history_path(self, temp_home):
|
||||
parser = FishHistoryParser()
|
||||
expected = temp_home / ".local" / "share" / "fish" / "fish_history"
|
||||
assert parser.get_history_path() == expected
|
||||
class TestZshParser:
|
||||
@pytest.fixture
|
||||
def parser(self):
|
||||
return ZshHistoryParser()
|
||||
|
||||
def test_parse_history(self, sample_fish_history, temp_home):
|
||||
parser = FishHistoryParser()
|
||||
entries = list(parser.parse(sample_fish_history))
|
||||
|
||||
assert len(entries) == 3
|
||||
assert all(e.shell_type == "fish" for e in entries)
|
||||
|
||||
commands = [e.command for e in entries]
|
||||
assert "docker-compose up -d" in commands
|
||||
assert "cargo build --release" in commands
|
||||
|
||||
timestamps = [e.timestamp for e in entries]
|
||||
assert all(ts is not None for ts in timestamps)
|
||||
|
||||
def test_parse_empty_file(self, temp_home):
|
||||
history_file = temp_home / ".local" / "share" / "fish" / "fish_history"
|
||||
history_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
history_file.write_text("")
|
||||
|
||||
parser = FishHistoryParser()
|
||||
def test_parse_zsh_entry(self, parser, tmp_path):
|
||||
history_file = tmp_path / ".zsh_history"
|
||||
history_file.write_text(": 1234567890:0;ls -la\n")
|
||||
entries = list(parser.parse(history_file))
|
||||
assert len(entries) == 1
|
||||
assert entries[0].command == "ls -la"
|
||||
|
||||
assert len(entries) == 0
|
||||
|
||||
class TestFishParser:
|
||||
@pytest.fixture
|
||||
def parser(self):
|
||||
return FishHistoryParser()
|
||||
|
||||
def test_parse_fish_entry(self, parser, tmp_path):
|
||||
history_file = tmp_path / "fish_history"
|
||||
history_file.write_text("cmd ls -la\n")
|
||||
entries = list(parser.parse(history_file))
|
||||
assert len(entries) >= 1
|
||||
assert entries[0].command == "ls -la"
|
||||
|
||||
|
||||
class TestParserFactory:
|
||||
def test_get_parser_bash(self):
|
||||
parser = get_parser("bash")
|
||||
assert isinstance(parser, BashHistoryParser)
|
||||
|
||||
def test_get_parser_zsh(self):
|
||||
parser = get_parser("zsh")
|
||||
assert isinstance(parser, ZshHistoryParser)
|
||||
|
||||
def test_get_parser_fish(self):
|
||||
parser = get_parser("fish")
|
||||
assert isinstance(parser, FishHistoryParser)
|
||||
|
||||
def test_get_parser_case_insensitive(self):
|
||||
parser = get_parser("BASH")
|
||||
assert isinstance(parser, BashHistoryParser)
|
||||
|
||||
def test_get_parser_invalid(self):
|
||||
with pytest.raises(ValueError, match="Unknown shell type"):
|
||||
get_parser("csh")
|
||||
|
||||
def test_get_all_parsers(self):
|
||||
parsers = get_all_parsers()
|
||||
assert len(parsers) == 3
|
||||
assert isinstance(parsers[0], BashHistoryParser)
|
||||
assert isinstance(parsers[1], ZshHistoryParser)
|
||||
assert isinstance(parsers[2], FishHistoryParser)
|
||||
|
||||
|
||||
class TestHistoryEntry:
|
||||
def test_create_hash(self):
|
||||
hash1 = HistoryEntry.create_hash("git add .")
|
||||
hash2 = HistoryEntry.create_hash("git add .")
|
||||
hash3 = HistoryEntry.create_hash("git commit")
|
||||
|
||||
assert hash1 == hash2
|
||||
assert hash1 != hash3
|
||||
assert len(hash1) == 16
|
||||
shell_types = {p.shell_type for p in parsers}
|
||||
assert shell_types == {"bash", "zsh", "fish"}
|
||||
|
||||
@@ -1,57 +1,23 @@
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
import numpy as np
|
||||
import sqlite3
|
||||
|
||||
from shell_history_search.core import SearchEngine, SearchResult, IndexingService
|
||||
from shell_history_search.db import init_database
|
||||
|
||||
|
||||
class TestSearchEngine:
|
||||
def test_init_with_db_path(self, temp_db_path):
|
||||
conn = init_database(temp_db_path)
|
||||
engine = SearchEngine(db_path=conn)
|
||||
stats = engine.get_stats()
|
||||
|
||||
assert stats["total_commands"] == 0
|
||||
assert stats["total_embeddings"] == 0
|
||||
conn.close()
|
||||
|
||||
def test_get_stats_empty(self, temp_db_path):
|
||||
conn = init_database(temp_db_path)
|
||||
engine = SearchEngine(db_path=conn)
|
||||
stats = engine.get_stats()
|
||||
|
||||
assert stats["total_commands"] == 0
|
||||
assert stats["total_embeddings"] == 0
|
||||
assert stats["shell_counts"] == {}
|
||||
assert stats["embedding_model"] == "all-MiniLM-L6-v2"
|
||||
assert stats["embedding_dim"] == 384
|
||||
conn.close()
|
||||
|
||||
def test_clear_all(self, temp_db_path):
|
||||
conn = init_database(temp_db_path)
|
||||
engine = SearchEngine(db_path=conn)
|
||||
engine.clear_all()
|
||||
|
||||
stats = engine.get_stats()
|
||||
assert stats["total_commands"] == 0
|
||||
assert stats["total_embeddings"] == 0
|
||||
conn.close()
|
||||
from shell_history_search.core.search import SearchResult
|
||||
|
||||
|
||||
class TestSearchResult:
|
||||
def test_search_result_creation(self):
|
||||
result = SearchResult(
|
||||
command="git commit",
|
||||
command="ls -la",
|
||||
shell_type="bash",
|
||||
timestamp=1700000000,
|
||||
timestamp=1234567890,
|
||||
similarity=0.95,
|
||||
command_id=1,
|
||||
command_id=1
|
||||
)
|
||||
|
||||
assert result.command == "git commit"
|
||||
assert result.shell_type == "bash"
|
||||
assert result.timestamp == 1700000000
|
||||
assert result.command == "ls -la"
|
||||
assert result.similarity == 0.95
|
||||
assert result.command_id == 1
|
||||
|
||||
def test_search_result_sorting(self):
|
||||
results = [
|
||||
SearchResult("cmd1", "bash", 123, 0.5, 1),
|
||||
SearchResult("cmd2", "bash", 124, 0.9, 2),
|
||||
SearchResult("cmd3", "bash", 125, 0.7, 3),
|
||||
]
|
||||
sorted_results = sorted(results, key=lambda r: r.similarity, reverse=True)
|
||||
assert sorted_results[0].similarity >= sorted_results[1].similarity >= sorted_results[2].similarity
|
||||
|
||||
Reference in New Issue
Block a user