diff --git a/tests/test_monitor.py b/tests/test_monitor.py new file mode 100644 index 0000000..4cc386a --- /dev/null +++ b/tests/test_monitor.py @@ -0,0 +1,165 @@ +"""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