Spaces:
Sleeping
Sleeping
| # ============================================================================ | |
| # PYDANTIC MODELS | |
| # ============================================================================ | |
| from pydantic import BaseModel, Field, validator, root_validator | |
| from typing import List, Optional, Dict, Any, Literal, Union | |
| from datetime import datetime | |
| from enum import Enum | |
| from .enums import TaskStatus, TaskType, TestCasePriority, TestCaseType, TestCaseCategory, OutputFormat | |
| from app.config import Config | |
| # ============================================================================ | |
| # FRONTEND-COMPATIBLE ENUMS (Match PM Tool Frontend) | |
| # ============================================================================ | |
| class TaskStatusFrontend(str, Enum): | |
| """Status values matching frontend test-case-form.tsx""" | |
| DRAFT = "Draft" | |
| READY = "Ready" | |
| IN_PROGRESS = "In Progress" | |
| BLOCKED = "Blocked" | |
| COMPLETED = "Completed" | |
| PASSED = "Passed" | |
| FAILED = "Failed" | |
| SKIPPED = "Skipped" | |
| class TaskPriorityFrontend(str, Enum): | |
| """Priority values matching frontend test-case-form.tsx""" | |
| CRITICAL = "Critical" | |
| HIGH = "High" | |
| MEDIUM = "Medium" | |
| LOW = "Low" | |
| class TaskSeverityFrontend(str, Enum): | |
| """Severity values matching frontend test-case-form.tsx""" | |
| BLOCKER = "Blocker" | |
| CRITICAL = "Critical" | |
| MAJOR = "Major" | |
| MINOR = "Minor" | |
| TRIVIAL = "Trivial" | |
| class TaskEnvironmentFrontend(str, Enum): | |
| """Environment values matching frontend test-case-form.tsx""" | |
| DEVELOPMENT = "Development" | |
| TESTING = "Testing" | |
| STAGING = "Staging" | |
| PRODUCTION = "Production" | |
| class Credentials(BaseModel): | |
| email: str = Field(..., description="PM Tool email") | |
| password: str = Field(..., description="PM Tool password") | |
| def validate_email(cls, v): | |
| if '@' not in v: | |
| raise ValueError('Invalid email format') | |
| return v.lower() | |
| class DateRange(BaseModel): | |
| from_date: Optional[str] = Field(None, alias="from", description="Start date (YYYY-MM-DD)") | |
| to_date: Optional[str] = Field(None, alias="to", description="End date (YYYY-MM-DD)") | |
| def validate_date(cls, v): | |
| if v: | |
| try: | |
| datetime.strptime(v, '%Y-%m-%d') | |
| except ValueError: | |
| raise ValueError('Date must be in YYYY-MM-DD format') | |
| return v | |
| # ============================================================================ | |
| # DOCUMENT PROCESSING MODELS | |
| # ============================================================================ | |
| class AttachmentConfig(BaseModel): | |
| """Configuration for individual attachment processing""" | |
| attachment_id: Optional[str] = Field(None, description="Attachment ID from PM Tool") | |
| filename: str = Field(..., description="Filename") | |
| skip: bool = Field(False, description="Skip processing this attachment") | |
| skip_reason: Optional[str] = Field(None, description="Reason for skipping") | |
| priority: int = Field(0, ge=0, le=10, description="Processing priority (higher = first)") | |
| max_tokens: Optional[int] = Field(None, description="Max tokens to use from this document") | |
| class DocumentProcessingConfig(BaseModel): | |
| """Global document processing configuration""" | |
| enabled: bool = Field(True, description="Enable document processing") | |
| max_total_tokens: int = Field(6000, ge=500, le=32000, description="Max total tokens for document context") | |
| max_chunk_tokens: int = Field(1500, ge=200, le=4000, description="Max tokens per chunk") | |
| overlap_tokens: int = Field(100, ge=0, le=500, description="Overlap between chunks") | |
| include_metadata: bool = Field(True, description="Include document metadata in context") | |
| auto_download: bool = Field(True, description="Auto-download attachments from PM Tool") | |
| download_dir: Optional[str] = Field(None, description="Directory to save downloaded files") | |
| class ProjectSelection(BaseModel): | |
| """Project selection options""" | |
| project_id: Optional[int] = Field(None, description="Project ID") | |
| project_name: Optional[str] = Field(None, description="Project name (auto-resolved)") | |
| include_all_tasks: bool = Field(False, description="Include all tasks from project") | |
| class GenerateRequest(BaseModel): | |
| # Authentication | |
| credentials: Credentials | |
| # Task Selection (provide at least one) | |
| task_id: Optional[int] = Field(None, description="Single task ID") | |
| task_ids: Optional[List[int]] = Field(None, description="Multiple task IDs") | |
| created_by_name: Optional[str] = Field(None, description="Filter by task creator") | |
| project_selection: Optional[ProjectSelection] = Field(None, description="Project selection options") | |
| # Legacy project_id for backward compatibility | |
| project_id: Optional[int] = Field(None, description="All tasks in project (legacy)") | |
| # Filters | |
| date_range: Optional[DateRange] = None | |
| status: Optional[List[TaskStatus]] = Field(None, description="Filter by task status") | |
| priority: Optional[List[TestCasePriority]] = None | |
| task_type: Optional[TaskType] = Field(TaskType.TASK, description="Task type filter") | |
| # Generation Settings | |
| test_case_count: int = Field(8, ge=1, le=15, description="Test cases per task") | |
| include_attachments: bool = Field(True, description="Fetch and analyze attachments") | |
| include_related_tasks: bool = Field(False, description="Cross-reference similar tasks") | |
| # Document Processing Settings | |
| document_processing: Optional[DocumentProcessingConfig] = Field( | |
| None, description="Document processing configuration" | |
| ) | |
| attachment_configs: Optional[List[AttachmentConfig]] = Field( | |
| None, description="Per-attachment processing config (skip, priority, etc.)" | |
| ) | |
| # AI Settings | |
| groq_model: str = Field(Config.GROQ_MODEL, description="Groq model to use") | |
| temperature: float = Field(Config.GROQ_TEMPERATURE, ge=0, le=1) | |
| # Output | |
| output_format: OutputFormat = Field(OutputFormat.JSON, description="Output format") | |
| include_metadata: bool = Field(True, description="Include generation metadata") | |
| def validate_task_selection(cls, values): | |
| # At least one selection method must be provided | |
| selections = [ | |
| values.get('task_id'), | |
| values.get('task_ids'), | |
| values.get('created_by_name'), | |
| values.get('project_id'), | |
| values.get('project_selection'), | |
| ] | |
| if not any(selections): | |
| raise ValueError('Provide at least one task selection method (task_id, task_ids, created_by_name, project_id, or project_selection)') | |
| return values | |
| class DocumentProcessingResult(BaseModel): | |
| """Result of document processing""" | |
| document_id: str | |
| filename: str | |
| processed: bool | |
| skipped: bool = False | |
| skip_reason: Optional[str] = None | |
| chunks_count: int = 0 | |
| tokens_used: int = 0 | |
| error: Optional[str] = None | |
| class TestStep(BaseModel): | |
| step_number: int | |
| action: str | |
| expected_result: str | |
| test_data: Optional[str] = None | |
| class TestCase(BaseModel): | |
| """Legacy test case format (kept for backward compatibility)""" | |
| test_case_id: str | |
| task_id: int | |
| task_code: str | |
| project_name: str | |
| issue_name: str | |
| feature_name: str | |
| task_name: str | |
| task_description: str | |
| created_by: str | |
| title: str | |
| objective: str | |
| preconditions: List[str] | |
| test_steps: List[TestStep] | |
| expected_outcome: str | |
| test_type: TestCaseType | |
| priority: TestCasePriority | |
| category: TestCaseCategory | |
| metadata: Optional[Dict[str, Any]] = None | |
| # ============================================================================ | |
| # FRONTEND-COMPATIBLE TEST CASE (Matches Task Interface) | |
| # ============================================================================ | |
| class FrontendTestCase(BaseModel): | |
| """ | |
| Test case format that matches the frontend Task interface exactly. | |
| This is what the frontend expects to save directly via tasksApi.create() | |
| Required fields (marked with * in frontend): | |
| - title: Test Case Title | |
| - description: Test Case Description | |
| All other fields are optional with sensible defaults. | |
| """ | |
| # Core Identity | |
| type: str = Field(default="Test Case", description="Always 'Test Case' for test cases") | |
| id: Optional[int] = Field(default=None, description="Auto-generated by PM Tool DB") | |
| # Required Fields | |
| title: str = Field(..., description="Test case title (REQUIRED)") | |
| description: str = Field(..., description="Test case description (REQUIRED)") | |
| # Linking | |
| issuesId: Optional[int] = Field(default=None, description="Feature/Issue ID this test case belongs to") | |
| projectId: Optional[int] = Field(default=None, description="Project ID") | |
| # Status & Priority | |
| status: str = Field(default="Draft", description="Test case status") | |
| priority: Optional[str] = Field(default="Medium", description="Priority: Critical, High, Medium, Low") | |
| severity: Optional[str] = Field(default="Minor", description="Severity: Blocker, Critical, Major, Minor, Trivial") | |
| # Test Details (Tab 2: Test Steps) | |
| stepsToReproduce: Optional[str] = Field(default=None, description="Numbered test steps as text") | |
| inputs: Optional[str] = Field(default=None, description="Test data/inputs") | |
| expectedBehavior: Optional[str] = Field(default=None, description="Expected result") | |
| actualBehavior: Optional[str] = Field(default="", description="Actual result (filled during testing)") | |
| # Technical Details (Tab 3) | |
| rootCause: Optional[str] = Field(default="", description="Root cause (filled after bug analysis)") | |
| resolution: Optional[str] = Field(default="", description="Resolution (filled after fix)") | |
| # Additional Fields | |
| environment: Optional[str] = Field(default="Staging", description="Test environment") | |
| version: Optional[str] = Field(default="", description="Version being tested") | |
| url: Optional[str] = Field(default="", description="Related URL") | |
| # Assignment & Dates | |
| assignedTo: Optional[int] = Field(default=None, description="Assigned employee ID") | |
| stateDate: Optional[str] = Field(default=None, description="Start date (ISO format)") | |
| endDate: Optional[str] = Field(default=None, description="End date (ISO format)") | |
| # Auto-generated fields | |
| taskCode: Optional[str] = Field(default=None, description="Auto-generated task code") | |
| createdAt: Optional[str] = Field(default=None, description="Creation timestamp") | |
| createdBy: Optional[int] = Field(default=None, description="Creator employee ID") | |
| createdByName: Optional[str] = Field(default=None, description="Creator name") | |
| # Related names (for display) | |
| projectName: Optional[str] = Field(default=None, description="Project name") | |
| issueName: Optional[str] = Field(default=None, description="Issue/Feature name") | |
| deliverableName: Optional[str] = Field(default=None, description="Deliverable/Feature name") | |
| # Sprint info | |
| sprintName: Optional[str] = Field(default=None, description="Sprint name") | |
| estimates: Optional[str] = Field(default="", description="Time estimate") | |
| esUnit: Optional[int] = Field(default=0, description="Estimate unit") | |
| remainingHr: Optional[int] = Field(default=0, description="Remaining hours") | |
| # Link back to original task (for reference) | |
| related_task_id: Optional[int] = Field(default=None, description="Original task ID this was generated from") | |
| related_task_code: Optional[str] = Field(default=None, description="Original task code") | |
| class Config: | |
| extra = "allow" # Allow additional fields from PM Tool | |
| class GenerateResponse(BaseModel): | |
| """Legacy response format""" | |
| success: bool | |
| message: str | |
| total_tasks_processed: int | |
| total_test_cases_generated: int | |
| failed_tasks: List[Dict[str, Any]] = [] | |
| test_cases: List[TestCase] | |
| metadata: Optional[Dict[str, Any]] = None | |
| download_url: Optional[str] = None | |
| class FrontendGenerateResponse(BaseModel): | |
| """ | |
| Response format for frontend integration. | |
| Returns test cases in Task format ready for tasksApi.create() | |
| """ | |
| success: bool | |
| message: str | |
| feature_id: Optional[int] = None | |
| feature_name: Optional[str] = None | |
| generated_at: Optional[str] = None | |
| test_cases: List[FrontendTestCase] | |
| metadata: Optional[Dict[str, Any]] = None | |
| # ============================================================================ | |
| # FRONTEND REQUEST SCHEMA | |
| # ============================================================================ | |
| class TaskInfo(BaseModel): | |
| """Task information from frontend""" | |
| id: int | |
| taskCode: Optional[str] = None | |
| title: str | |
| description: Optional[str] = None | |
| status: Optional[str] = None | |
| assignedTo: Optional[int] = None | |
| class FrontendGenerateRequest(BaseModel): | |
| """ | |
| Request format from frontend for feature-based test case generation. | |
| All dynamic data comes from the frontend's current state (issue-detail page). | |
| """ | |
| # Authentication (optional - can use session token) | |
| credentials: Optional[Credentials] = None | |
| session_token: Optional[str] = Field(None, description="PM Tool session token") | |
| # Feature Context (from issue state) | |
| feature_id: int = Field(..., description="Feature/Issue ID (required)") | |
| feature_title: Optional[str] = None | |
| feature_description: Optional[str] = None | |
| # Project Context | |
| project_id: Optional[int] = None | |
| project_name: Optional[str] = None | |
| # Deliverable Context | |
| deliverable_id: Optional[int] = None | |
| deliverable_name: Optional[str] = None | |
| # Tasks to generate test cases for | |
| tasks: List[TaskInfo] = Field(default=[], description="Tasks inside this feature") | |
| # Existing test cases to avoid duplicates | |
| existing_test_cases: Optional[List[Dict[str, str]]] = Field( | |
| default=None, | |
| description="Existing test cases (title, description) to avoid generating duplicates" | |
| ) | |
| # Additional context from user (text or extracted PDF content) | |
| additional_context: Optional[str] = Field( | |
| default=None, | |
| description="Additional context provided by user (text or PDF content) to enhance test case generation" | |
| ) | |
| # Generation Options | |
| test_case_count: int = Field(default=3, ge=1, le=10, description="Test cases per task") | |
| environment: str = Field(default="Staging", description="Test environment") | |
| include_attachments: bool = Field(default=True, description="Include document context") | |
| # AI Settings | |
| groq_model: Optional[str] = Field(None, description="Override default model") | |
| temperature: Optional[float] = Field(None, ge=0, le=1, description="Override temperature") | |
| def validate_environment(cls, v): | |
| valid = ['Development', 'Testing', 'Staging', 'Production'] | |
| if v not in valid: | |
| return 'Staging' | |
| return v | |
| class HealthResponse(BaseModel): | |
| status: str | |
| timestamp: str | |
| version: str | |
| groq_status: str | |
| pm_tool_status: str | |
| class ErrorResponse(BaseModel): | |
| error: str | |
| message: str | |
| details: Optional[Dict[str, Any]] = None | |
| timestamp: str | |