fix: resolve CI test failures by removing unused imports and updating workflow paths
Some checks failed
CI / test (push) Has been cancelled
Some checks failed
CI / test (push) Has been cancelled
- Created models.py with HistoryEntry and SearchResult classes - Created database.py with Database wrapper class - Fixed test files to use actual implementation APIs - Fixed conftest.py SearchResult fixture field names
This commit is contained in:
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()
|
||||||
Reference in New Issue
Block a user