""" Linear API simulator. Provides realistic mock responses for Linear API actions (GraphQL). """ from typing import Optional from .base import BaseSimulator from datetime import datetime import uuid class LinearSimulator(BaseSimulator): """Simulator for Linear API (GraphQL).""" def __init__(self): super().__init__('linear') def load_mock_responses(self): """Load Linear mock response templates.""" self.mock_responses = { 'create_issue': self._create_issue_template, 'read_issues': self._read_issues_template, 'update_issue': self._update_issue_template, 'add_comment': self._add_comment_template, 'get_issue': self._get_issue_template, 'search_issues': self._search_issues_template, } def get_required_permissions(self, action: str) -> set[str]: """Get required Linear permissions for an action.""" permissions_map = { 'create_issue': {'write'}, 'read_issues': {'read'}, 'update_issue': {'write'}, 'add_comment': {'write'}, 'get_issue': {'read'}, 'search_issues': {'read'}, } return permissions_map.get(action, set()) def validate_params(self, action: str, params: dict) -> tuple[bool, Optional[str]]: """Validate parameters for Linear actions.""" if action == 'create_issue': required = {'title', 'teamId'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" elif action == 'read_issues': # No required params - will return all issues pass elif action == 'update_issue': required = {'id'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" # Must have at least one field to update updateable = {'title', 'description', 'stateId', 'priority', 'assigneeId'} if not any(k in params for k in updateable): return False, "Must provide at least one field to update" elif action == 'add_comment': required = {'issueId', 'body'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" elif action == 'get_issue': required = {'id'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" elif action == 'search_issues': required = {'query'} missing = required - set(params.keys()) if missing: return False, f"Missing required parameters: {missing}" else: return False, f"Unknown action: {action}" return True, None def generate_mock_response(self, action: str, params: dict) -> dict: """Generate realistic Linear API response.""" if action not in self.mock_responses: raise ValueError(f"Unknown action: {action}") template_func = self.mock_responses[action] return template_func(params) def _generate_issue_id(self) -> str: """Generate a Linear-style issue ID.""" return str(uuid.uuid4()) def _create_issue_template(self, params: dict) -> dict: """Mock response for creating an issue.""" issue_id = self._generate_issue_id() issue_number = 1234 # Mock issue number return { "data": { "issueCreate": { "success": True, "issue": { "id": issue_id, "createdAt": datetime.utcnow().isoformat() + "Z", "updatedAt": datetime.utcnow().isoformat() + "Z", "number": issue_number, "title": params['title'], "description": params.get('description', ''), "priority": params.get('priority', 0), "estimate": params.get('estimate'), "team": { "id": params['teamId'], "name": "Engineering", "key": "ENG" }, "state": { "id": params.get('stateId', 'state-backlog'), "name": "Backlog", "type": "backlog" }, "assignee": { "id": params.get('assigneeId', 'unassigned'), "name": "Unassigned" if not params.get('assigneeId') else "Team Member" } if params.get('assigneeId') else None, "labels": { "nodes": [] }, "url": f"https://linear.app/team/issue/ENG-{issue_number}" } } } } def _read_issues_template(self, params: dict) -> dict: """Mock response for reading issues.""" # Generate a few mock issues issues = [] for i in range(3): issue_id = self._generate_issue_id() issues.append({ "id": issue_id, "createdAt": datetime.utcnow().isoformat() + "Z", "updatedAt": datetime.utcnow().isoformat() + "Z", "number": 1230 + i, "title": f"Sample issue {i+1}", "description": f"This is a sample issue for testing purposes", "priority": 1, "state": { "id": "state-in-progress" if i == 0 else "state-backlog", "name": "In Progress" if i == 0 else "Backlog", "type": "started" if i == 0 else "backlog" }, "team": { "id": params.get('teamId', 'team-eng'), "name": "Engineering", "key": "ENG" }, "assignee": { "id": f"user-{i}", "name": f"Developer {i+1}" } if i < 2 else None, "url": f"https://linear.app/team/issue/ENG-{1230+i}" }) return { "data": { "issues": { "nodes": issues, "pageInfo": { "hasNextPage": False, "endCursor": None } } } } def _update_issue_template(self, params: dict) -> dict: """Mock response for updating an issue.""" return { "data": { "issueUpdate": { "success": True, "issue": { "id": params['id'], "updatedAt": datetime.utcnow().isoformat() + "Z", "title": params.get('title', 'Updated Issue'), "description": params.get('description', ''), "priority": params.get('priority', 1), "state": { "id": params.get('stateId', 'state-in-progress'), "name": "In Progress" }, "assignee": { "id": params.get('assigneeId', 'user-1'), "name": "Assigned User" } if params.get('assigneeId') else None } } } } def _add_comment_template(self, params: dict) -> dict: """Mock response for adding a comment to an issue.""" comment_id = self._generate_issue_id() return { "data": { "commentCreate": { "success": True, "comment": { "id": comment_id, "createdAt": datetime.utcnow().isoformat() + "Z", "updatedAt": datetime.utcnow().isoformat() + "Z", "body": params['body'], "user": { "id": "user-current", "name": "Current User", "email": "user@company.com" }, "issue": { "id": params['issueId'], "title": "Related Issue" } } } } } def _get_issue_template(self, params: dict) -> dict: """Mock response for getting a specific issue.""" return { "data": { "issue": { "id": params['id'], "createdAt": "2025-11-15T10:00:00.000Z", "updatedAt": datetime.utcnow().isoformat() + "Z", "number": 1234, "title": "Implement new feature", "description": "We need to implement this new feature for our users.\n\nAcceptance criteria:\n- Feature works as expected\n- Tests are written\n- Documentation is updated", "priority": 2, "estimate": 5, "state": { "id": "state-in-progress", "name": "In Progress", "type": "started", "color": "#f2c94c" }, "team": { "id": "team-eng", "name": "Engineering", "key": "ENG" }, "assignee": { "id": "user-1", "name": "Jane Developer", "email": "jane@company.com", "avatarUrl": "https://avatar.linear.app/user-1" }, "creator": { "id": "user-pm", "name": "Product Manager" }, "labels": { "nodes": [ { "id": "label-feature", "name": "feature", "color": "#4ea7fc" } ] }, "comments": { "nodes": [ { "id": "comment-1", "body": "Working on this now", "createdAt": "2025-11-20T14:30:00.000Z", "user": { "name": "Jane Developer" } } ] }, "url": "https://linear.app/team/issue/ENG-1234" } } } def _search_issues_template(self, params: dict) -> dict: """Mock response for searching issues.""" # Simple search - return one matching issue return { "data": { "issueSearch": { "nodes": [ { "id": self._generate_issue_id(), "number": 1235, "title": f"Search result for: {params['query']}", "description": f"This issue matches your search query", "state": { "name": "Todo" }, "team": { "key": "ENG" }, "url": "https://linear.app/team/issue/ENG-1235" } ], "pageInfo": { "hasNextPage": False } } } }