diff --git a/codeframe/persistence/database.py b/codeframe/persistence/database.py index c2d924b5..ce74d8cb 100644 --- a/codeframe/persistence/database.py +++ b/codeframe/persistence/database.py @@ -631,6 +631,10 @@ def count_branches_for_issue(self, *args, **kwargs): """Delegate to git_branches.count_branches_for_issue().""" return self.git_branches.count_branches_for_issue(*args, **kwargs) + def get_branch_by_name_and_issues(self, *args, **kwargs): + """Delegate to git_branches.get_branch_by_name_and_issues().""" + return self.git_branches.get_branch_by_name_and_issues(*args, **kwargs) + def create_test_result(self, *args, **kwargs): """Delegate to test_results.create_test_result().""" return self.test_results.create_test_result(*args, **kwargs) diff --git a/codeframe/persistence/repositories/git_repository.py b/codeframe/persistence/repositories/git_repository.py index f6792b43..18f6bdf8 100644 --- a/codeframe/persistence/repositories/git_repository.py +++ b/codeframe/persistence/repositories/git_repository.py @@ -208,3 +208,33 @@ def count_branches_for_issue(self, issue_id: int) -> int: ) return cursor.fetchone()[0] + def get_branch_by_name_and_issues( + self, branch_name: str, issue_ids: List[int] + ) -> Optional[Dict[str, Any]]: + """Get a branch by name that belongs to one of the given issues. + + Single-query lookup for performance optimization. + + Args: + branch_name: Git branch name to find + issue_ids: List of issue IDs the branch must belong to + + Returns: + Branch dictionary or None if not found + """ + if not issue_ids: + return None + + cursor = self.conn.cursor() + placeholders = ",".join("?" * len(issue_ids)) + cursor.execute( + f""" + SELECT * FROM git_branches + WHERE branch_name = ? AND issue_id IN ({placeholders}) + LIMIT 1 + """, + (branch_name, *issue_ids), + ) + row = cursor.fetchone() + return dict(row) if row else None + diff --git a/codeframe/ui/routers/git.py b/codeframe/ui/routers/git.py new file mode 100644 index 00000000..d162f5cb --- /dev/null +++ b/codeframe/ui/routers/git.py @@ -0,0 +1,763 @@ +"""Git operations router for CodeFRAME (#270). + +This module provides REST API endpoints for git operations: +- Branch creation and management +- Commit creation and listing +- Git status retrieval + +Endpoints follow the pattern: /api/projects/{project_id}/git/* +""" + +import logging +import os +import re +from datetime import datetime, UTC +from pathlib import Path +from typing import List, Optional + +from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel, Field, field_validator +import git + +from codeframe.persistence.database import Database +from codeframe.ui.dependencies import get_db +from codeframe.auth.dependencies import get_current_user +from codeframe.auth.models import User +from codeframe.git.workflow_manager import GitWorkflowManager +from codeframe.ui.shared import manager +from codeframe.ui.websocket_broadcasts import ( + broadcast_branch_created, + broadcast_commit_created, +) + +# Git branch name validation pattern +# Allows: alphanumeric, hyphens, underscores, forward slashes, dots (not leading/trailing) +# Disallows: spaces, ~, ^, :, ?, *, [, \, .., @{, consecutive dots +BRANCH_NAME_PATTERN = re.compile(r'^[a-zA-Z0-9][-a-zA-Z0-9_./]*[a-zA-Z0-9]$|^[a-zA-Z0-9]$') + + +def validate_branch_name(branch_name: str) -> str: + """Validate git branch name for safety. + + Args: + branch_name: Branch name to validate + + Returns: + Validated branch name + + Raises: + ValueError: If branch name contains invalid characters + """ + if not branch_name: + raise ValueError("Branch name cannot be empty") + + # Check for dangerous patterns + dangerous_patterns = ['..', '@{', '~', '^', ':', '?', '*', '[', '\\', ' '] + for pattern in dangerous_patterns: + if pattern in branch_name: + raise ValueError(f"Branch name contains invalid character sequence: {pattern}") + + # Check against allowed pattern + if not BRANCH_NAME_PATTERN.match(branch_name): + raise ValueError( + "Branch name must start and end with alphanumeric characters " + "and contain only letters, numbers, hyphens, underscores, forward slashes, or dots" + ) + + return branch_name + + +def validate_file_paths(file_paths: List[str], repo_root: str) -> List[str]: + """Validate file paths to prevent directory traversal attacks. + + Args: + file_paths: List of file paths to validate + repo_root: Repository root directory (working tree) + + Returns: + List of validated, resolved file paths + + Raises: + ValueError: If any path is invalid or attempts to escape the workspace + """ + validated_paths = [] + repo_root_resolved = os.path.realpath(repo_root) + + for path in file_paths: + # Reject absolute paths + if os.path.isabs(path): + raise ValueError(f"Absolute paths not allowed: {path}") + + # Reject paths with '..' segments (directory traversal) + if '..' in path.split(os.sep) or '..' in path.split('/'): + raise ValueError(f"Path traversal not allowed: {path}") + + # Resolve the path against repo root + candidate = os.path.join(repo_root_resolved, path) + resolved = os.path.realpath(candidate) + + # Ensure resolved path is within repo root + try: + common = os.path.commonpath([repo_root_resolved, resolved]) + if common != repo_root_resolved: + raise ValueError(f"Path escapes workspace: {path}") + except ValueError: + # commonpath raises ValueError for paths on different drives (Windows) + raise ValueError(f"Path escapes workspace: {path}") + + validated_paths.append(path) + + return validated_paths + + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api/projects", tags=["git"]) + + +# ============================================================================ +# Request/Response Models +# ============================================================================ + + +class BranchCreateRequest(BaseModel): + """Request model for branch creation.""" + + issue_number: str = Field(..., min_length=1, description="Issue number (e.g., '1.1')") + issue_title: str = Field(..., min_length=1, description="Issue title for branch name") + + @field_validator("issue_number", "issue_title") + @classmethod + def strip_whitespace(cls, v: str) -> str: + """Strip whitespace and validate non-empty.""" + stripped = v.strip() + if not stripped: + raise ValueError("Field cannot be empty or whitespace only") + return stripped + + +class BranchResponse(BaseModel): + """Response model for branch details.""" + + id: int + branch_name: str + issue_id: int + status: str + created_at: str + merged_at: Optional[str] = None + merge_commit: Optional[str] = None + + +class BranchCreateResponse(BaseModel): + """Response model for branch creation.""" + + branch_name: str + issue_number: str + status: str + created_at: str + + +class BranchListResponse(BaseModel): + """Response model for branch listing.""" + + branches: List[BranchResponse] + + +class CommitRequest(BaseModel): + """Request model for commit creation.""" + + task_id: int = Field(..., description="Task ID this commit is for") + files_modified: List[str] = Field(..., min_length=1, description="List of modified files") + agent_id: str = Field(..., min_length=1, description="Agent ID making the commit") + + @field_validator("files_modified") + @classmethod + def validate_files(cls, v: List[str]) -> List[str]: + """Validate files list is not empty.""" + if not v: + raise ValueError("files_modified cannot be empty") + return v + + +class CommitResponse(BaseModel): + """Response model for commit creation.""" + + commit_hash: str + commit_message: str + files_changed: int + timestamp: str + + +class CommitListItem(BaseModel): + """Individual commit in list response.""" + + hash: str + short_hash: str + message: str + author: str + timestamp: str + files_changed: Optional[int] = None + + +class CommitListResponse(BaseModel): + """Response model for commit listing.""" + + commits: List[CommitListItem] + + +class GitStatusResponse(BaseModel): + """Response model for git status.""" + + current_branch: str + is_dirty: bool + modified_files: List[str] + untracked_files: List[str] + staged_files: List[str] + + +# ============================================================================ +# Helper Functions +# ============================================================================ + + +def get_project_or_404(db: Database, project_id: int) -> dict: + """Get project or raise 404. + + Args: + db: Database instance + project_id: Project ID + + Returns: + Project dictionary + + Raises: + HTTPException: 404 if project not found + """ + project = db.get_project(project_id) + if not project: + raise HTTPException(status_code=404, detail=f"Project {project_id} not found") + return project + + +def check_project_access(db: Database, user: User, project_id: int) -> None: + """Check user has project access or raise 403. + + Args: + db: Database instance + user: Current user + project_id: Project ID + + Raises: + HTTPException: 403 if user lacks access + """ + if not db.user_has_project_access(user.id, project_id): + raise HTTPException(status_code=403, detail="Access denied") + + +def get_git_workflow_manager(project: dict, db: Database) -> GitWorkflowManager: + """Get GitWorkflowManager for a project. + + Args: + project: Project dictionary + db: Database instance + + Returns: + GitWorkflowManager instance + + Raises: + HTTPException: 400 if project has no workspace + HTTPException: 500 if not a git repository + """ + workspace_path = project.get("workspace_path") + if not workspace_path: + raise HTTPException(status_code=400, detail="Project has no workspace") + + try: + return GitWorkflowManager(Path(workspace_path), db) + except git.InvalidGitRepositoryError: + raise HTTPException( + status_code=500, detail="Project workspace is not a git repository" + ) + except git.NoSuchPathError: + raise HTTPException(status_code=500, detail="Project workspace path does not exist") + + +# ============================================================================ +# Branch Endpoints +# ============================================================================ + + +@router.post("/{project_id}/git/branches", status_code=201, response_model=BranchCreateResponse) +async def create_branch( + project_id: int, + request: BranchCreateRequest, + db: Database = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """Create a feature branch for an issue. + + Args: + project_id: Project ID + request: Branch creation request + db: Database instance + current_user: Authenticated user + + Returns: + Created branch details + + Raises: + HTTPException: + - 400: Invalid parameters + - 403: Access denied + - 404: Project or issue not found + - 409: Branch already exists (including concurrent creation) + - 500: Git operation failed + """ + # Get project and check access + project = get_project_or_404(db, project_id) + check_project_access(db, current_user, project_id) + + # Find the issue - must exist before creating branch + issues = db.get_project_issues(project_id) + matching_issue = next( + (i for i in issues if i.issue_number == request.issue_number), + None + ) + if not matching_issue: + raise HTTPException( + status_code=404, + detail=f"Issue '{request.issue_number}' not found in project" + ) + issue_id = matching_issue.id + + # Get workflow manager + workflow_manager = get_git_workflow_manager(project, db) + + try: + # Create the branch + branch_name = workflow_manager.create_feature_branch( + request.issue_number, + request.issue_title + ) + + # Broadcast event + await broadcast_branch_created( + manager, + project_id, + branch_name, + request.issue_number, + issue_id, + ) + + logger.info(f"Created branch {branch_name} for project {project_id}") + + return BranchCreateResponse( + branch_name=branch_name, + issue_number=request.issue_number, + status="active", + created_at=datetime.now(UTC).isoformat().replace("+00:00", "Z"), + ) + + except ValueError as e: + # Branch already exists (pre-check) or validation error + error_msg = str(e).lower() + if "already exists" in error_msg: + raise HTTPException(status_code=409, detail=str(e)) + raise HTTPException(status_code=400, detail=str(e)) + except git.GitCommandError as e: + # Handle race condition: branch created between check and create_head + error_msg = str(e).lower() + if "already exists" in error_msg or "cannot lock ref" in error_msg: + logger.warning(f"Branch creation race condition detected: {e}") + raise HTTPException( + status_code=409, + detail="Branch already exists (concurrent creation detected)" + ) + logger.error(f"Git error creating branch: {e}") + raise HTTPException(status_code=500, detail=f"Git operation failed: {e}") + + +# Valid branch status values +VALID_BRANCH_STATUSES = {"active", "merged", "abandoned"} + + +@router.get("/{project_id}/git/branches", response_model=BranchListResponse) +async def list_branches( + project_id: int, + status: Optional[str] = Query(default="active", description="Filter by status (active, merged, abandoned)"), + db: Database = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """List branches for a project. + + Args: + project_id: Project ID + status: Optional status filter (active, merged, abandoned) + db: Database instance + current_user: Authenticated user + + Returns: + List of branches + + Raises: + HTTPException: + - 400: Invalid status value + - 403: Access denied + - 404: Project not found + """ + # Get project and check access + get_project_or_404(db, project_id) + check_project_access(db, current_user, project_id) + + # Validate status parameter + if status is not None and status not in VALID_BRANCH_STATUSES: + raise HTTPException( + status_code=400, + detail=f"Invalid status '{status}'. Must be one of: {', '.join(VALID_BRANCH_STATUSES)}" + ) + + # Get project issues to filter branches + project_issues = db.get_project_issues(project_id) + project_issue_ids = {i.id for i in project_issues} + + # Get branches by status (use explicit None check) + all_branches = db.get_branches_by_status(status) if status is not None else [] + + # Filter to only branches for this project's issues + project_branches = [ + b for b in all_branches + if b.get("issue_id") in project_issue_ids + ] + + # Convert to response format + branches = [ + BranchResponse( + id=b["id"], + branch_name=b["branch_name"], + issue_id=b["issue_id"], + status=b["status"], + created_at=b.get("created_at", ""), + merged_at=b.get("merged_at"), + merge_commit=b.get("merge_commit"), + ) + for b in project_branches + ] + + return BranchListResponse(branches=branches) + + +@router.get("/{project_id}/git/branches/{branch_name:path}", response_model=BranchResponse) +async def get_branch( + project_id: int, + branch_name: str, + db: Database = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """Get branch details. + + Args: + project_id: Project ID + branch_name: Branch name + db: Database instance + current_user: Authenticated user + + Returns: + Branch details + + Raises: + HTTPException: + - 403: Access denied + - 404: Project or branch not found + """ + # Get project and check access + get_project_or_404(db, project_id) + check_project_access(db, current_user, project_id) + + # Validate branch name + try: + validate_branch_name(branch_name) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + # Get project issues to verify branch belongs to project + project_issues = db.get_project_issues(project_id) + project_issue_ids = [i.id for i in project_issues] + + # Single-query lookup for performance + branch = db.get_branch_by_name_and_issues(branch_name, project_issue_ids) + if branch: + return BranchResponse( + id=branch["id"], + branch_name=branch["branch_name"], + issue_id=branch["issue_id"], + status=branch["status"], + created_at=branch.get("created_at", ""), + merged_at=branch.get("merged_at"), + merge_commit=branch.get("merge_commit"), + ) + + raise HTTPException(status_code=404, detail=f"Branch '{branch_name}' not found") + + +# ============================================================================ +# Commit Endpoints +# ============================================================================ + + +@router.post("/{project_id}/git/commit", status_code=201, response_model=CommitResponse) +async def create_commit( + project_id: int, + request: CommitRequest, + db: Database = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """Create a git commit for task changes. + + Args: + project_id: Project ID + request: Commit request + db: Database instance + current_user: Authenticated user + + Returns: + Commit details + + Raises: + HTTPException: + - 400: No files to commit or validation error + - 403: Access denied + - 404: Project or task not found + - 500: Git operation failed + """ + # Get project and check access + project = get_project_or_404(db, project_id) + check_project_access(db, current_user, project_id) + + # Validate files_modified is not empty + if not request.files_modified: + raise HTTPException(status_code=400, detail="No files to commit") + + # Get task and verify it belongs to this project + task = db.get_task(request.task_id) + if not task: + raise HTTPException(status_code=404, detail=f"Task {request.task_id} not found") + if task.project_id != project_id: + raise HTTPException(status_code=404, detail=f"Task {request.task_id} not found in project") + + # Get workflow manager + workflow_manager = get_git_workflow_manager(project, db) + + # Validate file paths to prevent directory traversal attacks + try: + validate_file_paths(request.files_modified, workflow_manager.repo.working_tree_dir) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + try: + # Create task dict for commit_task_changes + task_dict = { + "id": task.id, + "project_id": task.project_id, + "task_number": task.task_number, + "title": task.title, + "description": task.description, + } + + # Create commit + commit_hash = workflow_manager.commit_task_changes( + task_dict, + request.files_modified, + request.agent_id + ) + + # Get commit message from the specific commit (not HEAD, which may have moved) + commit = workflow_manager.repo.commit(commit_hash) + commit_message = commit.message.strip() + + # Broadcast event + await broadcast_commit_created( + manager, + project_id, + task.id, + commit_hash, + commit_message, + len(request.files_modified), + ) + + logger.info(f"Created commit {commit_hash[:7]} for task {task.task_number}") + + return CommitResponse( + commit_hash=commit_hash, + commit_message=commit_message, + files_changed=len(request.files_modified), + timestamp=datetime.now(UTC).isoformat().replace("+00:00", "Z"), + ) + + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + except KeyError as e: + raise HTTPException(status_code=400, detail=f"Invalid task data: {e}") + except git.GitCommandError as e: + logger.error(f"Git error creating commit: {e}") + raise HTTPException(status_code=500, detail=f"Git operation failed: {e}") + + +@router.get("/{project_id}/git/commits", response_model=CommitListResponse) +async def list_commits( + project_id: int, + branch: Optional[str] = Query(default=None, description="Branch name (default: current)"), + limit: int = Query(default=50, ge=1, le=100, description="Max commits to return"), + db: Database = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """List git commits for a project. + + Args: + project_id: Project ID + branch: Optional branch name (default: current branch) + limit: Maximum number of commits to return (1-100) + db: Database instance + current_user: Authenticated user + + Returns: + List of commits + + Raises: + HTTPException: + - 403: Access denied + - 404: Project not found + - 500: Git operation failed + """ + # Get project and check access + project = get_project_or_404(db, project_id) + check_project_access(db, current_user, project_id) + + # Get workflow manager + workflow_manager = get_git_workflow_manager(project, db) + + # Validate branch name if provided + if branch: + try: + validate_branch_name(branch) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + try: + # Get commit iterator + if branch: + commits_iter = workflow_manager.repo.iter_commits(branch, max_count=limit) + else: + commits_iter = workflow_manager.repo.iter_commits(max_count=limit) + + # Build commit list + commits = [] + for commit in commits_iter: + # Count files changed (if available) + try: + files_changed = commit.stats.total.get("files", 0) if commit.stats else None + except Exception: + files_changed = None + + commits.append( + CommitListItem( + hash=commit.hexsha, + short_hash=commit.hexsha[:7], + message=commit.message.strip().split("\n")[0], # First line only + author=str(commit.author), + timestamp=commit.committed_datetime.astimezone(UTC).isoformat().replace("+00:00", "Z"), + files_changed=files_changed, + ) + ) + + return CommitListResponse(commits=commits) + + except git.GitCommandError as e: + logger.error(f"Git error listing commits: {e}") + raise HTTPException(status_code=500, detail=f"Git operation failed: {e}") + except git.BadName as e: + # Invalid branch/ref name + logger.warning(f"Invalid branch name in list_commits: {e}") + raise HTTPException(status_code=400, detail=f"Invalid branch reference: {branch}") + except (ValueError, KeyError) as e: + logger.error(f"Data error listing commits: {e}") + raise HTTPException(status_code=500, detail=f"Data error: {e}") + + +# ============================================================================ +# Status Endpoint +# ============================================================================ + + +@router.get("/{project_id}/git/status", response_model=GitStatusResponse) +async def get_git_status( + project_id: int, + db: Database = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """Get git working tree status. + + Args: + project_id: Project ID + db: Database instance + current_user: Authenticated user + + Returns: + Git status including current branch and file states + + Raises: + HTTPException: + - 403: Access denied + - 404: Project not found + - 500: Git operation failed + """ + # Get project and check access + project = get_project_or_404(db, project_id) + check_project_access(db, current_user, project_id) + + # Get workflow manager + workflow_manager = get_git_workflow_manager(project, db) + + try: + repo = workflow_manager.repo + + # Get current branch + current_branch = workflow_manager.get_current_branch() + + # Check if dirty + is_dirty = repo.is_dirty(untracked_files=True) + + # Get modified files (tracked, unstaged changes) + modified_files = [item.a_path for item in repo.index.diff(None)] + + # Get untracked files + untracked_files = repo.untracked_files + + # Get staged files (handle repos with no commits/HEAD) + try: + if repo.head.is_valid(): + staged_files = [item.a_path for item in repo.index.diff("HEAD")] + else: + # No HEAD yet - all indexed files are staged + # entries.keys() returns (path, stage) tuples, extract just the path + staged_files = [path for path, _stage in repo.index.entries.keys()] if repo.index.entries else [] + except git.BadName: + # HEAD reference doesn't exist (empty repo) + staged_files = [] + + return GitStatusResponse( + current_branch=current_branch, + is_dirty=is_dirty, + modified_files=modified_files, + untracked_files=list(untracked_files), + staged_files=staged_files, + ) + + except git.GitCommandError as e: + logger.error(f"Git error getting status: {e}") + raise HTTPException(status_code=500, detail=f"Git operation failed: {e}") + except git.InvalidGitRepositoryError as e: + logger.error(f"Invalid git repository: {e}") + raise HTTPException(status_code=500, detail="Invalid git repository") + except (ValueError, KeyError, AttributeError) as e: + logger.error(f"Data error getting git status: {e}") + raise HTTPException(status_code=500, detail=f"Data error: {e}") diff --git a/codeframe/ui/server.py b/codeframe/ui/server.py index 6f5e6334..c8f60d08 100644 --- a/codeframe/ui/server.py +++ b/codeframe/ui/server.py @@ -24,6 +24,7 @@ checkpoints, context, discovery, + git, lint, metrics, projects, @@ -330,6 +331,7 @@ async def test_broadcast(message: dict, project_id: int = None): app.include_router(checkpoints.router) app.include_router(context.router) app.include_router(discovery.router) +app.include_router(git.router) app.include_router(lint.router) app.include_router(metrics.router) app.include_router(projects.router) diff --git a/codeframe/ui/websocket_broadcasts.py b/codeframe/ui/websocket_broadcasts.py index 6c392f5b..d47e76f4 100644 --- a/codeframe/ui/websocket_broadcasts.py +++ b/codeframe/ui/websocket_broadcasts.py @@ -236,6 +236,41 @@ async def broadcast_commit_created( logger.error(f"Failed to broadcast commit: {e}") +async def broadcast_branch_created( + manager, + project_id: int, + branch_name: str, + issue_number: str, + issue_id: Optional[int] = None, +) -> None: + """ + Broadcast git branch creation to connected clients. + + Args: + manager: ConnectionManager instance + project_id: Project ID + branch_name: Git branch name created + issue_number: Issue number the branch is for + issue_id: Optional issue ID + """ + message = { + "type": "branch_created", + "project_id": project_id, + "branch_name": branch_name, + "issue_number": issue_number, + "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"), + } + + if issue_id is not None: + message["issue_id"] = issue_id + + try: + await manager.broadcast(message, project_id=project_id) + logger.debug(f"Broadcast branch_created: {branch_name} for issue {issue_number}") + except Exception as e: + logger.error(f"Failed to broadcast branch creation: {e}") + + async def broadcast_activity_update( manager, project_id: int, diff --git a/tests/api/conftest.py b/tests/api/conftest.py index a8504dd2..fa389833 100644 --- a/tests/api/conftest.py +++ b/tests/api/conftest.py @@ -212,7 +212,9 @@ def clean_database_between_tests(api_client: TestClient) -> Generator[None, None cursor.execute("DELETE FROM checkpoints") cursor.execute("DELETE FROM memory") cursor.execute("DELETE FROM blockers") + cursor.execute("DELETE FROM changelog") # References projects, tasks cursor.execute("DELETE FROM tasks") + cursor.execute("DELETE FROM git_branches") # Must be before issues (FK constraint) cursor.execute("DELETE FROM issues") cursor.execute("DELETE FROM project_agents") # Multi-agent junction table cursor.execute("DELETE FROM agents") diff --git a/tests/api/test_git_api.py b/tests/api/test_git_api.py new file mode 100644 index 00000000..83559d84 --- /dev/null +++ b/tests/api/test_git_api.py @@ -0,0 +1,821 @@ +"""Tests for Git REST API endpoints (#270). + +This module tests the git router endpoints for: +- Branch creation and management +- Commit creation and listing +- Git status retrieval + +Tests follow RED-GREEN-REFACTOR TDD cycle. +""" + +import tempfile +from pathlib import Path +from typing import Generator +import pytest +import git + + +def get_app(): + """Get the current app instance after module reload. + + Imports app locally to ensure we get the freshly reloaded instance + after api_client fixture reloads codeframe.ui.server. + """ + from codeframe.ui.server import app + + return app + + +@pytest.fixture(scope="function") +def test_project_with_git(api_client) -> Generator[dict, None, None]: + """Create a test project with an initialized git repository. + + Args: + api_client: FastAPI test client + + Yields: + Project dictionary with id, workspace_path, and git repo + """ + db = get_app().state.db + + # Create temporary directory for git workspace + with tempfile.TemporaryDirectory() as temp_dir: + workspace_path = Path(temp_dir) / "project" + workspace_path.mkdir(parents=True) + + # Initialize git repository + repo = git.Repo.init(workspace_path) + + # Create initial commit (required for branch operations) + readme_path = workspace_path / "README.md" + readme_path.write_text("# Test Project\n") + repo.index.add(["README.md"]) + repo.index.commit("Initial commit") + + # Create project in database with workspace path + project_id = db.create_project( + name="Test Git Project", + description="Test project for git API tests", + workspace_path=str(workspace_path), + ) + + yield { + "id": project_id, + "workspace_path": str(workspace_path), + "repo": repo, + } + + +@pytest.fixture(scope="function") +def test_project_with_issue(test_project_with_git, api_client) -> Generator[dict, None, None]: + """Create a test project with an issue for branch creation. + + Args: + test_project_with_git: Project fixture with git repository + api_client: FastAPI test client + + Yields: + Dictionary with project_id, issue_id, issue_number, workspace_path + """ + from codeframe.core.models import Issue + + db = get_app().state.db + project = test_project_with_git + + # Create an issue for the project using Issue model + issue = Issue( + project_id=project["id"], + issue_number="1.1", + title="User Authentication", + description="Implement user authentication feature", + priority=1, + ) + issue_id = db.create_issue(issue) + + yield { + "project_id": project["id"], + "issue_id": issue_id, + "issue_number": "1.1", + "issue_title": "User Authentication", + "workspace_path": project["workspace_path"], + "repo": project["repo"], + } + + +@pytest.fixture(scope="function") +def test_project_with_task(test_project_with_issue, api_client) -> Generator[dict, None, None]: + """Create a test project with a task for commit operations. + + Args: + test_project_with_issue: Project fixture with issue + api_client: FastAPI test client + + Yields: + Dictionary with project_id, issue_id, task_id, workspace_path + """ + from codeframe.core.models import Task, TaskStatus + + db = get_app().state.db + project = test_project_with_issue + + # Create a task for the issue using Task model + task = Task( + project_id=project["project_id"], + task_number="1.1.1", + title="Implement login endpoint", + description="Create POST /api/login endpoint", + status=TaskStatus.IN_PROGRESS, + ) + task_id = db.create_task(task) + + yield { + **project, + "task_id": task_id, + "task_number": "1.1.1", + } + + +# ============================================================================ +# Branch Creation Tests +# ============================================================================ + + +class TestGitBranchCreation: + """Test branch creation endpoint: POST /api/projects/{id}/git/branches.""" + + def test_create_branch_endpoint_exists(self, api_client, test_project_with_issue): + """Test that POST /api/projects/{id}/git/branches endpoint exists.""" + project = test_project_with_issue + response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + # Should not return 404 (endpoint not found) + assert response.status_code != 404 + + def test_create_branch_returns_201(self, api_client, test_project_with_issue): + """Test that successful branch creation returns 201.""" + project = test_project_with_issue + response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + assert response.status_code == 201 + + def test_create_branch_returns_branch_name(self, api_client, test_project_with_issue): + """Test that response includes branch_name.""" + project = test_project_with_issue + response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + data = response.json() + assert "branch_name" in data + assert data["branch_name"].startswith("issue-") + + def test_create_branch_includes_issue_number_in_name(self, api_client, test_project_with_issue): + """Test that branch name includes issue number.""" + project = test_project_with_issue + response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + data = response.json() + assert project["issue_number"] in data["branch_name"] + + def test_create_branch_returns_status(self, api_client, test_project_with_issue): + """Test that response includes status as 'active'.""" + project = test_project_with_issue + response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + data = response.json() + assert "status" in data + assert data["status"] == "active" + + def test_create_branch_creates_git_branch(self, api_client, test_project_with_issue): + """Test that branch is actually created in git repository.""" + project = test_project_with_issue + response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + data = response.json() + + # Verify branch exists in git + repo = project["repo"] + branch_names = [b.name for b in repo.branches] + assert data["branch_name"] in branch_names + + def test_create_branch_project_not_found(self, api_client): + """Test that non-existent project returns 404.""" + response = api_client.post( + "/api/projects/99999/git/branches", + json={ + "issue_number": "1.1", + "issue_title": "Test Issue", + }, + ) + assert response.status_code == 404 + + def test_create_branch_requires_issue_number(self, api_client, test_project_with_issue): + """Test that issue_number is required.""" + project = test_project_with_issue + response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_title": project["issue_title"], + }, + ) + assert response.status_code == 422 + + def test_create_branch_requires_issue_title(self, api_client, test_project_with_issue): + """Test that issue_title is required.""" + project = test_project_with_issue + response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + }, + ) + assert response.status_code == 422 + + def test_create_duplicate_branch_returns_409_conflict(self, api_client, test_project_with_issue): + """Test that creating duplicate branch returns 409 Conflict.""" + project = test_project_with_issue + + # Create branch first time + api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + + # Try to create same branch again - should return 409 Conflict + response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + assert response.status_code == 409 + assert "already exists" in response.json()["detail"].lower() + + def test_create_branch_issue_not_found(self, api_client, test_project_with_git): + """Test that creating branch with non-existent issue returns 404.""" + project = test_project_with_git + response = api_client.post( + f"/api/projects/{project['id']}/git/branches", + json={ + "issue_number": "999.999", + "issue_title": "Non-existent Issue", + }, + ) + assert response.status_code == 404 + assert "not found" in response.json()["detail"].lower() + + +# ============================================================================ +# Branch Listing Tests +# ============================================================================ + + +class TestGitBranchListing: + """Test branch listing endpoint: GET /api/projects/{id}/git/branches.""" + + def test_list_branches_endpoint_exists(self, api_client, test_project_with_git): + """Test that GET /api/projects/{id}/git/branches endpoint exists.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/branches") + assert response.status_code != 404 + + def test_list_branches_returns_200(self, api_client, test_project_with_git): + """Test that successful request returns 200.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/branches") + assert response.status_code == 200 + + def test_list_branches_returns_list(self, api_client, test_project_with_git): + """Test that response is a list of branches.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/branches") + data = response.json() + assert "branches" in data + assert isinstance(data["branches"], list) + + def test_list_branches_empty_when_no_branches(self, api_client, test_project_with_git): + """Test that empty list returned when no feature branches exist.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/branches") + data = response.json() + assert data["branches"] == [] + + def test_list_branches_includes_created_branch(self, api_client, test_project_with_issue): + """Test that created branch appears in list.""" + project = test_project_with_issue + + # Create a branch + create_response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + created_branch = create_response.json()["branch_name"] + + # List branches + response = api_client.get(f"/api/projects/{project['project_id']}/git/branches") + data = response.json() + + branch_names = [b["branch_name"] for b in data["branches"]] + assert created_branch in branch_names + + def test_list_branches_project_not_found(self, api_client): + """Test that non-existent project returns 404.""" + response = api_client.get("/api/projects/99999/git/branches") + assert response.status_code == 404 + + +# ============================================================================ +# Branch Details Tests +# ============================================================================ + + +class TestGitBranchDetails: + """Test branch details endpoint: GET /api/projects/{id}/git/branches/{name}.""" + + def test_get_branch_endpoint_exists(self, api_client, test_project_with_issue): + """Test that GET /api/projects/{id}/git/branches/{name} endpoint exists.""" + project = test_project_with_issue + + # Create a branch first + create_response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + branch_name = create_response.json()["branch_name"] + + response = api_client.get( + f"/api/projects/{project['project_id']}/git/branches/{branch_name}" + ) + assert response.status_code != 404 + + def test_get_branch_returns_200(self, api_client, test_project_with_issue): + """Test that successful request returns 200.""" + project = test_project_with_issue + + create_response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + branch_name = create_response.json()["branch_name"] + + response = api_client.get( + f"/api/projects/{project['project_id']}/git/branches/{branch_name}" + ) + assert response.status_code == 200 + + def test_get_branch_includes_required_fields(self, api_client, test_project_with_issue): + """Test that response includes all required fields.""" + project = test_project_with_issue + + create_response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + branch_name = create_response.json()["branch_name"] + + response = api_client.get( + f"/api/projects/{project['project_id']}/git/branches/{branch_name}" + ) + data = response.json() + + assert "id" in data + assert "branch_name" in data + assert "issue_id" in data + assert "status" in data + assert "created_at" in data + + def test_get_branch_not_found(self, api_client, test_project_with_git): + """Test that non-existent branch returns 404.""" + project = test_project_with_git + response = api_client.get( + f"/api/projects/{project['id']}/git/branches/nonexistent-branch" + ) + assert response.status_code == 404 + + +# ============================================================================ +# Commit Creation Tests +# ============================================================================ + + +class TestGitCommitCreation: + """Test commit creation endpoint: POST /api/projects/{id}/git/commit.""" + + def test_commit_endpoint_exists(self, api_client, test_project_with_task): + """Test that POST /api/projects/{id}/git/commit endpoint exists.""" + project = test_project_with_task + + # Create a file to commit + workspace_path = Path(project["workspace_path"]) + test_file = workspace_path / "test.py" + test_file.write_text("# Test file\n") + + response = api_client.post( + f"/api/projects/{project['project_id']}/git/commit", + json={ + "task_id": project["task_id"], + "files_modified": ["test.py"], + "agent_id": "backend-worker-001", + }, + ) + assert response.status_code != 404 + + def test_commit_returns_201(self, api_client, test_project_with_task): + """Test that successful commit returns 201.""" + project = test_project_with_task + + # Create a file to commit + workspace_path = Path(project["workspace_path"]) + test_file = workspace_path / "auth.py" + test_file.write_text("# Authentication module\n") + + response = api_client.post( + f"/api/projects/{project['project_id']}/git/commit", + json={ + "task_id": project["task_id"], + "files_modified": ["auth.py"], + "agent_id": "backend-worker-001", + }, + ) + assert response.status_code == 201 + + def test_commit_returns_commit_hash(self, api_client, test_project_with_task): + """Test that response includes commit_hash.""" + project = test_project_with_task + + workspace_path = Path(project["workspace_path"]) + test_file = workspace_path / "login.py" + test_file.write_text("# Login endpoint\n") + + response = api_client.post( + f"/api/projects/{project['project_id']}/git/commit", + json={ + "task_id": project["task_id"], + "files_modified": ["login.py"], + "agent_id": "backend-worker-001", + }, + ) + data = response.json() + + assert "commit_hash" in data + assert len(data["commit_hash"]) == 40 # Full SHA hash + + def test_commit_returns_commit_message(self, api_client, test_project_with_task): + """Test that response includes commit_message.""" + project = test_project_with_task + + workspace_path = Path(project["workspace_path"]) + test_file = workspace_path / "session.py" + test_file.write_text("# Session management\n") + + response = api_client.post( + f"/api/projects/{project['project_id']}/git/commit", + json={ + "task_id": project["task_id"], + "files_modified": ["session.py"], + "agent_id": "backend-worker-001", + }, + ) + data = response.json() + + assert "commit_message" in data + assert isinstance(data["commit_message"], str) + + def test_commit_requires_task_id(self, api_client, test_project_with_task): + """Test that task_id is required.""" + project = test_project_with_task + response = api_client.post( + f"/api/projects/{project['project_id']}/git/commit", + json={ + "files_modified": ["test.py"], + "agent_id": "backend-worker-001", + }, + ) + assert response.status_code == 422 + + def test_commit_requires_files_modified(self, api_client, test_project_with_task): + """Test that files_modified is required.""" + project = test_project_with_task + response = api_client.post( + f"/api/projects/{project['project_id']}/git/commit", + json={ + "task_id": project["task_id"], + "agent_id": "backend-worker-001", + }, + ) + assert response.status_code == 422 + + def test_commit_requires_agent_id(self, api_client, test_project_with_task): + """Test that agent_id is required.""" + project = test_project_with_task + response = api_client.post( + f"/api/projects/{project['project_id']}/git/commit", + json={ + "task_id": project["task_id"], + "files_modified": ["test.py"], + }, + ) + assert response.status_code == 422 + + def test_commit_rejects_empty_files_list(self, api_client, test_project_with_task): + """Test that empty files_modified list is rejected.""" + project = test_project_with_task + response = api_client.post( + f"/api/projects/{project['project_id']}/git/commit", + json={ + "task_id": project["task_id"], + "files_modified": [], + "agent_id": "backend-worker-001", + }, + ) + # Pydantic validation returns 422 for min_length constraint + assert response.status_code == 422 + + def test_commit_project_not_found(self, api_client): + """Test that non-existent project returns 404.""" + response = api_client.post( + "/api/projects/99999/git/commit", + json={ + "task_id": 1, + "files_modified": ["test.py"], + "agent_id": "backend-worker-001", + }, + ) + assert response.status_code == 404 + + def test_commit_task_not_found(self, api_client, test_project_with_git): + """Test that non-existent task returns 404.""" + project = test_project_with_git + response = api_client.post( + f"/api/projects/{project['id']}/git/commit", + json={ + "task_id": 99999, + "files_modified": ["test.py"], + "agent_id": "backend-worker-001", + }, + ) + assert response.status_code == 404 + + def test_commit_rejects_absolute_paths(self, api_client, test_project_with_task): + """Test that absolute file paths are rejected (security).""" + project = test_project_with_task + response = api_client.post( + f"/api/projects/{project['project_id']}/git/commit", + json={ + "task_id": project["task_id"], + "files_modified": ["/etc/passwd"], + "agent_id": "backend-worker-001", + }, + ) + assert response.status_code == 400 + assert "absolute" in response.json()["detail"].lower() + + def test_commit_rejects_path_traversal(self, api_client, test_project_with_task): + """Test that path traversal attempts are rejected (security).""" + project = test_project_with_task + response = api_client.post( + f"/api/projects/{project['project_id']}/git/commit", + json={ + "task_id": project["task_id"], + "files_modified": ["../../../etc/passwd"], + "agent_id": "backend-worker-001", + }, + ) + assert response.status_code == 400 + assert "traversal" in response.json()["detail"].lower() + + def test_commit_rejects_workspace_escape(self, api_client, test_project_with_task): + """Test that paths escaping workspace are rejected (security).""" + project = test_project_with_task + # Use a path that doesn't have '..' but could resolve outside via symlinks + # This tests the commonpath check + response = api_client.post( + f"/api/projects/{project['project_id']}/git/commit", + json={ + "task_id": project["task_id"], + "files_modified": ["subdir/../../../outside.txt"], + "agent_id": "backend-worker-001", + }, + ) + assert response.status_code == 400 + + +# ============================================================================ +# Commit Listing Tests +# ============================================================================ + + +class TestGitCommitListing: + """Test commit listing endpoint: GET /api/projects/{id}/git/commits.""" + + def test_list_commits_endpoint_exists(self, api_client, test_project_with_git): + """Test that GET /api/projects/{id}/git/commits endpoint exists.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/commits") + assert response.status_code != 404 + + def test_list_commits_returns_200(self, api_client, test_project_with_git): + """Test that successful request returns 200.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/commits") + assert response.status_code == 200 + + def test_list_commits_returns_list(self, api_client, test_project_with_git): + """Test that response is a list of commits.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/commits") + data = response.json() + assert "commits" in data + assert isinstance(data["commits"], list) + + def test_list_commits_includes_initial_commit(self, api_client, test_project_with_git): + """Test that initial commit is included in list.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/commits") + data = response.json() + + # Should have at least the initial commit + assert len(data["commits"]) >= 1 + + def test_list_commits_with_limit(self, api_client, test_project_with_git): + """Test that limit parameter works.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/commits?limit=1") + data = response.json() + + assert len(data["commits"]) <= 1 + + def test_list_commits_commit_has_required_fields(self, api_client, test_project_with_git): + """Test that each commit has required fields.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/commits") + data = response.json() + + if data["commits"]: + commit = data["commits"][0] + assert "hash" in commit + assert "short_hash" in commit + assert "message" in commit + assert "author" in commit + assert "timestamp" in commit + + def test_list_commits_project_not_found(self, api_client): + """Test that non-existent project returns 404.""" + response = api_client.get("/api/projects/99999/git/commits") + assert response.status_code == 404 + + +# ============================================================================ +# Git Status Tests +# ============================================================================ + + +class TestGitStatus: + """Test git status endpoint: GET /api/projects/{id}/git/status.""" + + def test_status_endpoint_exists(self, api_client, test_project_with_git): + """Test that GET /api/projects/{id}/git/status endpoint exists.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/status") + assert response.status_code != 404 + + def test_status_returns_200(self, api_client, test_project_with_git): + """Test that successful request returns 200.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/status") + assert response.status_code == 200 + + def test_status_includes_current_branch(self, api_client, test_project_with_git): + """Test that response includes current_branch.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/status") + data = response.json() + + assert "current_branch" in data + assert isinstance(data["current_branch"], str) + + def test_status_includes_is_dirty(self, api_client, test_project_with_git): + """Test that response includes is_dirty flag.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/status") + data = response.json() + + assert "is_dirty" in data + assert isinstance(data["is_dirty"], bool) + + def test_status_includes_file_lists(self, api_client, test_project_with_git): + """Test that response includes file status lists.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/status") + data = response.json() + + assert "modified_files" in data + assert "untracked_files" in data + assert "staged_files" in data + assert isinstance(data["modified_files"], list) + assert isinstance(data["untracked_files"], list) + assert isinstance(data["staged_files"], list) + + def test_status_clean_repo(self, api_client, test_project_with_git): + """Test status on clean repository.""" + project = test_project_with_git + response = api_client.get(f"/api/projects/{project['id']}/git/status") + data = response.json() + + assert data["is_dirty"] is False + assert data["modified_files"] == [] + assert data["untracked_files"] == [] + + def test_status_with_untracked_file(self, api_client, test_project_with_git): + """Test status with untracked files.""" + project = test_project_with_git + + # Create an untracked file + workspace_path = Path(project["workspace_path"]) + new_file = workspace_path / "new_file.py" + new_file.write_text("# New file\n") + + response = api_client.get(f"/api/projects/{project['id']}/git/status") + data = response.json() + + assert data["is_dirty"] is True + assert "new_file.py" in data["untracked_files"] + + def test_status_project_not_found(self, api_client): + """Test that non-existent project returns 404.""" + response = api_client.get("/api/projects/99999/git/status") + assert response.status_code == 404 + + +# ============================================================================ +# Authorization Tests +# ============================================================================ + + +class TestGitApiAuthorization: + """Test authorization for git API endpoints.""" + + def test_create_branch_requires_auth(self, api_client, test_project_with_issue): + """Test that branch creation requires authentication.""" + project = test_project_with_issue + + # Remove auth header + del api_client.headers["Authorization"] + + response = api_client.post( + f"/api/projects/{project['project_id']}/git/branches", + json={ + "issue_number": project["issue_number"], + "issue_title": project["issue_title"], + }, + ) + + # Restore auth header for cleanup + from tests.api.conftest import create_test_jwt_token + api_client.headers["Authorization"] = f"Bearer {create_test_jwt_token()}" + + assert response.status_code == 401