This commit is contained in:
218
tests/test_storage.py
Normal file
218
tests/test_storage.py
Normal file
@@ -0,0 +1,218 @@
|
||||
"""Tests for DevTrace storage module."""
|
||||
|
||||
import pytest
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from src.storage.database import Database
|
||||
from src.storage.models import Session, FileEvent, CommandEvent, GitEvent
|
||||
|
||||
|
||||
@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)
|
||||
|
||||
|
||||
class TestDatabase:
|
||||
"""Tests for Database class."""
|
||||
|
||||
def test_initialize(self, temp_db):
|
||||
"""Test database initialization."""
|
||||
assert temp_db.db_path.exists()
|
||||
|
||||
def test_create_session(self, temp_db):
|
||||
"""Test creating a session."""
|
||||
session = temp_db.create_session(
|
||||
name="Test Session",
|
||||
directory="/tmp/test"
|
||||
)
|
||||
|
||||
assert session.id is not None
|
||||
assert session.name == "Test Session"
|
||||
assert session.directory == "/tmp/test"
|
||||
|
||||
def test_get_session(self, temp_db):
|
||||
"""Test retrieving a session."""
|
||||
created = temp_db.create_session(
|
||||
name="Test Session",
|
||||
directory="/tmp/test"
|
||||
)
|
||||
|
||||
retrieved = temp_db.get_session(created.id)
|
||||
|
||||
assert retrieved is not None
|
||||
assert retrieved.id == created.id
|
||||
assert retrieved.name == "Test Session"
|
||||
|
||||
def test_get_all_sessions(self, temp_db):
|
||||
"""Test retrieving all sessions."""
|
||||
temp_db.create_session("Session 1", "/tmp/test1")
|
||||
temp_db.create_session("Session 2", "/tmp/test2")
|
||||
|
||||
sessions = temp_db.get_all_sessions()
|
||||
|
||||
assert len(sessions) == 2
|
||||
|
||||
def test_end_session(self, temp_db):
|
||||
"""Test ending a session."""
|
||||
session = temp_db.create_session("Test", "/tmp/test")
|
||||
result = temp_db.end_session(session.id)
|
||||
|
||||
assert result is True
|
||||
|
||||
updated = temp_db.get_session(session.id)
|
||||
assert updated.end_time is not None
|
||||
|
||||
def test_add_file_event(self, temp_db):
|
||||
"""Test adding file events."""
|
||||
session = temp_db.create_session("Test", "/tmp/test")
|
||||
|
||||
event_id = temp_db.add_file_event(
|
||||
session_id=session.id,
|
||||
event_type="created",
|
||||
file_path="/tmp/test/file.py",
|
||||
content_hash="abc123"
|
||||
)
|
||||
|
||||
assert event_id is not None
|
||||
|
||||
events = temp_db.get_file_events(session.id)
|
||||
|
||||
assert len(events) == 1
|
||||
assert events[0].event_type == "created"
|
||||
assert events[0].file_path == "/tmp/test/file.py"
|
||||
|
||||
def test_add_command_event(self, temp_db):
|
||||
"""Test adding command events."""
|
||||
session = temp_db.create_session("Test", "/tmp/test")
|
||||
|
||||
event_id = temp_db.add_command_event(
|
||||
session_id=session.id,
|
||||
command="pytest tests/",
|
||||
exit_code=0
|
||||
)
|
||||
|
||||
assert event_id is not None
|
||||
|
||||
events = temp_db.get_command_events(session.id)
|
||||
|
||||
assert len(events) == 1
|
||||
assert events[0].command == "pytest tests/"
|
||||
assert events[0].exit_code == 0
|
||||
|
||||
def test_add_git_event(self, temp_db):
|
||||
"""Test adding git events."""
|
||||
session = temp_db.create_session("Test", "/tmp/test")
|
||||
|
||||
event_id = temp_db.add_git_event(
|
||||
session_id=session.id,
|
||||
operation="commit",
|
||||
branch="main",
|
||||
commit_hash="abc123def",
|
||||
details="Initial commit"
|
||||
)
|
||||
|
||||
assert event_id is not None
|
||||
|
||||
events = temp_db.get_git_events(session.id)
|
||||
|
||||
assert len(events) == 1
|
||||
assert events[0].operation == "commit"
|
||||
assert events[0].branch == "main"
|
||||
|
||||
def test_get_session_events(self, temp_db):
|
||||
"""Test getting all events for a session."""
|
||||
session = temp_db.create_session("Test", "/tmp/test")
|
||||
|
||||
temp_db.add_file_event(session.id, "created", "/tmp/test/file1.py")
|
||||
temp_db.add_command_event(session.id, "pytest")
|
||||
temp_db.add_git_event(session.id, "commit")
|
||||
|
||||
events = temp_db.get_session_events(session.id)
|
||||
|
||||
assert len(events) == 3
|
||||
|
||||
def test_get_stats(self, temp_db):
|
||||
"""Test getting database statistics."""
|
||||
session = temp_db.create_session("Test", "/tmp/test")
|
||||
|
||||
temp_db.add_file_event(session.id, "created", "/tmp/test/file1.py")
|
||||
temp_db.add_file_event(session.id, "modified", "/tmp/test/file2.py")
|
||||
temp_db.add_command_event(session.id, "ls -la")
|
||||
|
||||
stats = temp_db.get_stats()
|
||||
|
||||
assert stats["total_sessions"] == 1
|
||||
assert stats["file_events"] == 2
|
||||
assert stats["command_events"] == 1
|
||||
|
||||
|
||||
class TestModels:
|
||||
"""Tests for data models."""
|
||||
|
||||
def test_session_duration(self):
|
||||
"""Test session duration calculation."""
|
||||
start = datetime(2024, 1, 1, 10, 0, 0)
|
||||
end = datetime(2024, 1, 1, 11, 0, 0)
|
||||
|
||||
session = Session(
|
||||
id=1,
|
||||
name="Test",
|
||||
directory="/tmp",
|
||||
start_time=start,
|
||||
end_time=end
|
||||
)
|
||||
|
||||
assert session.duration == 3600.0
|
||||
|
||||
def test_file_event_filename(self):
|
||||
"""Test file event filename extraction."""
|
||||
event = FileEvent(
|
||||
id=1,
|
||||
session_id=1,
|
||||
event_type="created",
|
||||
file_path="/home/user/project/src/main.py"
|
||||
)
|
||||
|
||||
assert event.filename == "main.py"
|
||||
assert event.extension == "py"
|
||||
|
||||
def test_command_event_success(self):
|
||||
"""Test command event success check."""
|
||||
success_event = CommandEvent(
|
||||
id=1,
|
||||
session_id=1,
|
||||
command="ls",
|
||||
exit_code=0
|
||||
)
|
||||
|
||||
fail_event = CommandEvent(
|
||||
id=2,
|
||||
session_id=1,
|
||||
command="ls",
|
||||
exit_code=1
|
||||
)
|
||||
|
||||
assert success_event.was_successful is True
|
||||
assert fail_event.was_successful is False
|
||||
|
||||
def test_git_event_short_hash(self):
|
||||
"""Test git event short hash extraction."""
|
||||
event = GitEvent(
|
||||
id=1,
|
||||
session_id=1,
|
||||
operation="commit",
|
||||
commit_hash="abc123def456"
|
||||
)
|
||||
|
||||
assert event.short_hash == "abc123d"
|
||||
Reference in New Issue
Block a user