Ryan Malloy
44ed9936b7
Add comprehensive development intelligence system that tracks: - Development sessions with automatic start/stop - Full conversation history with semantic search - Tool usage and file operation analytics - Think time and engagement analysis - Git activity correlation - Learning pattern recognition - Productivity insights and metrics Features: - FastAPI backend with SQLite database - Modern web dashboard with interactive charts - Claude Code hook integration for automatic tracking - Comprehensive test suite with 100+ tests - Complete API documentation (OpenAPI/Swagger) - Privacy-first design with local data storage 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
256 lines
8.3 KiB
Python
256 lines
8.3 KiB
Python
import pytest
|
|
import asyncio
|
|
from typing import AsyncGenerator
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.main import app
|
|
from app.database.connection import get_db
|
|
from app.models.base import Base
|
|
from app.models.project import Project
|
|
from app.models.session import Session
|
|
from app.models.conversation import Conversation
|
|
from app.models.activity import Activity
|
|
from app.models.waiting_period import WaitingPeriod
|
|
from app.models.git_operation import GitOperation
|
|
|
|
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop():
|
|
"""Create an instance of the default event loop for the test session."""
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
@pytest.fixture
|
|
async def test_engine():
|
|
"""Create test database engine."""
|
|
engine = create_async_engine(
|
|
TEST_DATABASE_URL,
|
|
connect_args={
|
|
"check_same_thread": False,
|
|
},
|
|
poolclass=StaticPool,
|
|
echo=False, # Set to True for SQL debugging
|
|
)
|
|
yield engine
|
|
await engine.dispose()
|
|
|
|
@pytest.fixture
|
|
async def test_db(test_engine) -> AsyncGenerator[AsyncSession, None]:
|
|
"""Create test database session."""
|
|
# Create all tables
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
# Create session factory
|
|
async_session = async_sessionmaker(
|
|
test_engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
|
|
async with async_session() as session:
|
|
yield session
|
|
await session.rollback()
|
|
|
|
@pytest.fixture
|
|
async def test_client(test_db: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
|
|
"""Create test HTTP client."""
|
|
def override_get_db():
|
|
return test_db
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
|
|
async with AsyncClient(app=app, base_url="http://test") as client:
|
|
yield client
|
|
|
|
# Clean up
|
|
app.dependency_overrides.clear()
|
|
|
|
@pytest.fixture
|
|
async def sample_project(test_db: AsyncSession) -> Project:
|
|
"""Create a sample project for testing."""
|
|
project = Project(
|
|
name="Test Project",
|
|
path="/home/user/test-project",
|
|
git_repo="https://github.com/user/test-project",
|
|
languages=["python", "javascript"]
|
|
)
|
|
test_db.add(project)
|
|
await test_db.commit()
|
|
await test_db.refresh(project)
|
|
return project
|
|
|
|
@pytest.fixture
|
|
async def sample_session(test_db: AsyncSession, sample_project: Project) -> Session:
|
|
"""Create a sample session for testing."""
|
|
session = Session(
|
|
project_id=sample_project.id,
|
|
session_type="startup",
|
|
working_directory="/home/user/test-project",
|
|
git_branch="main",
|
|
environment={"user": "testuser", "pwd": "/home/user/test-project"}
|
|
)
|
|
test_db.add(session)
|
|
await test_db.commit()
|
|
await test_db.refresh(session)
|
|
return session
|
|
|
|
@pytest.fixture
|
|
async def sample_conversation(test_db: AsyncSession, sample_session: Session) -> Conversation:
|
|
"""Create a sample conversation for testing."""
|
|
conversation = Conversation(
|
|
session_id=sample_session.id,
|
|
user_prompt="How do I implement a feature?",
|
|
claude_response="You can implement it by following these steps...",
|
|
tools_used=["Edit", "Write"],
|
|
files_affected=["main.py", "utils.py"],
|
|
exchange_type="user_prompt"
|
|
)
|
|
test_db.add(conversation)
|
|
await test_db.commit()
|
|
await test_db.refresh(conversation)
|
|
return conversation
|
|
|
|
@pytest.fixture
|
|
async def sample_activity(test_db: AsyncSession, sample_session: Session) -> Activity:
|
|
"""Create a sample activity for testing."""
|
|
activity = Activity(
|
|
session_id=sample_session.id,
|
|
tool_name="Edit",
|
|
action="file_edit",
|
|
file_path="/home/user/test-project/main.py",
|
|
metadata={"lines_changed": 10},
|
|
success=True,
|
|
lines_added=5,
|
|
lines_removed=2
|
|
)
|
|
test_db.add(activity)
|
|
await test_db.commit()
|
|
await test_db.refresh(activity)
|
|
return activity
|
|
|
|
@pytest.fixture
|
|
async def sample_waiting_period(test_db: AsyncSession, sample_session: Session) -> WaitingPeriod:
|
|
"""Create a sample waiting period for testing."""
|
|
waiting_period = WaitingPeriod(
|
|
session_id=sample_session.id,
|
|
duration_seconds=30,
|
|
context_before="Claude finished responding",
|
|
context_after="User asked a follow-up question",
|
|
likely_activity="thinking"
|
|
)
|
|
test_db.add(waiting_period)
|
|
await test_db.commit()
|
|
await test_db.refresh(waiting_period)
|
|
return waiting_period
|
|
|
|
@pytest.fixture
|
|
async def sample_git_operation(test_db: AsyncSession, sample_session: Session) -> GitOperation:
|
|
"""Create a sample git operation for testing."""
|
|
git_operation = GitOperation(
|
|
session_id=sample_session.id,
|
|
operation="commit",
|
|
command="git commit -m 'Add new feature'",
|
|
result="[main 123abc] Add new feature",
|
|
success=True,
|
|
files_changed=["main.py", "utils.py"],
|
|
lines_added=15,
|
|
lines_removed=3,
|
|
commit_hash="123abc456def"
|
|
)
|
|
test_db.add(git_operation)
|
|
await test_db.commit()
|
|
await test_db.refresh(git_operation)
|
|
return git_operation
|
|
|
|
# Faker fixtures for generating test data
|
|
@pytest.fixture
|
|
def fake():
|
|
"""Faker instance for generating test data."""
|
|
from faker import Faker
|
|
return Faker()
|
|
|
|
@pytest.fixture
|
|
def project_factory(fake):
|
|
"""Factory for creating project test data."""
|
|
def _create_project_data(**overrides):
|
|
data = {
|
|
"name": fake.company(),
|
|
"path": fake.file_path(depth=3),
|
|
"git_repo": fake.url(),
|
|
"languages": fake.random_elements(
|
|
elements=["python", "javascript", "typescript", "go", "rust"],
|
|
length=fake.random_int(min=1, max=3),
|
|
unique=True
|
|
)
|
|
}
|
|
data.update(overrides)
|
|
return data
|
|
return _create_project_data
|
|
|
|
@pytest.fixture
|
|
def session_factory(fake):
|
|
"""Factory for creating session test data."""
|
|
def _create_session_data(**overrides):
|
|
data = {
|
|
"session_type": fake.random_element(elements=["startup", "resume", "clear"]),
|
|
"working_directory": fake.file_path(depth=3),
|
|
"git_branch": fake.word(),
|
|
"environment": {
|
|
"user": fake.user_name(),
|
|
"pwd": fake.file_path(depth=3),
|
|
"timestamp": fake.iso8601()
|
|
}
|
|
}
|
|
data.update(overrides)
|
|
return data
|
|
return _create_session_data
|
|
|
|
@pytest.fixture
|
|
def conversation_factory(fake):
|
|
"""Factory for creating conversation test data."""
|
|
def _create_conversation_data(**overrides):
|
|
data = {
|
|
"user_prompt": fake.sentence(nb_words=10),
|
|
"claude_response": fake.paragraph(nb_sentences=3),
|
|
"tools_used": fake.random_elements(
|
|
elements=["Edit", "Write", "Read", "Bash", "Grep"],
|
|
length=fake.random_int(min=1, max=3),
|
|
unique=True
|
|
),
|
|
"files_affected": [fake.file_path() for _ in range(fake.random_int(min=0, max=3))],
|
|
"exchange_type": fake.random_element(elements=["user_prompt", "claude_response"])
|
|
}
|
|
data.update(overrides)
|
|
return data
|
|
return _create_conversation_data
|
|
|
|
# Utility functions for tests
|
|
@pytest.fixture
|
|
def assert_response():
|
|
"""Helper for asserting API response structure."""
|
|
def _assert_response(response, status_code=200, required_keys=None):
|
|
assert response.status_code == status_code
|
|
if required_keys:
|
|
data = response.json()
|
|
for key in required_keys:
|
|
assert key in data
|
|
return response.json()
|
|
return _assert_response
|
|
|
|
@pytest.fixture
|
|
def create_test_data():
|
|
"""Helper for creating test data in database."""
|
|
async def _create_test_data(db: AsyncSession, model_class, count=1, **kwargs):
|
|
items = []
|
|
for i in range(count):
|
|
item = model_class(**kwargs)
|
|
db.add(item)
|
|
items.append(item)
|
|
await db.commit()
|
|
for item in items:
|
|
await db.refresh(item)
|
|
return items[0] if count == 1 else items
|
|
return _create_test_data |