app.py

import gradio as gr
from datetime import datetime
from groq import Groq

import traceback
import json

# --- 1. الحصول على مفتاح API ---
import os


# تهيئة المكونات

api_key_coder= os.environ.get('Chat_with_Your_Context')

# تهيئة المكونات


# --- 2. تعريف عميل Groq مباشرة (بدون LangChain) ---
class GroqLLM:
    def __init__(self, api_key, model="meta-llama/llama-4-scout-17b-16e-instruct", temperature=0.1):
        self.client = Groq(api_key=api_key)
        self.model = model
        self.temperature = temperature

    def invoke(self, prompt):
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                temperature=self.temperature,
                max_tokens=2000
            )
            return response.choices[0].message.content
        except Exception as e:
            return f"Error: {str(e)}"

# --- 3. تفعيل LLM ---
llm = GroqLLM(api_key=api_key_coder)

# --- 4. تعريف Dummy Runner و DOM Matcher ---
class DummyRunner:
    def run(self, test_script):
        # محاكاة لنتائج الاختبارات
        if "fail" in test_script.lower() or "assert false" in test_script.lower():
            return {
                "status": "failed",
                "error": "AssertionError: Expected True but got False",
                "logs": "Stack trace: line 10 in test_function",
                "dom": "<button id='submit-btn' class='btn'>Submit</button>"
            }
        return {"status": "passed", "message": "All tests passed successfully"}

class DOMMatcher:
    def find_similar(self, dom, failed_locator):
        # محاكاة لإيجاد locator بديل
        return "button#submit-btn", 0.92

runner = DummyRunner()
dom_matcher = DOMMatcher()

# --- 5. تعريف وظائف الأدوات (بدون LangChain) ---
def detect_failure(test_script):
    """اكتشاف فشل الاختبار"""
    result = runner.run(test_script)
    return result

def analyze_root_cause(failure_data):
    """تحليل سبب الفشل باستخدام LLM"""
    error = failure_data.get("error", "Unknown")
    logs = failure_data.get("logs", "")

    prompt = f"""
    Analyze this test failure:

    Error: {error}
    Logs: {logs}

    Provide:
    1. Root cause analysis
    2. Suggested fix
    """

    analysis = llm.invoke(prompt)
    return {"root_cause": analysis, "confidence": "high"}

def heal_locator(failure_data):
    """محاولة إصلاح الـ locator"""
    dom = failure_data.get("dom", "")
    error = failure_data.get("error", "")

    new_locator, score = dom_matcher.find_similar(dom, error)
    return {"suggested_locator": new_locator, "confidence": score}

def update_script(script_content, old_locator, new_locator):
    """تحديث السكربت بـ locator جديد"""
    return script_content.replace(old_locator, new_locator)

def reexecute_test(test_script):
    """إعادة تشغيل الاختبار"""
    return runner.run(test_script)

def generate_report(data):
    """توليد تقرير شامل"""
    prompt = f"""
    Generate a comprehensive QA report based on this data:
    {json.dumps(data, indent=2)}

    Include:
    - Test Execution Summary
    - Failures Detected
    - Root Cause Analysis
    - Healing Actions
    - Final Results
    - Recommendations
    """

    return llm.invoke(prompt)

# --- 6. الدالة الرئيسية التي تنفذ كل الخطوات ---
def run_complete_analysis(test_script):
    """تنفيذ تحليل كامل للاختبار"""

    report_data = {
        "original_script": test_script,
        "steps": [],
        "final_result": {},
        "healing_applied": False
    }

    # الخطوة 1: تشغيل الاختبار
    result = detect_failure(test_script)
    report_data["steps"].append({"step": "initial_execution", "result": result})

    # إذا فشل الاختبار
    if result["status"] == "failed":
        report_data["healing_applied"] = True

        # الخطوة 2: تحليل السبب
        analysis = analyze_root_cause(result)
        report_data["steps"].append({"step": "root_cause_analysis", "analysis": analysis})

        # الخطوة 3: محاولة الإصلاح
        healing = heal_locator(result)
        report_data["steps"].append({"step": "healing_attempt", "healing": healing})

        # الخطوة 4: تحديث السكربت
        if "suggested_locator" in healing:
            old = "button"
            new = healing["suggested_locator"]
            updated_script = update_script(test_script, old, new)
            report_data["steps"].append({"step": "script_updated", "new_script": updated_script})

            # الخطوة 5: إعادة التشغيل
            final_result = reexecute_test(updated_script)
            report_data["final_result"] = final_result
            report_data["steps"].append({"step": "re_execution", "result": final_result})
    else:
        report_data["final_result"] = result

    # الخطوة 6: توليد التقرير
    report = generate_report(report_data)
    report_data["full_report"] = report

    return report_data

# SmartQA Full System with Multi-Source Test Generation + HealTest Integration
# Uses existing GroqLLM instance: llm = GroqLLM(api_key=api_key_coder)

# ==============================
# 1. Knowledge Input Definition
# ==============================

class KnowledgeInput:
    def __init__(
        self,
        requirements=None,
        dom=None,
        api_spec=None,
        user_flows=None,
        source_code=None,
        recording=None
    ):
        self.requirements = requirements
        self.dom = dom
        self.api_spec = api_spec
        self.user_flows = user_flows
        self.source_code = source_code
        self.recording = recording


# ==============================
# 2. Knowledge Processor
# ==============================

class KnowledgeProcessor:

    def parse_requirements(self, text):
        return text.strip()

    def parse_dom(self, dom_text):
        return dom_text[:4000]

    def parse_api(self, api_text):
        return api_text[:4000]

    def parse_flows(self, flows_text):
        return flows_text.strip()

    def analyze_code(self, code_text):
        return code_text[:4000]

    def parse_recording(self, rec_text):
        return rec_text.strip()

    def process(self, knowledge: KnowledgeInput):
        data = {}

        if knowledge.requirements:
            data["req"] = self.parse_requirements(knowledge.requirements)

        if knowledge.dom:
            data["ui"] = self.parse_dom(knowledge.dom)

        if knowledge.api_spec:
            data["api"] = self.parse_api(knowledge.api_spec)

        if knowledge.user_flows:
            data["flows"] = self.parse_flows(knowledge.user_flows)

        if knowledge.source_code:
            data["code"] = self.analyze_code(knowledge.source_code)

        if knowledge.recording:
            data["record"] = self.parse_recording(knowledge.recording)

        return data


# ==============================
# 3. Test Generator (LLM-based)
# ==============================

class TestGenerator:
    def __init__(self, llm):
        self.llm = llm

    def generate_req_tests(self, data):
        prompt = f"""
        Generate Python Selenium automated test scripts from requirements.
        Requirements:\n{data['req']}

        Include:
        - pytest format
        - locators placeholders
        - assertions
        """
        return self.llm.invoke(prompt)

    def generate_ui_tests(self, data):
        prompt = f"""
        Generate Selenium UI tests from HTML DOM.
        DOM:\n{data['ui']}
        """
        return self.llm.invoke(prompt)

    def generate_api_tests(self, data):
        prompt = f"""
        Generate Python API tests using requests from OpenAPI/Swagger spec.
        Spec:\n{data['api']}
        """
        return self.llm.invoke(prompt)

    def generate_flow_tests(self, data):
        prompt = f"""
        Generate end-to-end Selenium tests from user flows.
        Flows:\n{data['flows']}
        """
        return self.llm.invoke(prompt)

    def generate_code_tests(self, data):
        prompt = f"""
        Analyze source code and generate relevant automated tests.
        Code:\n{data['code']}
        """
        return self.llm.invoke(prompt)

    def generate_record_tests(self, data):
        prompt = f"""
        Convert user interaction recording into Selenium test script.
        Recording:\n{data['record']}
        """
        return self.llm.invoke(prompt)

    def generate(self, processed_data):

        if "api" in processed_data:
            return self.generate_api_tests(processed_data)

        if "ui" in processed_data:
            return self.generate_ui_tests(processed_data)

        if "flows" in processed_data:
            return self.generate_flow_tests(processed_data)

        if "req" in processed_data:
            return self.generate_req_tests(processed_data)

        if "code" in processed_data:
            return self.generate_code_tests(processed_data)

        if "record" in processed_data:
            return self.generate_record_tests(processed_data)

        return "No valid input provided"


# ==============================
# 4. HealTest Engine (Wrapper)
# ==============================

class HealTestEngine:
    def __init__(self):
        pass

    def run_complete_analysis(self, test_script):
        # Uses existing functions from your notebook
        result = detect_failure(test_script)

        if result["status"] == "failed":
            analysis = analyze_root_cause(result)
            healed = heal_locator(result)
            updated_script = update_script(
                test_script,
                result.get("failed_locator", "old_locator"),
                healed.get("new_locator", "new_locator")
            )
            re_result = reexecute_test(updated_script)
        else:
            analysis = "No failure"
            healed = {}
            updated_script = test_script
            re_result = result

        report_data = {
            "original": test_script,
            "analysis": analysis,
            "healing": healed,
            "final_result": re_result
        }

        report = generate_report(report_data)

        return {
            "generated_test": test_script,
            "updated_test": updated_script,
            "initial_result": result,
            "final_result": re_result,
            "report": report
        }


# ==============================
# 5. SmartQA System
# ==============================

class SmartQASystem:
    def __init__(self, llm):
        self.processor = KnowledgeProcessor()
        self.generator = TestGenerator(llm)
        self.healer = HealTestEngine()

    def run(self, knowledge: KnowledgeInput):
        processed = self.processor.process(knowledge)
        generated_tests = self.generator.generate(processed)
        results = self.healer.run_complete_analysis(generated_tests)
        return results


# ==============================
# 6. Gradio Interface
# ==============================
import gradio as gr

# --- 3. تفعيل LLM ---

system = SmartQASystem(llm)


def run_smartqa(requirements, dom, api_spec, flows, code, recording):
    knowledge = KnowledgeInput(
        requirements=requirements,
        dom=dom,
        api_spec=api_spec,
        user_flows=flows,
        source_code=code,
        recording=recording
    )

    result = system.run(knowledge)

    return (
        result["generated_test"],
        result["updated_test"],
        str(result["initial_result"]),
        str(result["final_result"]),
        result["report"]
    )


with gr.Blocks() as demo:
    gr.Markdown("# 🧠 SmartQA — Multi-Source AI Test Generation & Self-Healing")
    gr.Markdown("Provide any knowledge source to generate and heal automated tests")

    with gr.Tab("Requirements"):
        req_input = gr.Textbox(lines=8, label="Requirements")

    with gr.Tab("UI / DOM"):
        dom_input = gr.Textbox(lines=12, label="HTML DOM")

    with gr.Tab("API Spec"):
        api_input = gr.Textbox(lines=12, label="OpenAPI / Swagger")

    with gr.Tab("User Flows"):
        flow_input = gr.Textbox(lines=8, label="User Flows")

    with gr.Tab("Source Code"):
        code_input = gr.Textbox(lines=12, label="Source Code")

    with gr.Tab("Recording"):
        rec_input = gr.Textbox(lines=8, label="Interaction Recording")

    # --- صف الأزرار أفقيًا ---
    with gr.Row():
        run_btn = gr.Button("🚀 Generate & Heal Tests", variant="primary")
        example_btn = gr.Button("📂 Load Example Data")

    # --- النتائج ---
    gr.Markdown("## Results")
    gen_out = gr.Code(label="Generated Test")
    upd_out = gr.Code(label="Healed Test")
    init_out = gr.Textbox(label="Initial Execution Result")
    final_out = gr.Textbox(label="Final Execution Result")
    report_out = gr.Textbox(lines=12, label="QA Report")

    # --- بيانات أمثلة ---
    example_requirements = """
User can login with email and password
User can search for a product
User can add product to cart
"""

    example_dom = """
<html>
  <body>
    <input id=\"email\" />
    <input id=\"password\" />
    <button id=\"login-btn\">Login</button>

    <input id=\"search\" />
    <button id=\"search-btn\">Search</button>

    <button id=\"add-cart\">Add to Cart</button>
  </body>
</html>
"""

    example_api = """
POST /login
Body: { email, password }

GET /products

POST /cart
Body: { product_id }
"""

    example_flows = """
Open login page
Enter email and password
Click login
Search product
Add to cart
"""

    example_code = """
@app.route('/login', methods=['POST'])
def login():
    email = request.json['email']
    password = request.json['password']
    if authenticate(email, password):
        return {'status': 'ok'}
    return {'status': 'fail'}, 401
"""

    example_recording = """
User navigates to /login
Types email test@mail.com
Types password 123456
Clicks Login button
Navigates to /products
Clicks Add to Cart
"""

    def load_examples():
        return (
            example_requirements,
            example_dom,
            example_api,
            example_flows,
            example_code,
            example_recording
        )

    # --- ربط الأزرار ---
    run_btn.click(
        fn=run_smartqa,
        inputs=[req_input, dom_input, api_input, flow_input, code_input, rec_input],
        outputs=[gen_out, upd_out, init_out, final_out, report_out]
    )

    example_btn.click(
        fn=load_examples,
        inputs=[],
        outputs=[req_input, dom_input, api_input, flow_input, code_input, rec_input]
    )

demo.launch(debug=True)