166 lines
5.3 KiB
Python
166 lines
5.3 KiB
Python
"""Tests for DevTrace monitor module."""
|
|
|
|
import pytest
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
|
|
from src.storage.database import Database
|
|
from src.monitor.filesystem import FileSystemMonitor, FileEventHandler
|
|
from src.monitor.commands import CommandCapture
|
|
from src.monitor.git import GitTracker
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_db():
|
|
"""Create a temporary database for testing."""
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
db_path = Path(f.name)
|
|
|
|
db = Database(db_path)
|
|
db.initialize()
|
|
|
|
yield db
|
|
|
|
db.close()
|
|
db_path.unlink(missing_ok=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir():
|
|
"""Create a temporary directory for testing."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
|
|
class TestFileSystemMonitor:
|
|
"""Tests for FileSystemMonitor."""
|
|
|
|
def test_monitor_creation(self, temp_db):
|
|
"""Test monitor creation."""
|
|
session_id = temp_db.create_session("Test", "/tmp").id
|
|
monitor = FileSystemMonitor(temp_db, session_id)
|
|
|
|
assert monitor.session_id == session_id
|
|
assert monitor.observer is None
|
|
|
|
def test_event_handler_should_watch(self, temp_db):
|
|
"""Test event handler path filtering."""
|
|
session_id = temp_db.create_session("Test", "/tmp").id
|
|
handler = FileEventHandler(temp_db, session_id)
|
|
|
|
assert handler._should_watch("/project/main.py") is True
|
|
assert handler._should_watch("/project/src/app.js") is True
|
|
assert handler._should_watch("/project/README.md") is True
|
|
|
|
def test_event_handler_ignores_ignored_patterns(self, temp_db):
|
|
"""Test event handler ignores specified patterns."""
|
|
session_id = temp_db.create_session("Test", "/tmp").id
|
|
handler = FileEventHandler(
|
|
temp_db,
|
|
session_id,
|
|
ignored_patterns={"__pycache__", ".git"}
|
|
)
|
|
|
|
assert handler._should_watch("/project/__pycache__/cache.py") is False
|
|
assert handler._should_watch("/project/.git/config") is False
|
|
assert handler._should_watch("/project/src/main.py") is True
|
|
|
|
def test_compute_content_hash(self, temp_db, temp_dir):
|
|
"""Test content hash computation."""
|
|
session_id = temp_db.create_session("Test", str(temp_dir)).id
|
|
handler = FileEventHandler(temp_db, session_id)
|
|
|
|
test_file = temp_dir / "test.py"
|
|
test_file.write_text("print('hello')")
|
|
|
|
hash1 = handler._compute_content_hash(str(test_file))
|
|
hash2 = handler._compute_content_hash(str(test_file))
|
|
|
|
assert hash1 is not None
|
|
assert hash1 == hash2
|
|
|
|
test_file.write_text("print('world')")
|
|
hash3 = handler._compute_content_hash(str(test_file))
|
|
|
|
assert hash1 != hash3
|
|
|
|
|
|
class TestCommandCapture:
|
|
"""Tests for CommandCapture."""
|
|
|
|
def test_capture_creation(self, temp_db):
|
|
"""Test capture creation."""
|
|
session_id = temp_db.create_session("Test", "/tmp").id
|
|
capture = CommandCapture(temp_db, session_id)
|
|
|
|
assert capture.session_id == session_id
|
|
|
|
@patch('builtins.open', side_effect=FileNotFoundError)
|
|
def test_capture_handles_missing_history(self, temp_db):
|
|
"""Test capture handles missing history file."""
|
|
session_id = temp_db.create_session("Test", "/tmp").id
|
|
capture = CommandCapture(temp_db, session_id)
|
|
|
|
result = capture.capture_new_commands()
|
|
|
|
assert result == []
|
|
|
|
def test_capture_command(self, temp_db):
|
|
"""Test direct command capture."""
|
|
session_id = temp_db.create_session("Test", "/tmp").id
|
|
capture = CommandCapture(temp_db, session_id)
|
|
|
|
event_id = capture.capture_command("ls -la", exit_code=0)
|
|
|
|
assert event_id is not None
|
|
|
|
|
|
class TestGitTracker:
|
|
"""Tests for GitTracker."""
|
|
|
|
def test_tracker_creation(self, temp_db):
|
|
"""Test tracker creation."""
|
|
session_id = temp_db.create_session("Test", "/tmp").id
|
|
tracker = GitTracker(temp_db, session_id)
|
|
|
|
assert tracker.session_id == session_id
|
|
|
|
@patch('subprocess.run')
|
|
def test_get_current_branch(self, mock_run, temp_db, temp_dir):
|
|
"""Test getting current branch."""
|
|
mock_run.return_value.stdout = "main\n"
|
|
mock_run.return_value.returncode = 0
|
|
|
|
session_id = temp_db.create_session("Test", str(temp_dir)).id
|
|
tracker = GitTracker(temp_db, session_id)
|
|
|
|
branch = tracker._get_current_branch(str(temp_dir))
|
|
|
|
assert branch == "main"
|
|
|
|
@patch('subprocess.run')
|
|
def test_get_current_commit(self, mock_run, temp_db, temp_dir):
|
|
"""Test getting current commit."""
|
|
mock_run.return_value.stdout = "abc123def456\n"
|
|
mock_run.return_value.returncode = 0
|
|
|
|
session_id = temp_db.create_session("Test", str(temp_dir)).id
|
|
tracker = GitTracker(temp_db, session_id)
|
|
|
|
commit = tracker._get_current_commit(str(temp_dir))
|
|
|
|
assert commit == "abc123def456"
|
|
|
|
@patch('subprocess.run')
|
|
def test_get_current_commit_not_git_repo(self, mock_run, temp_db, temp_dir):
|
|
"""Test getting commit when not in git repo."""
|
|
mock_run.return_value.returncode = 128
|
|
|
|
session_id = temp_db.create_session("Test", str(temp_dir)).id
|
|
tracker = GitTracker(temp_db, session_id)
|
|
|
|
commit = tracker._get_current_commit(str(temp_dir))
|
|
|
|
assert commit is None
|