server.py

import http.server
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
from urllib.parse import urlparse

class QAStudioRequestHandler(http.server.SimpleHTTPRequestHandler):
    def end_headers(self):
        # Allow CORS
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
        self.send_header('Access-Control-Allow-Headers', 'Content-Type')
        super().end_headers()

    def do_OPTIONS(self):
        self.send_response(200)
        self.end_headers()

    def do_POST(self):
        if self.path == '/api/run-generator':
            content_length = int(self.headers['Content-Length'])
            post_data = self.rfile.read(content_length)
            params = json.loads(post_data.decode('utf-8'))

            target_url = params.get('target_url')
            doc_content = params.get('doc_content')
            provider = params.get('provider', 'openai')
            api_key = params.get('api_key', '')
            hf_key = params.get('hf_key', '')
            hf_repo_id = params.get('hf_repo_id', '')
            category_count = params.get('category_count', '20')

            # Save doc content to a temp file
            with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
                f.write(doc_content)
                temp_doc_path = f.name

            # Set up environment variables for main.py and executor.py
            env = os.environ.copy()
            env['LLM_PROVIDER'] = provider
            if provider == 'openai':
                if api_key:
                    env['OPENAI_API_KEY'] = api_key
            else:
                if hf_key:
                    env['HF_TOKEN'] = hf_key
                    env['HUGGINGFACEHUB_API_TOKEN'] = hf_key
                if hf_repo_id:
                    env['HF_REPO_ID'] = hf_repo_id

            env['TARGET_PER_CATEGORY'] = str(category_count)
            env['API_BASE_URL'] = target_url
            env['CSV_PATH'] = 'output.csv'
            env['REPORT_PATH'] = 'execution_report.csv'
            env['SUMMARY_PATH'] = 'execution_summary.json'
            env['DOCUMENTATION_PATH'] = temp_doc_path

            try:
                # 1. Run main.py using subprocess (Stage 1: Generate Test Prompts)
                p1 = subprocess.Popen(
                    [sys.executable, 'main.py'],
                    stdin=subprocess.PIPE,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True,
                    env=env
                )
                # Pass doc path and target URL via stdin
                stdout, stderr = p1.communicate(input=f"{temp_doc_path}\n{target_url}\n")

                if p1.returncode != 0:
                    self.send_response(500)
                    self.send_header('Content-Type', 'application/json')
                    self.end_headers()
                    self.wfile.write(json.dumps({
                        'error': 'main.py test case generation failed',
                        'stdout': stdout,
                        'stderr': stderr
                    }).encode('utf-8'))
                    return

                # 2. Run executor.py (Stage 2: Run execution logs)
                p2 = subprocess.Popen(
                    [sys.executable, 'executor.py'],
                    stdin=subprocess.PIPE,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True,
                    env=env
                )
                stdout2, stderr2 = p2.communicate()

                # 3. Run eval.run_eval (Stage 3: Run DeepEval quality & correctness metrics)
                p3 = subprocess.Popen(
                    [sys.executable, '-m', 'eval.run_eval', '--csv', 'output.csv', '--report', 'execution_report.csv', '--output', 'eval_results.json'],
                    stdin=subprocess.PIPE,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True,
                    env=env
                )
                stdout3, stderr3 = p3.communicate()

                # Clean up temporary documentation file
                try:
                    os.unlink(temp_doc_path)
                except:
                    pass

                # Read output.csv, execution_report.csv, execution_summary.json, and eval_results.json
                csv_data_path = Path('output.csv')
                report_path = Path('execution_report.csv')
                summary_path = Path('execution_summary.json')
                eval_path = Path('eval_results.json')

                output_csv_content = csv_data_path.read_text(encoding='utf-8') if csv_data_path.exists() else ''
                report_content = report_path.read_text(encoding='utf-8') if report_path.exists() else ''
                summary_content = json.loads(summary_path.read_text(encoding='utf-8')) if summary_path.exists() else {}
                eval_content = json.loads(eval_path.read_text(encoding='utf-8')) if eval_path.exists() else {}

                self.send_response(200)
                self.send_header('Content-Type', 'application/json')
                self.end_headers()
                self.wfile.write(json.dumps({
                    'success': True,
                    'output_csv': output_csv_content,
                    'execution_report': report_content,
                    'execution_summary': summary_content,
                    'eval_results': eval_content
                }).encode('utf-8'))

            except Exception as e:
                self.send_response(500)
                self.send_header('Content-Type', 'application/json')
                self.end_headers()
                self.wfile.write(json.dumps({'error': str(e)}).encode('utf-8'))
        else:
            super().do_POST()

    def translate_path(self, path):
        # Serve static UI files
        parsed = urlparse(path)
        path = parsed.path
        if path == '/' or path == '/index.html':
            return str(Path('ui/index.html').absolute())
        
        if path.startswith('/ui/'):
            path = path[4:]
        
        target = Path('ui') / path.lstrip('/')
        if target.exists():
            return str(target.absolute())
        return super().translate_path(path)

def run(port=8080):
    server_address = ('', port)
    httpd = http.server.HTTPServer(server_address, QAStudioRequestHandler)
    print(f"Starting QA Studio local server on http://localhost:{port}...")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\nStopping QA Studio local server...")
        httpd.server_close()

if __name__ == '__main__':
    port = 8080
    env_port = os.environ.get('PORT')
    if env_port:
        try:
            port = int(env_port)
        except ValueError:
            pass
    elif len(sys.argv) > 1:
        try:
            port = int(sys.argv[1])
        except ValueError:
            pass
    run(port)