Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |
| } | |
| } | |
| } | |
| } | |