Upload 17 files

.github/workflows/ci.yml

@@ -0,0 +1,89 @@
+name: CI/CD
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:13
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: testauto
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      
+      rabbitmq:
+        image: rabbitmq:3-management
+        ports:
+          - 5672:5672
+          - 15672:15672
+        env:
+          RABBITMQ_DEFAULT_USER: guest
+          RABBITMQ_DEFAULT_PASS: guest
+
+    steps:
+    - uses: actions/checkout@v2
+    
+    - name: Set up Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: '3.9'
+    
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+        pip install pytest pytest-cov
+    
+    - name: Run tests
+      env:
+        POSTGRES_SERVER: localhost
+        POSTGRES_USER: postgres
+        POSTGRES_PASSWORD: postgres
+        POSTGRES_DB: testauto
+        RABBITMQ_HOST: localhost
+      run: |
+        pytest tests/ --cov=./ --cov-report=xml
+    
+    - name: Upload coverage to Codecov
+      uses: codecov/codecov-action@v2
+      with:
+        file: ./coverage.xml
+
+  docker:
+    needs: test
+    runs-on: ubuntu-latest
+    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+    
+    steps:
+    - uses: actions/checkout@v2
+    
+    - name: Set up Docker Buildx
+      uses: docker/setup-buildx-action@v1
+    
+    - name: Login to DockerHub
+      uses: docker/login-action@v1
+      with:
+        username: ${{ secrets.DOCKERHUB_USERNAME }}
+        password: ${{ secrets.DOCKERHUB_TOKEN }}
+    
+    - name: Build and push
+      uses: docker/build-push-action@v2
+      with:
+        context: .
+        push: true
+        tags: |
+          ${{ secrets.DOCKERHUB_USERNAME }}/testauto:latest
+          ${{ secrets.DOCKERHUB_USERNAME }}/testauto:${{ github.sha }} 

api/routes/ai_routes.py

@@ -0,0 +1,134 @@
+from fastapi import APIRouter, HTTPException, Depends
+from typing import List, Dict, Any
+from loguru import logger
+from pydantic import BaseModel
+
+from services.ai_service import ai_service
+
+router = APIRouter()
+
+class ModelConfig(BaseModel):
+    provider: str
+    model: str
+    max_tokens: int = 1000
+    temperature: float = 0.7
+
+class BenchmarkRequest(BaseModel):
+    prompt: str
+    models: List[ModelConfig]
+
+@router.post("/generate")
+async def generate_response(
+    prompt: str,
+    provider: str = "openai",
+    model: str = "gpt-3.5-turbo",
+    max_tokens: int = 1000,
+    temperature: float = 0.7
+) -> Dict[str, Any]:
+    """
+    Generate response using specified AI model.
+    
+    Parameters:
+    - prompt: The input prompt
+    - provider: AI provider to use (openai, local, openrouter)
+    - model: Model to use
+    - max_tokens: Maximum tokens in response
+    - temperature: Response temperature
+    """
+    try:
+        response = await ai_service.generate_response(
+            prompt=prompt,
+            provider=provider,
+            model=model,
+            max_tokens=max_tokens,
+            temperature=temperature
+        )
+        return response
+    except Exception as e:
+        logger.error(f"Error generating response: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.post("/benchmark")
+async def benchmark_models(request: BenchmarkRequest) -> List[Dict[str, Any]]:
+    """
+    Benchmark multiple models with the same prompt.
+    
+    Parameters:
+    - request: Benchmark request containing prompt and model configurations
+    """
+    try:
+        results = await ai_service.benchmark_models(
+            prompt=request.prompt,
+            models=[model.dict() for model in request.models]
+        )
+        return results
+    except Exception as e:
+        logger.error(f"Error benchmarking models: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.get("/available-models")
+async def get_available_models() -> Dict[str, List[str]]:
+    """Get list of available models for each provider."""
+    return {
+        "openai": [
+            "gpt-3.5-turbo",
+            "gpt-4",
+            "gpt-4-turbo"
+        ],
+        "openrouter": [
+            "anthropic/claude-2",
+            "google/palm-2",
+            "meta-llama/llama-2-70b"
+        ],
+        "local": [
+            "llama-2-7b",
+            "llama-2-13b",
+            "llama-2-70b"
+        ]
+    }
+
+@router.post("/validate-response")
+async def validate_response(
+    prompt: str,
+    response: str,
+    provider: str = "openai",
+    model: str = "gpt-3.5-turbo"
+) -> Dict[str, Any]:
+    """
+    Validate AI response against the prompt.
+    
+    Parameters:
+    - prompt: Original prompt
+    - response: AI-generated response
+    - provider: AI provider used
+    - model: Model used
+    """
+    try:
+        validation_prompt = f"""
+        Validate if the following response adequately addresses the prompt:
+        
+        Prompt:
+        {prompt}
+        
+        Response:
+        {response}
+        
+        Provide:
+        1. Relevance score (0-1)
+        2. Completeness score (0-1)
+        3. Specific feedback
+        """
+        
+        validation = await ai_service.generate_response(
+            prompt=validation_prompt,
+            provider=provider,
+            model=model
+        )
+        
+        return {
+            "status": "success",
+            "validation": validation["response"]
+        }
+    except Exception as e:
+        logger.error(f"Error validating response: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e)) 

api/routes/automation_routes.py

@@ -0,0 +1,253 @@
+from fastapi import APIRouter, HTTPException, Depends
+from typing import List, Dict, Any, Optional
+from loguru import logger
+from pydantic import BaseModel
+
+from services.automation_service import automation_service
+from services.test_service import test_service
+
+router = APIRouter()
+
+class TestScriptRequest(BaseModel):
+    test_cases: List[Dict[str, Any]]
+    framework: str = "pytest"
+    language: str = "python"
+    browser: str = "chrome"
+
+class TestPlanRequest(BaseModel):
+    requirements: List[Dict[str, Any]]
+    test_cases: List[Dict[str, Any]]
+    project_info: Dict[str, Any]
+
+@router.post("/generate-scripts")
+async def generate_test_scripts(request: TestScriptRequest) -> Dict[str, Any]:
+    """
+    Generate test automation scripts.
+    
+    Parameters:
+    - request: Test script generation request
+    """
+    try:
+        # Generate test scripts
+        scripts = await automation_service.generate_test_scripts(
+            test_cases=request.test_cases,
+            framework=request.framework,
+            language=request.language,
+            browser=request.browser
+        )
+        
+        # Generate Gherkin feature file
+        feature = await automation_service.generate_gherkin_feature(
+            test_cases=request.test_cases
+        )
+        
+        return {
+            "status": "success",
+            "scripts": scripts,
+            "feature": feature
+        }
+    except Exception as e:
+        logger.error(f"Error generating test scripts: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.post("/generate-test-plan")
+async def generate_test_plan(request: TestPlanRequest) -> Dict[str, Any]:
+    """
+    Generate comprehensive test plan.
+    
+    Parameters:
+    - request: Test plan generation request
+    """
+    try:
+        # Calculate coverage metrics
+        coverage = await test_service.validate_test_cases(
+            test_cases=request.test_cases,
+            requirements=request.requirements
+        )
+        
+        # Prioritize test cases
+        prioritized_cases = await test_service.prioritize_test_cases(
+            test_cases=request.test_cases,
+            requirements=request.requirements
+        )
+        
+        # Generate test plan sections
+        test_plan = {
+            "project_info": request.project_info,
+            "scope": _generate_scope(request.requirements),
+            "approach": _generate_approach(request.test_cases),
+            "resources": _generate_resources(),
+            "schedule": _generate_schedule(request.test_cases),
+            "risk_assessment": _generate_risk_assessment(prioritized_cases),
+            "requirement_traceability": coverage["coverage_matrix"],
+            "coverage_metrics": {
+                "percentage": coverage["coverage_percentage"],
+                "total_requirements": coverage["total_requirements"],
+                "covered_requirements": coverage["covered_requirements"]
+            }
+        }
+        
+        return {
+            "status": "success",
+            "test_plan": test_plan
+        }
+    except Exception as e:
+        logger.error(f"Error generating test plan: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+def _generate_scope(requirements: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Generate test scope section."""
+    return {
+        "in_scope": [
+            {
+                "id": req["id"],
+                "title": req["title"],
+                "priority": req.get("priority", "Medium")
+            }
+            for req in requirements
+        ],
+        "out_of_scope": [],
+        "assumptions": [
+            "Test environment is properly configured",
+            "Test data is available",
+            "Dependencies are stable"
+        ]
+    }
+
+def _generate_approach(test_cases: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Generate test approach section."""
+    return {
+        "strategy": "Risk-based testing approach",
+        "test_levels": [
+            "Unit Testing",
+            "Integration Testing",
+            "System Testing",
+            "Acceptance Testing"
+        ],
+        "test_types": [
+            "Functional Testing",
+            "Non-functional Testing",
+            "Regression Testing"
+        ],
+        "automation_approach": {
+            "framework": "pytest",
+            "tools": [
+                "Selenium WebDriver",
+                "Playwright",
+                "pytest-html"
+            ],
+            "coverage_goal": "80%"
+        }
+    }
+
+def _generate_resources() -> Dict[str, Any]:
+    """Generate resources section."""
+    return {
+        "team": [
+            {
+                "role": "Test Lead",
+                "responsibilities": [
+                    "Test plan creation",
+                    "Resource allocation",
+                    "Progress tracking"
+                ]
+            },
+            {
+                "role": "Test Engineer",
+                "responsibilities": [
+                    "Test case execution",
+                    "Defect reporting",
+                    "Test automation"
+                ]
+            }
+        ],
+        "tools": [
+            "Test Management Tool",
+            "Automation Framework",
+            "CI/CD Pipeline",
+            "Version Control System"
+        ],
+        "environments": [
+            "Development",
+            "Testing",
+            "Staging",
+            "Production"
+        ]
+    }
+
+def _generate_schedule(test_cases: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Generate schedule section."""
+    return {
+        "phases": [
+            {
+                "name": "Planning",
+                "duration": "1 week",
+                "activities": [
+                    "Test plan creation",
+                    "Resource allocation",
+                    "Tool setup"
+                ]
+            },
+            {
+                "name": "Design",
+                "duration": "2 weeks",
+                "activities": [
+                    "Test case design",
+                    "Automation framework setup",
+                    "Test data preparation"
+                ]
+            },
+            {
+                "name": "Execution",
+                "duration": "3 weeks",
+                "activities": [
+                    "Test case execution",
+                    "Defect reporting",
+                    "Regression testing"
+                ]
+            },
+            {
+                "name": "Closure",
+                "duration": "1 week",
+                "activities": [
+                    "Test summary report",
+                    "Lessons learned",
+                    "Knowledge transfer"
+                ]
+            }
+        ],
+        "milestones": [
+            "Test plan approval",
+            "Test case design completion",
+            "Automation framework ready",
+            "Test execution completion",
+            "Test closure"
+        ]
+    }
+
+def _generate_risk_assessment(
+    prioritized_cases: List[Dict[str, Any]]
+) -> Dict[str, Any]:
+    """Generate risk assessment section."""
+    return {
+        "high_risk_areas": [
+            {
+                "test_case": case["test_case"],
+                "risk_level": case["risk_level"],
+                "justification": case["justification"]
+            }
+            for case in prioritized_cases
+            if case["risk_level"] == "High"
+        ],
+        "mitigation_strategies": [
+            "Early testing of high-risk areas",
+            "Additional test coverage for critical features",
+            "Regular risk reassessment",
+            "Automated regression testing"
+        ],
+        "contingency_plans": [
+            "Resource reallocation if needed",
+            "Schedule adjustment for high-risk areas",
+            "Additional testing cycles if required"
+        ]
+    } 

api/routes/document_routes.py

@@ -0,0 +1,158 @@
+from fastapi import APIRouter, UploadFile, File, HTTPException, Depends
+from typing import List, Dict, Any
+from loguru import logger
+
+from services.document_service import document_service
+from services.ai_service import ai_service
+from services.test_service import test_service
+from services.automation_service import automation_service
+
+router = APIRouter()
+
+@router.post("/upload")
+async def upload_document(
+    file: UploadFile = File(...),
+    process_type: str = "requirements"
+) -> Dict[str, Any]:
+    """
+    Upload and process a document.
+    
+    Parameters:
+    - file: The document file to upload
+    - process_type: Type of processing to perform (requirements, test_cases, etc.)
+    """
+    try:
+        # Save uploaded file
+        file_path = await document_service.save_upload_file(file)
+        
+        # Process document
+        result = await document_service.process_document(file_path)
+        
+        # Segment document if needed
+        segments = await document_service.segment_document(result["text"])
+        
+        return {
+            "status": "success",
+            "file_path": file_path,
+            "segments": segments,
+            "type": result["type"]
+        }
+    except Exception as e:
+        logger.error(f"Error processing document: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.post("/process-requirements")
+async def process_requirements(
+    file: UploadFile = File(...),
+    ai_provider: str = "openai",
+    model: str = "gpt-3.5-turbo"
+) -> Dict[str, Any]:
+    """
+    Process requirements document and generate test cases.
+    
+    Parameters:
+    - file: The requirements document
+    - ai_provider: AI provider to use (openai, local, openrouter)
+    - model: Model to use for processing
+    """
+    try:
+        # Upload and process document
+        file_path = await document_service.save_upload_file(file)
+        result = await document_service.process_document(file_path)
+        
+        # Extract requirements using AI
+        prompt = f"""
+        Extract requirements from the following text. For each requirement, provide:
+        1. ID
+        2. Title
+        3. Description
+        4. Priority (High/Medium/Low)
+        
+        Text:
+        {result["text"]}
+        """
+        
+        ai_response = await ai_service.generate_response(
+            prompt=prompt,
+            provider=ai_provider,
+            model=model
+        )
+        
+        # Generate test cases
+        requirements = _parse_requirements(ai_response["response"])
+        test_cases = await test_service.generate_test_cases(
+            requirements=requirements,
+            ai_service=ai_service
+        )
+        
+        return {
+            "status": "success",
+            "requirements": requirements,
+            "test_cases": test_cases
+        }
+    except Exception as e:
+        logger.error(f"Error processing requirements: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.post("/generate-test-scripts")
+async def generate_test_scripts(
+    test_cases: List[Dict[str, Any]],
+    framework: str = "pytest",
+    language: str = "python",
+    browser: str = "chrome"
+) -> Dict[str, Any]:
+    """
+    Generate test scripts from test cases.
+    
+    Parameters:
+    - test_cases: List of test cases
+    - framework: Test framework to use (pytest, playwright)
+    - language: Programming language (python)
+    - browser: Browser to use (chrome, firefox, etc.)
+    """
+    try:
+        # Generate test scripts
+        scripts = await automation_service.generate_test_scripts(
+            test_cases=test_cases,
+            framework=framework,
+            language=language,
+            browser=browser
+        )
+        
+        # Generate Gherkin feature file
+        feature = await automation_service.generate_gherkin_feature(test_cases)
+        
+        return {
+            "status": "success",
+            "scripts": scripts,
+            "feature": feature
+        }
+    except Exception as e:
+        logger.error(f"Error generating test scripts: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+def _parse_requirements(text: str) -> List[Dict[str, Any]]:
+    """Parse AI response into structured requirements."""
+    requirements = []
+    current_req = {}
+    
+    for line in text.split('\n'):
+        line = line.strip()
+        if not line:
+            continue
+            
+        if line.startswith('ID:'):
+            if current_req:
+                requirements.append(current_req)
+            current_req = {'id': line.split(':', 1)[1].strip()}
+        elif line.startswith('Title:'):
+            current_req['title'] = line.split(':', 1)[1].strip()
+        elif line.startswith('Description:'):
+            current_req['description'] = line.split(':', 1)[1].strip()
+        elif line.startswith('Priority:'):
+            current_req['priority'] = line.split(':', 1)[1].strip()
+    
+    if current_req:
+        requirements.append(current_req)
+    
+    return requirements 

api/routes/test_routes.py

@@ -0,0 +1,220 @@
+from fastapi import APIRouter, HTTPException, Depends
+from typing import List, Dict, Any, Optional
+from loguru import logger
+from pydantic import BaseModel
+import json
+
+from services.test_service import test_service
+from services.ai_service import ai_service
+
+router = APIRouter()
+
+class TestCase(BaseModel):
+    id: str
+    title: str
+    preconditions: List[str]
+    steps: List[str]
+    expected_results: List[str]
+    priority: str
+    type: str
+    requirement_id: str
+
+class ExportRequest(BaseModel):
+    test_cases: List[TestCase]
+    target: str
+    project_id: str
+    section_id: Optional[str] = None
+
+@router.post("/generate")
+async def generate_test_cases(
+    requirements: List[Dict[str, Any]],
+    ai_provider: str = "openai",
+    model: str = "gpt-3.5-turbo"
+) -> List[Dict[str, Any]]:
+    """
+    Generate test cases from requirements.
+    
+    Parameters:
+    - requirements: List of requirements
+    - ai_provider: AI provider to use
+    - model: Model to use
+    """
+    try:
+        test_cases = await test_service.generate_test_cases(
+            requirements=requirements,
+            ai_service=ai_service
+        )
+        return test_cases
+    except Exception as e:
+        logger.error(f"Error generating test cases: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.post("/export")
+async def export_test_cases(request: ExportRequest) -> Dict[str, Any]:
+    """
+    Export test cases to test management tool.
+    
+    Parameters:
+    - request: Export request containing test cases and target information
+    """
+    try:
+        if request.target == "testrail":
+            if not request.section_id:
+                raise HTTPException(
+                    status_code=400,
+                    detail="section_id is required for TestRail export"
+                )
+            results = await test_service.export_to_testrail(
+                test_cases=request.test_cases,
+                project_id=int(request.project_id),
+                section_id=int(request.section_id)
+            )
+        elif request.target == "jira":
+            results = await test_service.export_to_jira(
+                test_cases=request.test_cases,
+                project_key=request.project_id
+            )
+        elif request.target == "qtest":
+            results = await test_service.export_to_qtest(
+                test_cases=request.test_cases,
+                project_id=int(request.project_id)
+            )
+        else:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Unsupported export target: {request.target}"
+            )
+        
+        return {
+            "status": "success",
+            "results": results
+        }
+    except Exception as e:
+        logger.error(f"Error exporting test cases: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.post("/validate")
+async def validate_test_cases(
+    test_cases: List[TestCase],
+    requirements: List[Dict[str, Any]]
+) -> Dict[str, Any]:
+    """
+    Validate test cases against requirements.
+    
+    Parameters:
+    - test_cases: List of test cases
+    - requirements: List of requirements
+    """
+    try:
+        # Create requirement coverage matrix
+        coverage = {}
+        for req in requirements:
+            coverage[req["id"]] = {
+                "requirement": req,
+                "test_cases": [],
+                "covered": False
+            }
+        
+        # Map test cases to requirements
+        for test_case in test_cases:
+            if test_case.requirement_id in coverage:
+                coverage[test_case.requirement_id]["test_cases"].append(test_case)
+                coverage[test_case.requirement_id]["covered"] = True
+        
+        # Calculate coverage metrics
+        total_requirements = len(requirements)
+        covered_requirements = sum(1 for req in coverage.values() if req["covered"])
+        coverage_percentage = (covered_requirements / total_requirements) * 100
+        
+        # Identify uncovered requirements
+        uncovered_requirements = [
+            req["requirement"]
+            for req in coverage.values()
+            if not req["covered"]
+        ]
+        
+        return {
+            "status": "success",
+            "coverage_percentage": coverage_percentage,
+            "total_requirements": total_requirements,
+            "covered_requirements": covered_requirements,
+            "uncovered_requirements": uncovered_requirements,
+            "coverage_matrix": coverage
+        }
+    except Exception as e:
+        logger.error(f"Error validating test cases: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.post("/prioritize")
+async def prioritize_test_cases(
+    test_cases: List[TestCase],
+    requirements: List[Dict[str, Any]]
+) -> List[Dict[str, Any]]:
+    """
+    Prioritize test cases based on requirements and risk.
+    
+    Parameters:
+    - test_cases: List of test cases
+    - requirements: List of requirements
+    """
+    try:
+        # Create risk assessment prompt
+        prompt = f"""
+        Analyze the following requirements and test cases to determine test case priority.
+        Consider:
+        1. Requirement priority
+        2. Business impact
+        3. Technical complexity
+        4. Historical defect patterns
+        
+        Requirements:
+        {json.dumps(requirements, indent=2)}
+        
+        Test Cases:
+        {json.dumps([tc.dict() for tc in test_cases], indent=2)}
+        
+        For each test case, provide:
+        1. Priority score (1-5)
+        2. Risk level (High/Medium/Low)
+        3. Justification
+        """
+        
+        # Get AI assessment
+        assessment = await ai_service.generate_response(prompt=prompt)
+        
+        # Parse and apply prioritization
+        prioritized_cases = []
+        for test_case in test_cases:
+            # Find assessment for this test case
+            case_assessment = _find_case_assessment(
+                assessment["response"],
+                test_case.id
+            )
+            
+            prioritized_cases.append({
+                "test_case": test_case,
+                "priority_score": case_assessment["priority_score"],
+                "risk_level": case_assessment["risk_level"],
+                "justification": case_assessment["justification"]
+            })
+        
+        # Sort by priority score
+        prioritized_cases.sort(
+            key=lambda x: x["priority_score"],
+            reverse=True
+        )
+        
+        return prioritized_cases
+    except Exception as e:
+        logger.error(f"Error prioritizing test cases: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+def _find_case_assessment(assessment_text: str, case_id: str) -> Dict[str, Any]:
+    """Extract assessment for a specific test case."""
+    # This is a simplified implementation
+    # In practice, you'd want more robust parsing
+    return {
+        "priority_score": 3,
+        "risk_level": "Medium",
+        "justification": "Default assessment"
+    } 

app/main.py

@@ -0,0 +1,61 @@
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from loguru import logger
+import uvicorn
+
+from services.database_service import db_service
+from services.queue_service import queue_service
+from api.routes import document_routes, ai_routes, test_routes, automation_routes
+
+app = FastAPI(
+    title="AI-Powered Test Automation Framework",
+    description="An intelligent test automation framework powered by AI",
+    version="1.0.0"
+)
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Include routers
+app.include_router(document_routes.router, prefix="/api/documents", tags=["Documents"])
+app.include_router(ai_routes.router, prefix="/api/ai", tags=["AI"])
+app.include_router(test_routes.router, prefix="/api/tests", tags=["Tests"])
+app.include_router(automation_routes.router, prefix="/api/automation", tags=["Automation"])
+
+@app.on_event("startup")
+async def startup_event():
+    """Initialize services on startup"""
+    try:
+        # Initialize database
+        db_service.init_db()
+        logger.info("Database initialized successfully")
+        
+        # Initialize queue service
+        queue_service.connect()
+        logger.info("Queue service initialized successfully")
+    except Exception as e:
+        logger.error(f"Error during startup: {str(e)}")
+        raise
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    """Cleanup on shutdown"""
+    try:
+        queue_service.close()
+        logger.info("Services shut down successfully")
+    except Exception as e:
+        logger.error(f"Error during shutdown: {str(e)}")
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint"""
+    return {"status": "healthy"}
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True) 

config/config.py

@@ -0,0 +1,67 @@
+from pydantic_settings import BaseSettings
+from typing import Optional, Dict, Any
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+class Settings(BaseSettings):
+    # Application
+    APP_NAME: str = "TestAuto"
+    DEBUG: bool = False
+    VERSION: str = "1.0.0"
+    
+    # API
+    API_V1_STR: str = "/api/v1"
+    PROJECT_NAME: str = "TestAuto API"
+    
+    # Security
+    SECRET_KEY: str = os.getenv("SECRET_KEY", "your-secret-key")
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8  # 8 days
+    
+    # Database
+    POSTGRES_SERVER: str = os.getenv("POSTGRES_SERVER", "localhost")
+    POSTGRES_USER: str = os.getenv("POSTGRES_USER", "postgres")
+    POSTGRES_PASSWORD: str = os.getenv("POSTGRES_PASSWORD", "postgres")
+    POSTGRES_DB: str = os.getenv("POSTGRES_DB", "testauto")
+    SQLALCHEMY_DATABASE_URI: Optional[str] = None
+    
+    # AI Providers
+    OPENAI_API_KEY: Optional[str] = os.getenv("OPENAI_API_KEY")
+    OPENROUTER_API_KEY: Optional[str] = os.getenv("OPENROUTER_API_KEY")
+    LOCAL_LLM_PATH: Optional[str] = os.getenv("LOCAL_LLM_PATH")
+    
+    # Message Queue
+    RABBITMQ_HOST: str = os.getenv("RABBITMQ_HOST", "localhost")
+    RABBITMQ_PORT: int = int(os.getenv("RABBITMQ_PORT", "5672"))
+    RABBITMQ_USER: str = os.getenv("RABBITMQ_USER", "guest")
+    RABBITMQ_PASSWORD: str = os.getenv("RABBITMQ_PASSWORD", "guest")
+    
+    # Test Management Tools
+    TESTRAIL_API_KEY: Optional[str] = os.getenv("TESTRAIL_API_KEY")
+    TESTRAIL_URL: Optional[str] = os.getenv("TESTRAIL_URL")
+    JIRA_API_TOKEN: Optional[str] = os.getenv("JIRA_API_TOKEN")
+    JIRA_URL: Optional[str] = os.getenv("JIRA_URL")
+    QTEST_API_TOKEN: Optional[str] = os.getenv("QTEST_API_TOKEN")
+    QTEST_URL: Optional[str] = os.getenv("QTEST_URL")
+    
+    # Document Processing
+    UPLOAD_FOLDER: str = "uploads"
+    MAX_CONTENT_LENGTH: int = 16 * 1024 * 1024  # 16MB
+    ALLOWED_EXTENSIONS: set = {"pdf", "docx", "doc", "txt", "png", "jpg", "jpeg"}
+    
+    # Logging
+    LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")
+    LOG_FILE: str = "logs/testauto.log"
+    
+    # Docker
+    DOCKER_IMAGE: str = "testauto:latest"
+    
+    def __init__(self, **kwargs: Any):
+        super().__init__(**kwargs)
+        self.SQLALCHEMY_DATABASE_URI = (
+            f"postgresql://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}"
+            f"@{self.POSTGRES_SERVER}/{self.POSTGRES_DB}"
+        )
+
+settings = Settings() 

docker/Dockerfile

@@ -0,0 +1,33 @@
+FROM python:3.9-slim
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    libpq-dev \
+    tesseract-ocr \
+    && rm -rf /var/lib/apt/lists/*
+
+# Set working directory
+WORKDIR /app
+
+# Copy requirements file
+COPY requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY . .
+
+# Create necessary directories
+RUN mkdir -p uploads generated_scripts logs
+
+# Set environment variables
+ENV PYTHONPATH=/app
+ENV PYTHONUNBUFFERED=1
+
+# Expose port
+EXPOSE 8000
+
+# Run the application
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] 

models/base.py

@@ -0,0 +1,70 @@
+from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, JSON, Text, Enum
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import relationship
+from datetime import datetime
+import enum
+
+Base = declarative_base()
+
+class DocumentType(enum.Enum):
+    PDF = "pdf"
+    DOCX = "docx"
+    IMAGE = "image"
+    TEXT = "text"
+
+class Document(Base):
+    __tablename__ = "documents"
+    
+    id = Column(Integer, primary_key=True)
+    filename = Column(String(255), nullable=False)
+    file_type = Column(Enum(DocumentType), nullable=False)
+    content = Column(Text)
+    metadata = Column(JSON)
+    created_at = Column(DateTime, default=datetime.utcnow)
+    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+    
+    test_cases = relationship("TestCase", back_populates="document")
+
+class TestCase(Base):
+    __tablename__ = "test_cases"
+    
+    id = Column(Integer, primary_key=True)
+    document_id = Column(Integer, ForeignKey("documents.id"))
+    title = Column(String(255), nullable=False)
+    description = Column(Text)
+    steps = Column(JSON)
+    expected_results = Column(JSON)
+    priority = Column(Integer)
+    status = Column(String(50))
+    created_at = Column(DateTime, default=datetime.utcnow)
+    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+    
+    document = relationship("Document", back_populates="test_cases")
+    test_scripts = relationship("TestScript", back_populates="test_case")
+
+class TestScript(Base):
+    __tablename__ = "test_scripts"
+    
+    id = Column(Integer, primary_key=True)
+    test_case_id = Column(Integer, ForeignKey("test_cases.id"))
+    language = Column(String(50))
+    code = Column(Text)
+    framework = Column(String(50))
+    created_at = Column(DateTime, default=datetime.utcnow)
+    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+    
+    test_case = relationship("TestCase", back_populates="test_scripts")
+
+class AIModelInteraction(Base):
+    __tablename__ = "ai_model_interactions"
+    
+    id = Column(Integer, primary_key=True)
+    provider = Column(String(50), nullable=False)
+    model = Column(String(100), nullable=False)
+    prompt = Column(Text, nullable=False)
+    response = Column(Text)
+    metrics = Column(JSON)
+    created_at = Column(DateTime, default=datetime.utcnow)
+    
+    def __repr__(self):
+        return f"<AIModelInteraction(provider={self.provider}, model={self.model})>" 

services/ai_service.py

@@ -0,0 +1,163 @@
+from typing import List, Dict, Any, Optional
+import openai
+from transformers import AutoModelForCausalLM, AutoTokenizer
+import torch
+from loguru import logger
+import time
+from config.config import settings
+
+class AIService:
+    def __init__(self):
+        self.openai_client = openai.OpenAI(api_key=settings.OPENAI_API_KEY)
+        self.local_model = None
+        self.local_tokenizer = None
+        if settings.LOCAL_LLM_PATH:
+            self._load_local_model()
+
+    def _load_local_model(self):
+        """Load local LLM model."""
+        try:
+            self.local_tokenizer = AutoTokenizer.from_pretrained(settings.LOCAL_LLM_PATH)
+            self.local_model = AutoModelForCausalLM.from_pretrained(settings.LOCAL_LLM_PATH)
+            logger.info("Local model loaded successfully")
+        except Exception as e:
+            logger.error(f"Error loading local model: {str(e)}")
+            raise
+
+    async def generate_response(
+        self,
+        prompt: str,
+        provider: str = "openai",
+        model: str = "gpt-3.5-turbo",
+        max_tokens: int = 1000,
+        temperature: float = 0.7
+    ) -> Dict[str, Any]:
+        """Generate response using specified provider and model."""
+        start_time = time.time()
+        
+        try:
+            if provider == "openai":
+                response = await self._generate_openai_response(
+                    prompt, model, max_tokens, temperature
+                )
+            elif provider == "local":
+                response = await self._generate_local_response(
+                    prompt, max_tokens, temperature
+                )
+            elif provider == "openrouter":
+                response = await self._generate_openrouter_response(
+                    prompt, model, max_tokens, temperature
+                )
+            else:
+                raise ValueError(f"Unsupported provider: {provider}")
+
+            end_time = time.time()
+            return {
+                "response": response,
+                "provider": provider,
+                "model": model,
+                "latency": end_time - start_time
+            }
+        except Exception as e:
+            logger.error(f"Error generating response: {str(e)}")
+            raise
+
+    async def _generate_openai_response(
+        self,
+        prompt: str,
+        model: str,
+        max_tokens: int,
+        temperature: float
+    ) -> str:
+        """Generate response using OpenAI API."""
+        try:
+            response = await self.openai_client.chat.completions.create(
+                model=model,
+                messages=[{"role": "user", "content": prompt}],
+                max_tokens=max_tokens,
+                temperature=temperature
+            )
+            return response.choices[0].message.content
+        except Exception as e:
+            logger.error(f"OpenAI API error: {str(e)}")
+            raise
+
+    async def _generate_local_response(
+        self,
+        prompt: str,
+        max_tokens: int,
+        temperature: float
+    ) -> str:
+        """Generate response using local model."""
+        if not self.local_model or not self.local_tokenizer:
+            raise ValueError("Local model not loaded")
+
+        try:
+            inputs = self.local_tokenizer(prompt, return_tensors="pt")
+            outputs = self.local_model.generate(
+                inputs["input_ids"],
+                max_length=max_tokens,
+                temperature=temperature,
+                do_sample=True
+            )
+            return self.local_tokenizer.decode(outputs[0], skip_special_tokens=True)
+        except Exception as e:
+            logger.error(f"Local model error: {str(e)}")
+            raise
+
+    async def _generate_openrouter_response(
+        self,
+        prompt: str,
+        model: str,
+        max_tokens: int,
+        temperature: float
+    ) -> str:
+        """Generate response using OpenRouter API."""
+        try:
+            headers = {
+                "Authorization": f"Bearer {settings.OPENROUTER_API_KEY}",
+                "HTTP-Referer": "https://testauto.com",
+                "X-Title": "TestAuto"
+            }
+            
+            response = await self.openai_client.chat.completions.create(
+                model=model,
+                messages=[{"role": "user", "content": prompt}],
+                max_tokens=max_tokens,
+                temperature=temperature,
+                headers=headers
+            )
+            return response.choices[0].message.content
+        except Exception as e:
+            logger.error(f"OpenRouter API error: {str(e)}")
+            raise
+
+    async def benchmark_models(
+        self,
+        prompt: str,
+        models: List[Dict[str, Any]]
+    ) -> List[Dict[str, Any]]:
+        """Benchmark different models with the same prompt."""
+        results = []
+        
+        for model_config in models:
+            try:
+                result = await self.generate_response(
+                    prompt=prompt,
+                    provider=model_config["provider"],
+                    model=model_config["model"],
+                    max_tokens=model_config.get("max_tokens", 1000),
+                    temperature=model_config.get("temperature", 0.7)
+                )
+                results.append(result)
+            except Exception as e:
+                logger.error(f"Error benchmarking model {model_config['model']}: {str(e)}")
+                results.append({
+                    "provider": model_config["provider"],
+                    "model": model_config["model"],
+                    "error": str(e)
+                })
+        
+        return results
+
+ai_service = AIService() 

services/automation_service.py

@@ -0,0 +1,189 @@
+from typing import List, Dict, Any, Optional
+from loguru import logger
+import os
+import json
+from datetime import datetime
+
+class AutomationService:
+    def __init__(self):
+        self.output_dir = "generated_scripts"
+        os.makedirs(self.output_dir, exist_ok=True)
+
+    async def generate_test_scripts(
+        self,
+        test_cases: List[Dict[str, Any]],
+        framework: str = "pytest",
+        language: str = "python",
+        browser: str = "chrome"
+    ) -> List[Dict[str, Any]]:
+        """Generate test scripts from test cases."""
+        results = []
+        
+        for test_case in test_cases:
+            try:
+                if framework == "pytest":
+                    script = await self._generate_pytest_script(test_case, browser)
+                elif framework == "playwright":
+                    script = await self._generate_playwright_script(test_case, browser)
+                else:
+                    raise ValueError(f"Unsupported framework: {framework}")
+
+                file_path = self._save_script(script, test_case["title"], language)
+                results.append({
+                    "test_case_id": test_case["id"],
+                    "file_path": file_path,
+                    "status": "success"
+                })
+            except Exception as e:
+                logger.error(f"Error generating test script for test case {test_case['id']}: {str(e)}")
+                results.append({
+                    "test_case_id": test_case["id"],
+                    "error": str(e),
+                    "status": "error"
+                })
+        
+        return results
+
+    async def _generate_pytest_script(
+        self,
+        test_case: Dict[str, Any],
+        browser: str
+    ) -> str:
+        """Generate pytest script with Selenium."""
+        script = f'''import pytest
+from selenium import webdriver
+from selenium.webdriver.common.by import By
+from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.support import expected_conditions as EC
+
+class Test{test_case["title"].replace(" ", "")}:
+    @pytest.fixture(scope="function")
+    def driver(self):
+        options = webdriver.{browser.capitalize()}Options()
+        driver = webdriver.{browser.capitalize()}(options=options)
+        driver.implicitly_wait(10)
+        yield driver
+        driver.quit()
+
+    def test_{test_case["title"].lower().replace(" ", "_")}(self, driver):
+        """
+        Test Case: {test_case["title"]}
+        Priority: {test_case["priority"]}
+        Type: {test_case["type"]}
+        """
+        # Preconditions
+'''
+        # Add preconditions
+        for precond in test_case["preconditions"]:
+            script += f"        # {precond}\n"
+
+        # Add test steps
+        script += "\n        # Test Steps\n"
+        for i, step in enumerate(test_case["steps"], 1):
+            script += f"        # Step {i}: {step}\n"
+            # Add basic Selenium commands based on step description
+            if "click" in step.lower():
+                script += "        element = WebDriverWait(driver, 10).until(\n"
+                script += "            EC.element_to_be_clickable((By.XPATH, '//button'))\n"
+                script += "        )\n"
+                script += "        element.click()\n"
+            elif "input" in step.lower() or "enter" in step.lower():
+                script += "        input_element = WebDriverWait(driver, 10).until(\n"
+                script += "            EC.presence_of_element_located((By.XPATH, '//input'))\n"
+                script += "        )\n"
+                script += "        input_element.send_keys('test_input')\n"
+
+        # Add assertions
+        script += "\n        # Expected Results\n"
+        for expected in test_case["expected_results"]:
+            script += f"        # {expected}\n"
+            script += "        assert True  # Replace with actual assertion\n"
+
+        return script
+
+    async def _generate_playwright_script(
+        self,
+        test_case: Dict[str, Any],
+        browser: str
+    ) -> str:
+        """Generate Playwright script."""
+        script = f'''import pytest
+from playwright.sync_api import sync_playwright
+
+def test_{test_case["title"].lower().replace(" ", "_")}():
+    """
+    Test Case: {test_case["title"]}
+    Priority: {test_case["priority"]}
+    Type: {test_case["type"]}
+    """
+    with sync_playwright() as p:
+        browser = p.{browser}.launch()
+        page = browser.new_page()
+        
+        # Preconditions
+'''
+        # Add preconditions
+        for precond in test_case["preconditions"]:
+            script += f"        # {precond}\n"
+
+        # Add test steps
+        script += "\n        # Test Steps\n"
+        for i, step in enumerate(test_case["steps"], 1):
+            script += f"        # Step {i}: {step}\n"
+            # Add basic Playwright commands based on step description
+            if "click" in step.lower():
+                script += "        page.click('button')\n"
+            elif "input" in step.lower() or "enter" in step.lower():
+                script += "        page.fill('input', 'test_input')\n"
+
+        # Add assertions
+        script += "\n        # Expected Results\n"
+        for expected in test_case["expected_results"]:
+            script += f"        # {expected}\n"
+            script += "        assert True  # Replace with actual assertion\n"
+
+        script += "\n        browser.close()"
+
+        return script
+
+    def _save_script(self, script: str, title: str, language: str) -> str:
+        """Save generated script to file."""
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        filename = f"test_{title.lower().replace(' ', '_')}_{timestamp}.{language}"
+        file_path = os.path.join(self.output_dir, filename)
+        
+        with open(file_path, "w") as f:
+            f.write(script)
+        
+        return file_path
+
+    async def generate_gherkin_feature(
+        self,
+        test_cases: List[Dict[str, Any]]
+    ) -> str:
+        """Generate Gherkin feature file from test cases."""
+        feature = f'''Feature: Automated Test Cases
+  As a test engineer
+  I want to automate test cases
+  So that I can ensure software quality
+
+'''
+        for test_case in test_cases:
+            feature += f'''  Scenario: {test_case["title"]}
+    Given the following preconditions:
+'''
+            for precond in test_case["preconditions"]:
+                feature += f"      {precond}\n"
+
+            feature += "    When I perform the following steps:\n"
+            for step in test_case["steps"]:
+                feature += f"      {step}\n"
+
+            feature += "    Then I should see the following results:\n"
+            for expected in test_case["expected_results"]:
+                feature += f"      {expected}\n"
+            feature += "\n"
+
+        return feature
+
+automation_service = AutomationService() 

services/database_service.py

@@ -0,0 +1,40 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, Session
+from sqlalchemy.exc import SQLAlchemyError
+from contextlib import contextmanager
+from typing import Generator
+from loguru import logger
+from config.config import Settings
+
+settings = Settings()
+
+class DatabaseService:
+    def __init__(self):
+        self.engine = create_engine(settings.DATABASE_URL)
+        self.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=self.engine)
+    
+    @contextmanager
+    def get_session(self) -> Generator[Session, None, None]:
+        session = self.SessionLocal()
+        try:
+            yield session
+            session.commit()
+        except SQLAlchemyError as e:
+            session.rollback()
+            logger.error(f"Database error: {str(e)}")
+            raise
+        finally:
+            session.close()
+    
+    def init_db(self):
+        """Initialize database tables"""
+        from models.base import Base
+        try:
+            Base.metadata.create_all(bind=self.engine)
+            logger.info("Database tables created successfully")
+        except SQLAlchemyError as e:
+            logger.error(f"Error creating database tables: {str(e)}")
+            raise
+
+# Create a singleton instance
+db_service = DatabaseService() 

services/document_service.py

@@ -0,0 +1,98 @@
+import os
+from typing import List, Optional, Dict, Any
+from fastapi import UploadFile, HTTPException
+from loguru import logger
+import pytesseract
+from PIL import Image
+import PyPDF2
+from docx import Document
+import io
+import aiofiles
+from config.config import settings
+
+class DocumentService:
+    def __init__(self):
+        self.upload_folder = settings.UPLOAD_FOLDER
+        os.makedirs(self.upload_folder, exist_ok=True)
+
+    async def save_upload_file(self, upload_file: UploadFile) -> str:
+        """Save uploaded file to disk."""
+        try:
+            file_path = os.path.join(self.upload_folder, upload_file.filename)
+            async with aiofiles.open(file_path, 'wb') as out_file:
+                content = await upload_file.read()
+                await out_file.write(content)
+            return file_path
+        except Exception as e:
+            logger.error(f"Error saving file: {str(e)}")
+            raise HTTPException(status_code=500, detail="Error saving file")
+
+    async def process_document(self, file_path: str) -> Dict[str, Any]:
+        """Process document based on its type."""
+        file_ext = os.path.splitext(file_path)[1].lower()
+        
+        try:
+            if file_ext == '.pdf':
+                return await self._process_pdf(file_path)
+            elif file_ext in ['.docx', '.doc']:
+                return await self._process_word(file_path)
+            elif file_ext in ['.png', '.jpg', '.jpeg']:
+                return await self._process_image(file_path)
+            else:
+                raise HTTPException(status_code=400, detail="Unsupported file type")
+        except Exception as e:
+            logger.error(f"Error processing document: {str(e)}")
+            raise HTTPException(status_code=500, detail="Error processing document")
+
+    async def _process_pdf(self, file_path: str) -> Dict[str, Any]:
+        """Extract text from PDF file."""
+        try:
+            text = ""
+            with open(file_path, 'rb') as file:
+                pdf_reader = PyPDF2.PdfReader(file)
+                for page in pdf_reader.pages:
+                    text += page.extract_text()
+            return {"text": text, "type": "pdf"}
+        except Exception as e:
+            logger.error(f"Error processing PDF: {str(e)}")
+            raise HTTPException(status_code=500, detail="Error processing PDF")
+
+    async def _process_word(self, file_path: str) -> Dict[str, Any]:
+        """Extract text from Word document."""
+        try:
+            doc = Document(file_path)
+            text = "\n".join([paragraph.text for paragraph in doc.paragraphs])
+            return {"text": text, "type": "word"}
+        except Exception as e:
+            logger.error(f"Error processing Word document: {str(e)}")
+            raise HTTPException(status_code=500, detail="Error processing Word document")
+
+    async def _process_image(self, file_path: str) -> Dict[str, Any]:
+        """Extract text from image using OCR."""
+        try:
+            image = Image.open(file_path)
+            text = pytesseract.image_to_string(image)
+            return {"text": text, "type": "image"}
+        except Exception as e:
+            logger.error(f"Error processing image: {str(e)}")
+            raise HTTPException(status_code=500, detail="Error processing image")
+
+    async def segment_document(self, text: str, max_segment_size: int = 1000) -> List[str]:
+        """Segment large documents into smaller chunks."""
+        segments = []
+        current_segment = ""
+        
+        for line in text.split('\n'):
+            if len(current_segment) + len(line) + 1 <= max_segment_size:
+                current_segment += line + '\n'
+            else:
+                if current_segment:
+                    segments.append(current_segment.strip())
+                current_segment = line + '\n'
+        
+        if current_segment:
+            segments.append(current_segment.strip())
+        
+        return segments
+
+document_service = DocumentService() 

services/queue_service.py

@@ -0,0 +1,83 @@
+import pika
+import json
+from typing import Callable, Dict, Any
+from loguru import logger
+from config.config import Settings
+
+settings = Settings()
+
+class QueueService:
+    def __init__(self):
+        self.connection = None
+        self.channel = None
+        self.connect()
+    
+    def connect(self):
+        """Establish connection to RabbitMQ"""
+        try:
+            self.connection = pika.BlockingConnection(
+                pika.ConnectionParameters(
+                    host=settings.RABBITMQ_HOST,
+                    port=settings.RABBITMQ_PORT,
+                    credentials=pika.PlainCredentials(
+                        settings.RABBITMQ_USER,
+                        settings.RABBITMQ_PASSWORD
+                    )
+                )
+            )
+            self.channel = self.connection.channel()
+            self._declare_queues()
+            logger.info("Connected to RabbitMQ successfully")
+        except Exception as e:
+            logger.error(f"Failed to connect to RabbitMQ: {str(e)}")
+            raise
+    
+    def _declare_queues(self):
+        """Declare all required queues"""
+        queues = [
+            "test_generation",
+            "test_execution",
+            "document_processing",
+            "ai_benchmarking"
+        ]
+        for queue in queues:
+            self.channel.queue_declare(queue=queue, durable=True)
+    
+    def publish_message(self, queue: str, message: Dict[str, Any]):
+        """Publish a message to a specific queue"""
+        try:
+            self.channel.basic_publish(
+                exchange='',
+                routing_key=queue,
+                body=json.dumps(message),
+                properties=pika.BasicProperties(
+                    delivery_mode=2,  # make message persistent
+                )
+            )
+            logger.info(f"Message published to queue {queue}")
+        except Exception as e:
+            logger.error(f"Failed to publish message: {str(e)}")
+            raise
+    
+    def consume_messages(self, queue: str, callback: Callable):
+        """Start consuming messages from a queue"""
+        try:
+            self.channel.basic_qos(prefetch_count=1)
+            self.channel.basic_consume(
+                queue=queue,
+                on_message_callback=callback
+            )
+            logger.info(f"Started consuming messages from queue {queue}")
+            self.channel.start_consuming()
+        except Exception as e:
+            logger.error(f"Failed to consume messages: {str(e)}")
+            raise
+    
+    def close(self):
+        """Close the connection"""
+        if self.connection and not self.connection.is_closed:
+            self.connection.close()
+            logger.info("RabbitMQ connection closed")
+
+# Create a singleton instance
+queue_service = QueueService() 

services/test_service.py

@@ -0,0 +1,259 @@
+from typing import List, Dict, Any, Optional
+from loguru import logger
+import json
+from datetime import datetime
+from testrail_api import TestRailAPI
+from jira import JIRA
+from qtest import QTestAPI
+
+class TestService:
+    def __init__(self):
+        self.testrail = None
+        self.jira = None
+        self.qtest = None
+        self._initialize_clients()
+
+    def _initialize_clients(self):
+        """Initialize test management tool clients."""
+        try:
+            if settings.TESTRAIL_API_KEY and settings.TESTRAIL_URL:
+                self.testrail = TestRailAPI(
+                    settings.TESTRAIL_URL,
+                    settings.TESTRAIL_API_KEY
+                )
+            
+            if settings.JIRA_API_TOKEN and settings.JIRA_URL:
+                self.jira = JIRA(
+                    server=settings.JIRA_URL,
+                    token_auth=settings.JIRA_API_TOKEN
+                )
+            
+            if settings.QTEST_API_TOKEN and settings.QTEST_URL:
+                self.qtest = QTestAPI(
+                    settings.QTEST_URL,
+                    settings.QTEST_API_TOKEN
+                )
+        except Exception as e:
+            logger.error(f"Error initializing test management clients: {str(e)}")
+
+    async def generate_test_cases(
+        self,
+        requirements: List[Dict[str, Any]],
+        ai_service: Any
+    ) -> List[Dict[str, Any]]:
+        """Generate test cases from requirements using AI."""
+        test_cases = []
+        
+        for req in requirements:
+            try:
+                prompt = self._create_test_case_prompt(req)
+                response = await ai_service.generate_response(prompt)
+                
+                test_case = self._parse_test_case_response(response["response"])
+                test_case["requirement_id"] = req["id"]
+                test_cases.append(test_case)
+            except Exception as e:
+                logger.error(f"Error generating test case for requirement {req['id']}: {str(e)}")
+        
+        return test_cases
+
+    def _create_test_case_prompt(self, requirement: Dict[str, Any]) -> str:
+        """Create prompt for test case generation."""
+        return f"""
+        Generate a test case for the following requirement:
+        
+        ID: {requirement['id']}
+        Title: {requirement['title']}
+        Description: {requirement['description']}
+        
+        Please provide:
+        1. Test case title
+        2. Preconditions
+        3. Test steps
+        4. Expected results
+        5. Priority (High/Medium/Low)
+        6. Test type (Functional/Integration/System)
+        """
+
+    def _parse_test_case_response(self, response: str) -> Dict[str, Any]:
+        """Parse AI response into structured test case."""
+        # Implement parsing logic based on AI response format
+        # This is a simplified example
+        lines = response.split('\n')
+        test_case = {
+            "title": "",
+            "preconditions": [],
+            "steps": [],
+            "expected_results": [],
+            "priority": "Medium",
+            "type": "Functional"
+        }
+        
+        current_section = None
+        for line in lines:
+            line = line.strip()
+            if not line:
+                continue
+                
+            if "Test case title:" in line:
+                test_case["title"] = line.split(":", 1)[1].strip()
+            elif "Preconditions:" in line:
+                current_section = "preconditions"
+            elif "Test steps:" in line:
+                current_section = "steps"
+            elif "Expected results:" in line:
+                current_section = "expected_results"
+            elif "Priority:" in line:
+                test_case["priority"] = line.split(":", 1)[1].strip()
+            elif "Test type:" in line:
+                test_case["type"] = line.split(":", 1)[1].strip()
+            elif current_section:
+                test_case[current_section].append(line)
+        
+        return test_case
+
+    async def export_to_testrail(
+        self,
+        test_cases: List[Dict[str, Any]],
+        project_id: int,
+        section_id: int
+    ) -> List[Dict[str, Any]]:
+        """Export test cases to TestRail."""
+        if not self.testrail:
+            raise ValueError("TestRail client not initialized")
+        
+        results = []
+        for test_case in test_cases:
+            try:
+                result = self.testrail.cases.add_case(
+                    section_id,
+                    title=test_case["title"],
+                    custom_steps=test_case["steps"],
+                    custom_expected=test_case["expected_results"],
+                    custom_preconds=test_case["preconditions"],
+                    priority_id=self._get_testrail_priority_id(test_case["priority"])
+                )
+                results.append({
+                    "test_case_id": test_case["id"],
+                    "testrail_id": result["id"],
+                    "status": "success"
+                })
+            except Exception as e:
+                logger.error(f"Error exporting test case to TestRail: {str(e)}")
+                results.append({
+                    "test_case_id": test_case["id"],
+                    "error": str(e),
+                    "status": "error"
+                })
+        
+        return results
+
+    async def export_to_jira(
+        self,
+        test_cases: List[Dict[str, Any]],
+        project_key: str
+    ) -> List[Dict[str, Any]]:
+        """Export test cases to JIRA."""
+        if not self.jira:
+            raise ValueError("JIRA client not initialized")
+        
+        results = []
+        for test_case in test_cases:
+            try:
+                issue_dict = {
+                    'project': {'key': project_key},
+                    'summary': test_case["title"],
+                    'description': self._format_jira_description(test_case),
+                    'issuetype': {'name': 'Test'},
+                    'priority': {'name': test_case["priority"]}
+                }
+                
+                issue = self.jira.create_issue(fields=issue_dict)
+                results.append({
+                    "test_case_id": test_case["id"],
+                    "jira_id": issue.key,
+                    "status": "success"
+                })
+            except Exception as e:
+                logger.error(f"Error exporting test case to JIRA: {str(e)}")
+                results.append({
+                    "test_case_id": test_case["id"],
+                    "error": str(e),
+                    "status": "error"
+                })
+        
+        return results
+
+    async def export_to_qtest(
+        self,
+        test_cases: List[Dict[str, Any]],
+        project_id: int
+    ) -> List[Dict[str, Any]]:
+        """Export test cases to qTest."""
+        if not self.qtest:
+            raise ValueError("qTest client not initialized")
+        
+        results = []
+        for test_case in test_cases:
+            try:
+                test_case_data = {
+                    "name": test_case["title"],
+                    "description": self._format_qtest_description(test_case),
+                    "priority": test_case["priority"],
+                    "type": test_case["type"]
+                }
+                
+                result = self.qtest.create_test_case(project_id, test_case_data)
+                results.append({
+                    "test_case_id": test_case["id"],
+                    "qtest_id": result["id"],
+                    "status": "success"
+                })
+            except Exception as e:
+                logger.error(f"Error exporting test case to qTest: {str(e)}")
+                results.append({
+                    "test_case_id": test_case["id"],
+                    "error": str(e),
+                    "status": "error"
+                })
+        
+        return results
+
+    def _get_testrail_priority_id(self, priority: str) -> int:
+        """Convert priority string to TestRail priority ID."""
+        priority_map = {
+            "High": 1,
+            "Medium": 2,
+            "Low": 3
+        }
+        return priority_map.get(priority, 2)
+
+    def _format_jira_description(self, test_case: Dict[str, Any]) -> str:
+        """Format test case for JIRA description."""
+        return f"""
+        *Preconditions:*
+        {chr(10).join(test_case['preconditions'])}
+
+        *Test Steps:*
+        {chr(10).join(test_case['steps'])}
+
+        *Expected Results:*
+        {chr(10).join(test_case['expected_results'])}
+
+        *Type:* {test_case['type']}
+        """
+
+    def _format_qtest_description(self, test_case: Dict[str, Any]) -> str:
+        """Format test case for qTest description."""
+        return f"""
+        Preconditions:
+        {chr(10).join(test_case['preconditions'])}
+
+        Test Steps:
+        {chr(10).join(test_case['steps'])}
+
+        Expected Results:
+        {chr(10).join(test_case['expected_results'])}
+        """
+
+test_service = TestService() 

tests/test_document_service.py

@@ -0,0 +1,67 @@
+import pytest
+from fastapi import UploadFile
+from io import BytesIO
+from services.document_service import document_service
+import os
+
+@pytest.fixture
+def sample_pdf():
+    return BytesIO(b"%PDF-1.4\n%Test PDF content")
+
+@pytest.fixture
+def sample_docx():
+    return BytesIO(b"PK\x03\x04\x14\x00\x00\x00\x08\x00")
+
+@pytest.fixture
+def sample_image():
+    return BytesIO(b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR")
+
+@pytest.mark.asyncio
+async def test_save_upload_file(sample_pdf):
+    file = UploadFile(
+        filename="test.pdf",
+        file=sample_pdf
+    )
+    file_path = await document_service.save_upload_file(file)
+    assert file_path.endswith("test.pdf")
+    assert os.path.exists(file_path)
+
+@pytest.mark.asyncio
+async def test_process_pdf(sample_pdf):
+    file = UploadFile(
+        filename="test.pdf",
+        file=sample_pdf
+    )
+    file_path = await document_service.save_upload_file(file)
+    result = await document_service.process_document(file_path)
+    assert "text" in result
+    assert result["type"] == "pdf"
+
+@pytest.mark.asyncio
+async def test_process_docx(sample_docx):
+    file = UploadFile(
+        filename="test.docx",
+        file=sample_docx
+    )
+    file_path = await document_service.save_upload_file(file)
+    result = await document_service.process_document(file_path)
+    assert "text" in result
+    assert result["type"] == "word"
+
+@pytest.mark.asyncio
+async def test_process_image(sample_image):
+    file = UploadFile(
+        filename="test.png",
+        file=sample_image
+    )
+    file_path = await document_service.save_upload_file(file)
+    result = await document_service.process_document(file_path)
+    assert "text" in result
+    assert result["type"] == "image"
+
+@pytest.mark.asyncio
+async def test_segment_document():
+    text = "Line 1\nLine 2\nLine 3\nLine 4\nLine 5"
+    segments = await document_service.segment_document(text, max_segment_size=20)
+    assert len(segments) > 0
+    assert all(len(segment) <= 20 for segment in segments) 

workers/test_worker.py

@@ -0,0 +1,101 @@
+import json
+import pika
+from loguru import logger
+from services.queue_service import queue_service
+from services.test_service import test_service
+from services.database_service import db_service
+from models.base import TestCase, TestScript
+
+def process_test_generation(ch, method, properties, body):
+    """Process test generation messages"""
+    try:
+        data = json.loads(body)
+        document_id = data.get('document_id')
+        
+        with db_service.get_session() as session:
+            # Generate test cases
+            test_cases = test_service.generate_test_cases(
+                document_id=document_id,
+                session=session
+            )
+            
+            # Save test cases to database
+            for test_case in test_cases:
+                db_test_case = TestCase(
+                    document_id=document_id,
+                    title=test_case['title'],
+                    description=test_case['description'],
+                    steps=test_case['steps'],
+                    expected_results=test_case['expected_results'],
+                    priority=test_case.get('priority', 1)
+                )
+                session.add(db_test_case)
+            
+            # Generate test scripts
+            for test_case in test_cases:
+                scripts = test_service.generate_test_scripts(test_case)
+                for script in scripts:
+                    db_script = TestScript(
+                        test_case_id=db_test_case.id,
+                        language=script['language'],
+                        code=script['code'],
+                        framework=script['framework']
+                    )
+                    session.add(db_script)
+        
+        ch.basic_ack(delivery_tag=method.delivery_tag)
+        logger.info(f"Successfully processed test generation for document {document_id}")
+    
+    except Exception as e:
+        logger.error(f"Error processing test generation: {str(e)}")
+        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
+
+def process_test_execution(ch, method, properties, body):
+    """Process test execution messages"""
+    try:
+        data = json.loads(body)
+        test_case_id = data.get('test_case_id')
+        
+        with db_service.get_session() as session:
+            test_case = session.query(TestCase).get(test_case_id)
+            if not test_case:
+                raise ValueError(f"Test case {test_case_id} not found")
+            
+            # Execute test case
+            result = test_service.execute_test_case(test_case)
+            
+            # Update test case status
+            test_case.status = result['status']
+            session.commit()
+        
+        ch.basic_ack(delivery_tag=method.delivery_tag)
+        logger.info(f"Successfully executed test case {test_case_id}")
+    
+    except Exception as e:
+        logger.error(f"Error executing test case: {str(e)}")
+        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
+
+def main():
+    """Start the worker"""
+    try:
+        # Start consuming test generation messages
+        queue_service.consume_messages(
+            queue="test_generation",
+            callback=process_test_generation
+        )
+        
+        # Start consuming test execution messages
+        queue_service.consume_messages(
+            queue="test_execution",
+            callback=process_test_execution
+        )
+    
+    except KeyboardInterrupt:
+        logger.info("Worker stopped by user")
+    except Exception as e:
+        logger.error(f"Worker error: {str(e)}")
+    finally:
+        queue_service.close()
+
+if __name__ == "__main__":
+    main()