claude/claude-code-tracker
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Initial commit: Claude Code Project Tracker

Browse source 
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>
This commit is contained in:
Ryan Malloy 2025-08-11 02:59:21 -06:00
commit 44ed9936b7
48 changed files with 9707 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

256
tests/conftest.py Normal file
View file

@ -0,0 +1,256 @@
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

413
tests/fixtures.py Normal file
View file

@ -0,0 +1,413 @@
"""
Test fixtures and sample data for the Claude Code Project Tracker.
"""
from datetime import datetime, timedelta
from typing import Dict, List, Any
from faker import Faker
fake = Faker()
class TestDataFactory:
"""Factory for creating realistic test data."""
@staticmethod
def create_project_data(**overrides) -> Dict[str, Any]:
"""Create sample project data."""
data = {
"name": fake.company(),
"path": fake.file_path(depth=3, extension=""),
"git_repo": fake.url().replace("http://", "https://github.com/"),
"languages": fake.random_elements(
elements=["python", "javascript", "typescript", "go", "rust", "java", "cpp"],
length=fake.random_int(min=1, max=4),
unique=True
),
"created_at": fake.date_time_between(start_date="-1y", end_date="now"),
"last_session": fake.date_time_between(start_date="-30d", end_date="now"),
"total_sessions": fake.random_int(min=1, max=50),
"total_time_minutes": fake.random_int(min=30, max=2000),
"files_modified_count": fake.random_int(min=5, max=100),
"lines_changed_count": fake.random_int(min=100, max=5000)
}
data.update(overrides)
return data
@staticmethod
def create_session_data(project_id: int = 1, **overrides) -> Dict[str, Any]:
"""Create sample session data."""
start_time = fake.date_time_between(start_date="-7d", end_date="now")
duration = fake.random_int(min=5, max=180) # 5 minutes to 3 hours
data = {
"project_id": project_id,
"start_time": start_time,
"end_time": start_time + timedelta(minutes=duration),
"session_type": fake.random_element(elements=["startup", "resume", "clear"]),
"working_directory": fake.file_path(depth=3, extension=""),
"git_branch": fake.random_element(elements=["main", "develop", "feature/new-feature", "bugfix/issue-123"]),
"environment": {
"user": fake.user_name(),
"pwd": fake.file_path(depth=3, extension=""),
"python_version": "3.11.0",
"node_version": "18.17.0"
},
"duration_minutes": duration,
"activity_count": fake.random_int(min=3, max=50),
"conversation_count": fake.random_int(min=2, max=30),
"files_touched": [fake.file_path() for _ in range(fake.random_int(min=1, max=8))]
}
data.update(overrides)
return data
@staticmethod
def create_conversation_data(session_id: int = 1, **overrides) -> Dict[str, Any]:
"""Create sample conversation data."""
prompts = [
"How do I implement user authentication?",
"Can you help me debug this error?",
"What's the best way to structure this code?",
"Help me optimize this function for performance",
"How do I add tests for this component?",
"Can you review this code for best practices?",
"What libraries should I use for this feature?",
"How do I handle errors in this async function?"
]
responses = [
"I can help you implement user authentication. Here's a comprehensive approach...",
"Let me analyze this error and provide a solution...",
"For code structure, I recommend following these patterns...",
"Here are several optimization strategies for your function...",
"Let's add comprehensive tests for this component...",
"I've reviewed your code and here are my recommendations...",
"For this feature, I suggest these well-maintained libraries...",
"Here's how to properly handle errors in async functions..."
]
data = {
"session_id": session_id,
"timestamp": fake.date_time_between(start_date="-7d", end_date="now"),
"user_prompt": fake.random_element(elements=prompts),
"claude_response": fake.random_element(elements=responses),
"tools_used": fake.random_elements(
elements=["Edit", "Write", "Read", "Bash", "Grep", "Glob", "Task"],
length=fake.random_int(min=1, max=4),
unique=True
),
"files_affected": [
fake.file_path(extension=ext)
for ext in fake.random_elements(
elements=["py", "js", "ts", "go", "rs", "java", "cpp", "md"],
length=fake.random_int(min=0, max=3),
unique=True
)
],
"context": {
"intent": fake.random_element(elements=["debugging", "implementation", "learning", "optimization"]),
"complexity": fake.random_element(elements=["low", "medium", "high"])
},
"tokens_input": fake.random_int(min=50, max=500),
"tokens_output": fake.random_int(min=100, max=1000),
"exchange_type": fake.random_element(elements=["user_prompt", "claude_response"])
}
data.update(overrides)
return data
@staticmethod
def create_activity_data(session_id: int = 1, **overrides) -> Dict[str, Any]:
"""Create sample activity data."""
tools = {
"Edit": {
"action": "file_edit",
"metadata": {"lines_changed": fake.random_int(min=1, max=50)},
"lines_added": fake.random_int(min=0, max=30),
"lines_removed": fake.random_int(min=0, max=20)
},
"Write": {
"action": "file_write",
"metadata": {"new_file": fake.boolean()},
"lines_added": fake.random_int(min=10, max=100),
"lines_removed": 0
},
"Read": {
"action": "file_read",
"metadata": {"file_size": fake.random_int(min=100, max=5000)},
"lines_added": 0,
"lines_removed": 0
},
"Bash": {
"action": "command_execution",
"metadata": {
"command": fake.random_element(elements=[
"npm install",
"pytest",
"git status",
"python main.py",
"docker build ."
]),
"exit_code": 0
},
"lines_added": 0,
"lines_removed": 0
},
"Grep": {
"action": "search",
"metadata": {
"pattern": fake.word(),
"matches_found": fake.random_int(min=0, max=20)
},
"lines_added": 0,
"lines_removed": 0
}
}
tool_name = fake.random_element(elements=list(tools.keys()))
tool_config = tools[tool_name]
data = {
"session_id": session_id,
"timestamp": fake.date_time_between(start_date="-7d", end_date="now"),
"tool_name": tool_name,
"action": tool_config["action"],
"file_path": fake.file_path() if tool_name in ["Edit", "Write", "Read"] else None,
"metadata": tool_config["metadata"],
"success": fake.boolean(chance_of_getting_true=90),
"error_message": None if fake.boolean(chance_of_getting_true=90) else "Operation failed",
"lines_added": tool_config.get("lines_added", 0),
"lines_removed": tool_config.get("lines_removed", 0)
}
data.update(overrides)
return data
@staticmethod
def create_waiting_period_data(session_id: int = 1, **overrides) -> Dict[str, Any]:
"""Create sample waiting period data."""
start_time = fake.date_time_between(start_date="-7d", end_date="now")
duration = fake.random_int(min=5, max=300) # 5 seconds to 5 minutes
activities = {
"thinking": "User is contemplating the response",
"research": "User is looking up documentation",
"external_work": "User is working in another application",
"break": "User stepped away from the computer"
}
likely_activity = fake.random_element(elements=list(activities.keys()))
data = {
"session_id": session_id,
"start_time": start_time,
"end_time": start_time + timedelta(seconds=duration),
"duration_seconds": duration,
"context_before": "Claude finished providing a detailed explanation",
"context_after": fake.random_element(elements=[
"User asked a follow-up question",
"User requested clarification",
"User provided additional context",
"User asked about a different topic"
]),
"likely_activity": likely_activity
}
data.update(overrides)
return data
@staticmethod
def create_git_operation_data(session_id: int = 1, **overrides) -> Dict[str, Any]:
"""Create sample git operation data."""
operations = {
"commit": {
"command": "git commit -m 'Add new feature'",
"result": "[main abc123] Add new feature\\n 2 files changed, 15 insertions(+), 3 deletions(-)",
"files_changed": ["main.py", "utils.py"],
"lines_added": 15,
"lines_removed": 3,
"commit_hash": fake.sha1()[:8]
},
"push": {
"command": "git push origin main",
"result": "To https://github.com/user/repo.git\\n abc123..def456 main -> main",
"files_changed": [],
"lines_added": 0,
"lines_removed": 0,
"commit_hash": None
},
"pull": {
"command": "git pull origin main",
"result": "Already up to date.",
"files_changed": [],
"lines_added": 0,
"lines_removed": 0,
"commit_hash": None
},
"branch": {
"command": "git checkout -b feature/new-feature",
"result": "Switched to a new branch 'feature/new-feature'",
"files_changed": [],
"lines_added": 0,
"lines_removed": 0,
"commit_hash": None
}
}
operation = fake.random_element(elements=list(operations.keys()))
op_config = operations[operation]
data = {
"session_id": session_id,
"timestamp": fake.date_time_between(start_date="-7d", end_date="now"),
"operation": operation,
"command": op_config["command"],
"result": op_config["result"],
"success": fake.boolean(chance_of_getting_true=95),
"files_changed": op_config["files_changed"],
"lines_added": op_config["lines_added"],
"lines_removed": op_config["lines_removed"],
"commit_hash": op_config["commit_hash"],
"branch_from": fake.random_element(elements=["main", "develop", "feature/old-feature"]) if operation == "merge" else None,
"branch_to": fake.random_element(elements=["main", "develop"]) if operation == "merge" else None
}
data.update(overrides)
return data
class SampleDataSets:
"""Pre-defined sample data sets for different test scenarios."""
@staticmethod
def productive_session() -> Dict[str, List[Dict[str, Any]]]:
"""Data for a highly productive development session."""
project_data = TestDataFactory.create_project_data(
name="E-commerce Platform",
languages=["python", "javascript", "typescript"],
total_sessions=25,
total_time_minutes=800
)
session_data = TestDataFactory.create_session_data(
session_type="startup",
duration_minutes=120,
activity_count=45,
conversation_count=12
)
conversations = [
TestDataFactory.create_conversation_data(
user_prompt="How do I implement user authentication with JWT?",
tools_used=["Edit", "Write"],
files_affected=["auth.py", "models.py"]
),
TestDataFactory.create_conversation_data(
user_prompt="Can you help me optimize this database query?",
tools_used=["Edit", "Read"],
files_affected=["queries.py"]
),
TestDataFactory.create_conversation_data(
user_prompt="What's the best way to handle async errors?",
tools_used=["Edit"],
files_affected=["error_handler.py"]
)
]
activities = [
TestDataFactory.create_activity_data(tool_name="Edit", lines_added=25, lines_removed=5),
TestDataFactory.create_activity_data(tool_name="Write", lines_added=50, lines_removed=0),
TestDataFactory.create_activity_data(tool_name="Bash", metadata={"command": "pytest"}),
TestDataFactory.create_activity_data(tool_name="Read", file_path="/docs/api.md")
]
return {
"project": project_data,
"session": session_data,
"conversations": conversations,
"activities": activities
}
@staticmethod
def debugging_session() -> Dict[str, List[Dict[str, Any]]]:
"""Data for a debugging-focused session with lots of investigation."""
project_data = TestDataFactory.create_project_data(
name="Bug Tracking System",
languages=["go", "typescript"]
)
session_data = TestDataFactory.create_session_data(
session_type="resume",
duration_minutes=90,
activity_count=35,
conversation_count=8
)
conversations = [
TestDataFactory.create_conversation_data(
user_prompt="I'm getting a panic in my Go application, can you help debug?",
tools_used=["Read", "Grep"],
files_affected=["main.go", "handler.go"],
context={"intent": "debugging", "complexity": "high"}
),
TestDataFactory.create_conversation_data(
user_prompt="The tests are failing intermittently, what could be wrong?",
tools_used=["Read", "Bash"],
files_affected=["test_handler.go"]
)
]
# Lots of read operations for debugging
activities = [
TestDataFactory.create_activity_data(tool_name="Read") for _ in range(8)
] + [
TestDataFactory.create_activity_data(tool_name="Grep", metadata={"pattern": "error", "matches_found": 12}),
TestDataFactory.create_activity_data(tool_name="Bash", metadata={"command": "go test -v"}),
TestDataFactory.create_activity_data(tool_name="Edit", lines_added=3, lines_removed=1)
]
# Longer thinking periods during debugging
waiting_periods = [
TestDataFactory.create_waiting_period_data(
duration_seconds=180,
likely_activity="research",
context_after="User asked about error patterns"
),
TestDataFactory.create_waiting_period_data(
duration_seconds=120,
likely_activity="thinking",
context_after="User provided stack trace"
)
]
return {
"project": project_data,
"session": session_data,
"conversations": conversations,
"activities": activities,
"waiting_periods": waiting_periods
}
@staticmethod
def learning_session() -> Dict[str, List[Dict[str, Any]]]:
"""Data for a learning-focused session with lots of questions."""
project_data = TestDataFactory.create_project_data(
name="Learning Rust",
languages=["rust"],
total_sessions=5,
total_time_minutes=200
)
conversations = [
TestDataFactory.create_conversation_data(
user_prompt="What's the difference between String and &str in Rust?",
context={"intent": "learning", "complexity": "medium"}
),
TestDataFactory.create_conversation_data(
user_prompt="How do I handle ownership and borrowing correctly?",
context={"intent": "learning", "complexity": "high"}
),
TestDataFactory.create_conversation_data(
user_prompt="Can you explain Rust's trait system?",
context={"intent": "learning", "complexity": "high"}
)
]
return {
"project": project_data,
"conversations": conversations
}

525
tests/test_api.py Normal file
View file

@ -0,0 +1,525 @@
"""
Tests for the Claude Code Project Tracker API endpoints.
"""
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.project import Project
from app.models.session import Session
from tests.fixtures import TestDataFactory
class TestSessionAPI:
"""Test session management endpoints."""
@pytest.mark.api
async def test_start_session_startup(self, test_client: AsyncClient):
"""Test starting a new session with startup type."""
session_data = TestDataFactory.create_session_data(
session_type="startup",
working_directory="/home/user/test-project",
git_branch="main"
)
response = await test_client.post("/api/session/start", json=session_data)
assert response.status_code == 201
data = response.json()
assert "session_id" in data
assert "project_id" in data
assert data["status"] == "started"
@pytest.mark.api
async def test_start_session_creates_project(self, test_client: AsyncClient):
"""Test that starting a session creates a project if it doesn't exist."""
session_data = {
"session_type": "startup",
"working_directory": "/home/user/new-project",
"git_repo": "https://github.com/user/new-project",
"environment": {"user": "testuser"}
}
response = await test_client.post("/api/session/start", json=session_data)
assert response.status_code == 201
data = response.json()
assert data["project_id"] is not None
@pytest.mark.api
async def test_start_session_resume(self, test_client: AsyncClient):
"""Test resuming an existing session."""
session_data = TestDataFactory.create_session_data(session_type="resume")
response = await test_client.post("/api/session/start", json=session_data)
assert response.status_code == 201
data = response.json()
assert "session_id" in data
@pytest.mark.api
async def test_end_session(self, test_client: AsyncClient, sample_session: Session):
"""Test ending an active session."""
end_data = {
"session_id": sample_session.id,
"end_reason": "normal"
}
response = await test_client.post("/api/session/end", json=end_data)
assert response.status_code == 200
data = response.json()
assert data["session_id"] == sample_session.id
assert data["status"] == "ended"
@pytest.mark.api
async def test_end_nonexistent_session(self, test_client: AsyncClient):
"""Test ending a session that doesn't exist."""
end_data = {
"session_id": 99999,
"end_reason": "normal"
}
response = await test_client.post("/api/session/end", json=end_data)
assert response.status_code == 404
class TestConversationAPI:
"""Test conversation tracking endpoints."""
@pytest.mark.api
async def test_log_conversation(self, test_client: AsyncClient, sample_session: Session):
"""Test logging a conversation exchange."""
conversation_data = TestDataFactory.create_conversation_data(
session_id=sample_session.id,
user_prompt="How do I implement authentication?",
claude_response="You can implement authentication using...",
tools_used=["Edit", "Write"]
)
response = await test_client.post("/api/conversation", json=conversation_data)
assert response.status_code == 201
@pytest.mark.api
async def test_log_conversation_user_prompt_only(self, test_client: AsyncClient, sample_session: Session):
"""Test logging just a user prompt."""
conversation_data = {
"session_id": sample_session.id,
"timestamp": "2024-01-01T12:00:00Z",
"user_prompt": "How do I fix this error?",
"exchange_type": "user_prompt"
}
response = await test_client.post("/api/conversation", json=conversation_data)
assert response.status_code == 201
@pytest.mark.api
async def test_log_conversation_invalid_session(self, test_client: AsyncClient):
"""Test logging conversation with invalid session ID."""
conversation_data = TestDataFactory.create_conversation_data(session_id=99999)
response = await test_client.post("/api/conversation", json=conversation_data)
assert response.status_code == 404
class TestActivityAPI:
"""Test activity tracking endpoints."""
@pytest.mark.api
async def test_record_activity_edit(self, test_client: AsyncClient, sample_session: Session):
"""Test recording an Edit tool activity."""
activity_data = TestDataFactory.create_activity_data(
session_id=sample_session.id,
tool_name="Edit",
action="file_edit",
file_path="/home/user/test.py",
lines_added=10,
lines_removed=3
)
response = await test_client.post("/api/activity", json=activity_data)
assert response.status_code == 201
@pytest.mark.api
async def test_record_activity_bash(self, test_client: AsyncClient, sample_session: Session):
"""Test recording a Bash command activity."""
activity_data = TestDataFactory.create_activity_data(
session_id=sample_session.id,
tool_name="Bash",
action="command_execution",
metadata={"command": "pytest", "exit_code": 0}
)
response = await test_client.post("/api/activity", json=activity_data)
assert response.status_code == 201
@pytest.mark.api
async def test_record_activity_failed(self, test_client: AsyncClient, sample_session: Session):
"""Test recording a failed activity."""
activity_data = TestDataFactory.create_activity_data(
session_id=sample_session.id,
success=False,
error_message="Permission denied"
)
response = await test_client.post("/api/activity", json=activity_data)
assert response.status_code == 201
class TestWaitingAPI:
"""Test waiting period tracking endpoints."""
@pytest.mark.api
async def test_start_waiting_period(self, test_client: AsyncClient, sample_session: Session):
"""Test starting a waiting period."""
waiting_data = {
"session_id": sample_session.id,
"timestamp": "2024-01-01T12:00:00Z",
"context_before": "Claude finished responding"
}
response = await test_client.post("/api/waiting/start", json=waiting_data)
assert response.status_code == 201
@pytest.mark.api
async def test_end_waiting_period(self, test_client: AsyncClient, sample_session: Session):
"""Test ending a waiting period."""
# First start a waiting period
start_data = {
"session_id": sample_session.id,
"timestamp": "2024-01-01T12:00:00Z",
"context_before": "Claude finished responding"
}
await test_client.post("/api/waiting/start", json=start_data)
# Then end it
end_data = {
"session_id": sample_session.id,
"timestamp": "2024-01-01T12:05:00Z",
"duration_seconds": 300,
"context_after": "User asked follow-up question"
}
response = await test_client.post("/api/waiting/end", json=end_data)
assert response.status_code == 200
class TestGitAPI:
"""Test git operation tracking endpoints."""
@pytest.mark.api
async def test_record_git_commit(self, test_client: AsyncClient, sample_session: Session):
"""Test recording a git commit operation."""
git_data = TestDataFactory.create_git_operation_data(
session_id=sample_session.id,
operation="commit",
command="git commit -m 'Add feature'",
files_changed=["main.py", "utils.py"],
lines_added=20,
lines_removed=5
)
response = await test_client.post("/api/git", json=git_data)
assert response.status_code == 201
@pytest.mark.api
async def test_record_git_push(self, test_client: AsyncClient, sample_session: Session):
"""Test recording a git push operation."""
git_data = TestDataFactory.create_git_operation_data(
session_id=sample_session.id,
operation="push",
command="git push origin main"
)
response = await test_client.post("/api/git", json=git_data)
assert response.status_code == 201
class TestProjectAPI:
"""Test project data retrieval endpoints."""
@pytest.mark.api
async def test_list_projects(self, test_client: AsyncClient, sample_project: Project):
"""Test listing all projects."""
response = await test_client.get("/api/projects")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) >= 1
project = data[0]
assert "id" in project
assert "name" in project
assert "path" in project
@pytest.mark.api
async def test_list_projects_with_pagination(self, test_client: AsyncClient):
"""Test project listing with pagination parameters."""
response = await test_client.get("/api/projects?limit=5&offset=0")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) <= 5
@pytest.mark.api
async def test_get_project_timeline(self, test_client: AsyncClient, sample_project: Project):
"""Test getting detailed project timeline."""
response = await test_client.get(f"/api/projects/{sample_project.id}/timeline")
assert response.status_code == 200
data = response.json()
assert "project" in data
assert "timeline" in data
assert isinstance(data["timeline"], list)
@pytest.mark.api
async def test_get_nonexistent_project_timeline(self, test_client: AsyncClient):
"""Test getting timeline for non-existent project."""
response = await test_client.get("/api/projects/99999/timeline")
assert response.status_code == 404
class TestAnalyticsAPI:
"""Test analytics and insights endpoints."""
@pytest.mark.api
async def test_get_productivity_metrics(self, test_client: AsyncClient, sample_project: Project):
"""Test getting productivity analytics."""
response = await test_client.get("/api/analytics/productivity")
assert response.status_code == 200
data = response.json()
assert "engagement_score" in data
assert "average_session_length" in data
assert "think_time_average" in data
@pytest.mark.api
async def test_get_productivity_metrics_for_project(self, test_client: AsyncClient, sample_project: Project):
"""Test getting productivity analytics for specific project."""
response = await test_client.get(f"/api/analytics/productivity?project_id={sample_project.id}")
assert response.status_code == 200
data = response.json()
assert isinstance(data, dict)
@pytest.mark.api
async def test_search_conversations(self, test_client: AsyncClient, sample_conversation):
"""Test searching through conversations."""
response = await test_client.get("/api/conversations/search?query=implement feature")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
if data: # If there are results
result = data[0]
assert "id" in result
assert "user_prompt" in result
assert "relevance_score" in result
@pytest.mark.api
async def test_search_conversations_with_project_filter(self, test_client: AsyncClient, sample_project: Project):
"""Test searching conversations within a specific project."""
response = await test_client.get(f"/api/conversations/search?query=debug&project_id={sample_project.id}")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
class TestErrorHandling:
"""Test error handling and edge cases."""
@pytest.mark.api
async def test_malformed_json(self, test_client: AsyncClient):
"""Test handling of malformed JSON requests."""
response = await test_client.post(
"/api/session/start",
content="{'invalid': json}",
headers={"Content-Type": "application/json"}
)
assert response.status_code == 422
@pytest.mark.api
async def test_missing_required_fields(self, test_client: AsyncClient):
"""Test handling of requests with missing required fields."""
response = await test_client.post("/api/session/start", json={})
assert response.status_code == 422
data = response.json()
assert "detail" in data
@pytest.mark.api
async def test_invalid_session_id_type(self, test_client: AsyncClient):
"""Test handling of invalid data types."""
conversation_data = {
"session_id": "invalid", # Should be int
"user_prompt": "Test message",
"exchange_type": "user_prompt"
}
response = await test_client.post("/api/conversation", json=conversation_data)
assert response.status_code == 422
@pytest.mark.api
async def test_nonexistent_endpoint(self, test_client: AsyncClient):
"""Test accessing non-existent endpoint."""
response = await test_client.get("/api/nonexistent")
assert response.status_code == 404
class TestIntegrationScenarios:
"""Test complete workflow scenarios."""
@pytest.mark.integration
async def test_complete_session_workflow(self, test_client: AsyncClient):
"""Test a complete session from start to finish."""
# Start session
session_data = TestDataFactory.create_session_data(
working_directory="/home/user/integration-test"
)
start_response = await test_client.post("/api/session/start", json=session_data)
assert start_response.status_code == 201
session_info = start_response.json()
session_id = session_info["session_id"]
# Log user prompt
prompt_data = {
"session_id": session_id,
"timestamp": "2024-01-01T12:00:00Z",
"user_prompt": "Help me implement a REST API",
"exchange_type": "user_prompt"
}
prompt_response = await test_client.post("/api/conversation", json=prompt_data)
assert prompt_response.status_code == 201
# Start waiting period
waiting_start = {
"session_id": session_id,
"timestamp": "2024-01-01T12:00:01Z",
"context_before": "Claude is processing the request"
}
waiting_response = await test_client.post("/api/waiting/start", json=waiting_start)
assert waiting_response.status_code == 201
# Record some activities
activities = [
{
"session_id": session_id,
"tool_name": "Write",
"action": "file_write",
"file_path": "/home/user/integration-test/api.py",
"timestamp": "2024-01-01T12:01:00Z",
"success": True,
"lines_added": 50
},
{
"session_id": session_id,
"tool_name": "Edit",
"action": "file_edit",
"file_path": "/home/user/integration-test/main.py",
"timestamp": "2024-01-01T12:02:00Z",
"success": True,
"lines_added": 10,
"lines_removed": 2
}
]
for activity in activities:
activity_response = await test_client.post("/api/activity", json=activity)
assert activity_response.status_code == 201
# End waiting period
waiting_end = {
"session_id": session_id,
"timestamp": "2024-01-01T12:05:00Z",
"duration_seconds": 300,
"context_after": "User reviewed the implementation"
}
waiting_end_response = await test_client.post("/api/waiting/end", json=waiting_end)
assert waiting_end_response.status_code == 200
# Record git commit
git_data = {
"session_id": session_id,
"operation": "commit",
"command": "git commit -m 'Implement REST API'",
"timestamp": "2024-01-01T12:06:00Z",
"result": "[main abc123] Implement REST API",
"success": True,
"files_changed": ["api.py", "main.py"],
"lines_added": 60,
"lines_removed": 2,
"commit_hash": "abc12345"
}
git_response = await test_client.post("/api/git", json=git_data)
assert git_response.status_code == 201
# End session
end_data = {
"session_id": session_id,
"end_reason": "normal"
}
end_response = await test_client.post("/api/session/end", json=end_data)
assert end_response.status_code == 200
# Verify project was created and populated
projects_response = await test_client.get("/api/projects")
assert projects_response.status_code == 200
projects = projects_response.json()
# Find our project
test_project = None
for project in projects:
if project["path"] == "/home/user/integration-test":
test_project = project
break
assert test_project is not None
assert test_project["total_sessions"] >= 1
# Get project timeline
timeline_response = await test_client.get(f"/api/projects/{test_project['id']}/timeline")
assert timeline_response.status_code == 200
timeline = timeline_response.json()
assert len(timeline["timeline"]) > 0 # Should have recorded events
@pytest.mark.integration
@pytest.mark.slow
async def test_analytics_with_sample_data(self, test_client: AsyncClient, test_db: AsyncSession):
"""Test analytics calculations with realistic sample data."""
# This test would create a full dataset and verify analytics work correctly
# Marked as slow since it involves more data processing
productivity_response = await test_client.get("/api/analytics/productivity?days=7")
assert productivity_response.status_code == 200
metrics = productivity_response.json()
assert "engagement_score" in metrics
assert isinstance(metrics["engagement_score"], (int, float))
assert 0 <= metrics["engagement_score"] <= 100

469
tests/test_hooks.py Normal file
View file

@ -0,0 +1,469 @@
"""
Tests for Claude Code hook simulation and integration.
"""
import pytest
import json
import subprocess
from unittest.mock import patch, MagicMock
from httpx import AsyncClient
from tests.fixtures import TestDataFactory
class TestHookSimulation:
"""Test hook payload generation and API integration."""
def test_session_start_hook_payload(self):
"""Test generating SessionStart hook payload."""
# Simulate environment variables that would be set by Claude Code
mock_env = {
"PWD": "/home/user/test-project",
"USER": "testuser"
}
with patch.dict("os.environ", mock_env):
with patch("subprocess.check_output") as mock_git:
# Mock git commands
mock_git.side_effect = [
b"main\n", # git branch --show-current
b"https://github.com/user/test-project.git\n" # git config --get remote.origin.url
]
payload = {
"session_type": "startup",
"working_directory": mock_env["PWD"],
"git_branch": "main",
"git_repo": "https://github.com/user/test-project.git",
"environment": {
"pwd": mock_env["PWD"],
"user": mock_env["USER"],
"timestamp": "2024-01-01T12:00:00Z"
}
}
# Verify payload structure matches API expectations
assert "session_type" in payload
assert "working_directory" in payload
assert payload["session_type"] in ["startup", "resume", "clear"]
def test_user_prompt_hook_payload(self):
"""Test generating UserPromptSubmit hook payload."""
mock_prompt = "How do I implement user authentication?"
mock_session_id = "123"
payload = {
"session_id": int(mock_session_id),
"timestamp": "2024-01-01T12:00:00Z",
"user_prompt": mock_prompt,
"exchange_type": "user_prompt"
}
assert payload["session_id"] == 123
assert payload["user_prompt"] == mock_prompt
assert payload["exchange_type"] == "user_prompt"
def test_post_tool_use_edit_payload(self):
"""Test generating PostToolUse Edit hook payload."""
mock_file_path = "/home/user/test-project/main.py"
mock_session_id = "123"
payload = {
"session_id": int(mock_session_id),
"tool_name": "Edit",
"action": "file_edit",
"file_path": mock_file_path,
"timestamp": "2024-01-01T12:00:00Z",
"metadata": {"success": True},
"success": True
}
assert payload["tool_name"] == "Edit"
assert payload["file_path"] == mock_file_path
assert payload["success"] is True
def test_post_tool_use_bash_payload(self):
"""Test generating PostToolUse Bash hook payload."""
mock_command = "pytest --cov=app"
mock_session_id = "123"
payload = {
"session_id": int(mock_session_id),
"tool_name": "Bash",
"action": "command_execution",
"timestamp": "2024-01-01T12:00:00Z",
"metadata": {
"command": mock_command,
"success": True
},
"success": True
}
assert payload["tool_name"] == "Bash"
assert payload["metadata"]["command"] == mock_command
def test_notification_hook_payload(self):
"""Test generating Notification hook payload."""
mock_session_id = "123"
payload = {
"session_id": int(mock_session_id),
"timestamp": "2024-01-01T12:00:00Z",
"context_before": "Claude is waiting for input"
}
assert payload["session_id"] == 123
assert "context_before" in payload
def test_stop_hook_payload(self):
"""Test generating Stop hook payload."""
mock_session_id = "123"
payload = {
"session_id": int(mock_session_id),
"timestamp": "2024-01-01T12:00:00Z",
"claude_response": "Response completed",
"exchange_type": "claude_response"
}
assert payload["exchange_type"] == "claude_response"
def test_json_escaping_in_payloads(self):
"""Test that special characters in payloads are properly escaped."""
# Test prompt with quotes and newlines
problematic_prompt = 'How do I handle "quotes" and\nnewlines in JSON?'
payload = {
"session_id": 1,
"user_prompt": problematic_prompt,
"exchange_type": "user_prompt"
}
# Should be able to serialize to JSON without errors
json_str = json.dumps(payload)
parsed = json.loads(json_str)
assert parsed["user_prompt"] == problematic_prompt
def test_file_path_escaping(self):
"""Test that file paths with spaces are handled correctly."""
file_path_with_spaces = "/home/user/My Projects/test project/main.py"
payload = {
"session_id": 1,
"tool_name": "Edit",
"file_path": file_path_with_spaces
}
json_str = json.dumps(payload)
parsed = json.loads(json_str)
assert parsed["file_path"] == file_path_with_spaces
class TestHookIntegration:
"""Test actual hook integration with the API."""
@pytest.mark.hooks
async def test_session_start_hook_integration(self, test_client: AsyncClient):
"""Test complete SessionStart hook workflow."""
# Simulate the payload that would be sent by the hook
hook_payload = TestDataFactory.create_session_data(
session_type="startup",
working_directory="/home/user/hook-test",
git_branch="main",
git_repo="https://github.com/user/hook-test.git"
)
response = await test_client.post("/api/session/start", json=hook_payload)
assert response.status_code == 201
data = response.json()
assert "session_id" in data
assert "project_id" in data
# Store session ID for subsequent hook calls
return data["session_id"]
@pytest.mark.hooks
async def test_user_prompt_hook_integration(self, test_client: AsyncClient):
"""Test UserPromptSubmit hook integration."""
# First create a session
session_id = await self.test_session_start_hook_integration(test_client)
# Simulate user prompt hook
hook_payload = {
"session_id": session_id,
"timestamp": "2024-01-01T12:00:00Z",
"user_prompt": "This prompt came from a hook",
"exchange_type": "user_prompt"
}
response = await test_client.post("/api/conversation", json=hook_payload)
assert response.status_code == 201
@pytest.mark.hooks
async def test_post_tool_use_hook_integration(self, test_client: AsyncClient):
"""Test PostToolUse hook integration."""
session_id = await self.test_session_start_hook_integration(test_client)
# Simulate Edit tool hook
hook_payload = {
"session_id": session_id,
"tool_name": "Edit",
"action": "file_edit",
"file_path": "/home/user/hook-test/main.py",
"timestamp": "2024-01-01T12:01:00Z",
"metadata": {"lines_changed": 5},
"success": True,
"lines_added": 3,
"lines_removed": 2
}
response = await test_client.post("/api/activity", json=hook_payload)
assert response.status_code == 201
@pytest.mark.hooks
async def test_waiting_period_hook_integration(self, test_client: AsyncClient):
"""Test Notification and waiting period hooks."""
session_id = await self.test_session_start_hook_integration(test_client)
# Start waiting period
start_payload = {
"session_id": session_id,
"timestamp": "2024-01-01T12:00:00Z",
"context_before": "Claude finished responding"
}
start_response = await test_client.post("/api/waiting/start", json=start_payload)
assert start_response.status_code == 201
# End waiting period
end_payload = {
"session_id": session_id,
"timestamp": "2024-01-01T12:05:00Z",
"duration_seconds": 300,
"context_after": "User submitted new prompt"
}
end_response = await test_client.post("/api/waiting/end", json=end_payload)
assert end_response.status_code == 200
@pytest.mark.hooks
async def test_stop_hook_integration(self, test_client: AsyncClient):
"""Test Stop hook integration."""
session_id = await self.test_session_start_hook_integration(test_client)
# Simulate Stop hook (Claude response)
hook_payload = {
"session_id": session_id,
"timestamp": "2024-01-01T12:10:00Z",
"claude_response": "Here's the implementation you requested...",
"tools_used": ["Edit", "Write"],
"files_affected": ["main.py", "utils.py"],
"exchange_type": "claude_response"
}
response = await test_client.post("/api/conversation", json=hook_payload)
assert response.status_code == 201
@pytest.mark.hooks
async def test_git_hook_integration(self, test_client: AsyncClient):
"""Test git operation hook integration."""
session_id = await self.test_session_start_hook_integration(test_client)
hook_payload = TestDataFactory.create_git_operation_data(
session_id=session_id,
operation="commit",
command="git commit -m 'Test commit from hook'",
files_changed=["main.py"],
lines_added=10,
lines_removed=2
)
response = await test_client.post("/api/git", json=hook_payload)
assert response.status_code == 201
class TestHookEnvironment:
"""Test hook environment and configuration."""
def test_session_id_persistence(self):
"""Test session ID storage and retrieval mechanism."""
session_file = "/tmp/claude-session-id"
test_session_id = "12345"
# Simulate writing session ID to file (as hooks would do)
with open(session_file, "w") as f:
f.write(test_session_id)
# Simulate reading session ID from file (as subsequent hooks would do)
with open(session_file, "r") as f:
retrieved_id = f.read().strip()
assert retrieved_id == test_session_id
# Clean up
import os
if os.path.exists(session_file):
os.remove(session_file)
def test_git_environment_detection(self):
"""Test git repository detection logic."""
with patch("subprocess.check_output") as mock_subprocess:
# Mock successful git commands
mock_subprocess.side_effect = [
b"main\n", # git branch --show-current
b"https://github.com/user/repo.git\n" # git config --get remote.origin.url
]
# Simulate what hooks would do
try:
branch = subprocess.check_output(
["git", "branch", "--show-current"],
stderr=subprocess.DEVNULL,
text=True
).strip()
repo = subprocess.check_output(
["git", "config", "--get", "remote.origin.url"],
stderr=subprocess.DEVNULL,
text=True
).strip()
assert branch == "main"
assert repo == "https://github.com/user/repo.git"
except subprocess.CalledProcessError:
# If git commands fail, hooks should handle gracefully
branch = "unknown"
repo = "null"
# Test non-git directory handling
with patch("subprocess.check_output", side_effect=subprocess.CalledProcessError(128, "git")):
try:
branch = subprocess.check_output(
["git", "branch", "--show-current"],
stderr=subprocess.DEVNULL,
text=True
).strip()
except subprocess.CalledProcessError:
branch = "unknown"
assert branch == "unknown"
def test_hook_error_handling(self):
"""Test hook error handling and graceful failures."""
# Test network failure (API unreachable)
import requests
from unittest.mock import patch
with patch("requests.post", side_effect=requests.exceptions.ConnectionError("Connection refused")):
# Hooks should not crash if API is unreachable
# This would be handled by curl in the actual hooks with > /dev/null 2>&1
try:
# Simulate what would happen in a hook
response = requests.post("http://localhost:8000/api/session/start", json={})
assert False, "Should have raised ConnectionError"
except requests.exceptions.ConnectionError:
# This is expected - hook should handle gracefully
pass
def test_hook_command_construction(self):
"""Test building hook commands with proper escaping."""
# Test session start hook command construction
session_data = {
"session_type": "startup",
"working_directory": "/home/user/test project", # Space in path
"git_branch": "feature/new-feature",
"environment": {"user": "testuser"}
}
# Construct JSON payload like the hook would
json_payload = json.dumps(session_data)
# Verify it can be parsed back
parsed = json.loads(json_payload)
assert parsed["working_directory"] == "/home/user/test project"
def test_hook_timing_and_ordering(self):
"""Test that hook timing and ordering work correctly."""
# Simulate rapid-fire hook calls (as would happen during active development)
timestamps = [
"2024-01-01T12:00:00Z", # Session start
"2024-01-01T12:00:01Z", # User prompt
"2024-01-01T12:00:02Z", # Waiting start
"2024-01-01T12:00:05Z", # Tool use (Edit)
"2024-01-01T12:00:06Z", # Tool use (Write)
"2024-01-01T12:00:10Z", # Waiting end
"2024-01-01T12:00:11Z", # Claude response
]
# Verify timestamps are in chronological order
for i in range(1, len(timestamps)):
assert timestamps[i] > timestamps[i-1]
class TestHookConfiguration:
"""Test hook configuration validation and setup."""
def test_hook_config_validation(self):
"""Test validation of hook configuration JSON."""
# Load and validate the provided hook configuration
with open("config/claude-hooks.json", "r") as f:
config = json.load(f)
# Verify required hook types are present
required_hooks = ["SessionStart", "UserPromptSubmit", "PostToolUse", "Notification", "Stop"]
for hook_type in required_hooks:
assert hook_type in config["hooks"]
# Verify SessionStart has all session types
session_start_hooks = config["hooks"]["SessionStart"]
matchers = [hook.get("matcher") for hook in session_start_hooks if "matcher" in hook]
assert "startup" in matchers
assert "resume" in matchers
assert "clear" in matchers
# Verify PostToolUse has main tools
post_tool_hooks = config["hooks"]["PostToolUse"]
tool_matchers = [hook.get("matcher") for hook in post_tool_hooks if "matcher" in hook]
assert "Edit" in tool_matchers
assert "Write" in tool_matchers
assert "Read" in tool_matchers
assert "Bash" in tool_matchers
def test_hook_command_structure(self):
"""Test that hook commands have proper structure."""
with open("config/claude-hooks.json", "r") as f:
config = json.load(f)
for hook_type, hooks in config["hooks"].items():
for hook in hooks:
assert "command" in hook
command = hook["command"]
# All commands should make HTTP requests to localhost:8000
assert "http://localhost:8000" in command
assert "curl" in command
# Commands should run in background and suppress output
assert "> /dev/null 2>&1 &" in command
def test_session_id_handling_in_config(self):
"""Test that hook configuration properly handles session ID."""
with open("config/claude-hooks.json", "r") as f:
config = json.load(f)
# Non-SessionStart hooks should use session ID from temp file
for hook_type, hooks in config["hooks"].items():
if hook_type != "SessionStart":
for hook in hooks:
command = hook["command"]
# Should reference the session ID temp file
assert "CLAUDE_SESSION_FILE" in command
assert "/tmp/claude-session-id" in command