Update app.py

app.py

@@ -1,233 +1,87 @@
-import gradio as gr
 import pandas as pd
 import requests
-import json
-from sentence_transformers import SentenceTransformer
-from sklearn.metrics.pairwise import cosine_similarity
-import numpy as np
-from typing import Dict, Tuple
-import io
+import gradio as gr
+from sentence_transformers import SentenceTransformer, util
+from sklearn.linear_model import LogisticRegression
+import os
 
 # Load semantic similarity model
-model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
+model = SentenceTransformer("multi-qa-mpnet-base-dot-v1")
 
-def calculate_semantic_similarity(text1: str, text2: str) -> float:
-    """Calculate semantic similarity between two texts using AI embeddings"""
-    embeddings = model.encode([text1, text2])
-    similarity = cosine_similarity([embeddings[0]], [embeddings[1]])[0][0]
-    return float(similarity)
+# Initialize classifier
+classifier = LogisticRegression()
+feedback_file = "feedback.csv"
 
-def is_empty_response(response):
-    """Check if response is empty or null"""
-    if response is None or response == '':
-        return True
-    if isinstance(response, str):
-        return response.strip() in ['', 'null', 'undefined']
-    return False
+# Load feedback if available
+if os.path.exists(feedback_file):
+    feedback_data = pd.read_csv(feedback_file)
+    if len(feedback_data) > 10:
+        X = feedback_data[["similarity", "resp_length"]]
+        y = feedback_data["label"]
+        classifier.fit(X, y)
+else:
+    feedback_data = pd.DataFrame(columns=["similarity","resp_length","label"])
 
-def validate_json_response(expected, actual) -> Tuple[bool, str]:
-    """Validate if JSON responses match"""
-    try:
-        expected_json = json.loads(expected) if isinstance(expected, str) else expected
-        actual_json = json.loads(actual) if isinstance(actual, str) else actual
-        
-        if expected_json == actual_json:
-            return True, "Exact match"
-        return False, "JSON structure mismatch"
-    except Exception as e:
-        return False, f"Invalid JSON: {str(e)}"
-
-def run_single_test(endpoint: str, method: str, bearer_token: str, 
-                    expected_status: int, expected_body: str) -> Dict:
-    """Run a single API test with AI-powered semantic validation"""
-    result = {
-        'Endpoint': endpoint,
-        'Method': method,
-        'Expected Status': expected_status,
-        'Actual Status': None,
-        'Expected Body': expected_body[:100] + '...' if len(expected_body) > 100 else expected_body,
-        'Actual Body': None,
-        'Outcome': 'FAIL',
-        'Similarity': None,
-        'Notes': ''
-    }
+def run_tests(file, bearer_token=None):
+    df = pd.read_excel(file)
+    results = []
+    headers = {}
+    if bearer_token:
+        headers["Authorization"] = f"Bearer {bearer_token}"
     
-    try:
-        headers = {}
-        if bearer_token and bearer_token.strip():
-            headers['Authorization'] = f'Bearer {bearer_token.strip()}'
-        
-        # Make API call
-        if method.upper() == 'GET':
-            response = requests.get(endpoint, headers=headers, timeout=10)
-        elif method.upper() == 'POST':
-            response = requests.post(endpoint, headers=headers, timeout=10)
-        elif method.upper() == 'PUT':
-            response = requests.put(endpoint, headers=headers, timeout=10)
-        elif method.upper() == 'DELETE':
-            response = requests.delete(endpoint, headers=headers, timeout=10)
-        else:
-            result['Notes'] = f'Unsupported method: {method}'
-            return result
-        
-        result['Actual Status'] = response.status_code
-        actual_body = response.text
-        result['Actual Body'] = actual_body[:100] + '...' if len(actual_body) > 100 else actual_body
-        
-        # Check status code
-        if response.status_code != expected_status:
-            result['Outcome'] = 'FAIL'
-            result['Notes'] = f'Status mismatch: expected {expected_status}, got {response.status_code}'
-            return result
-        
-        # Status matches - now check body
-        if not expected_body or is_empty_response(expected_body):
-            result['Outcome'] = 'PASS'
-            result['Notes'] = 'Status matches, no body validation'
-            return result
+    for _, row in df.iterrows():
+        test_name = row["TestName"]
+        endpoint = row["Endpoint"]
+        method = row["Method"]
+        payload = row.get("Payload", None)
+        expected_status = row["ExpectedStatus"]
+        expected_body = str(row["ExpectedBody"])
         
-        if is_empty_response(actual_body):
-            result['Outcome'] = 'UNCERTAIN'
-            result['Similarity'] = 0.0
-            result['Notes'] = 'Response is empty (manual review needed)'
-            return result
-        
-        # Try JSON validation first
-        json_match, json_reason = validate_json_response(expected_body, actual_body)
-        if json_match:
-            result['Outcome'] = 'PASS'
-            result['Similarity'] = 1.0
-            result['Notes'] = 'Exact JSON match'
-            return result
-        
-        # Use AI semantic similarity
         try:
-            similarity = calculate_semantic_similarity(expected_body, actual_body)
-            result['Similarity'] = round(similarity, 3)
+            response = requests.request(method, endpoint, headers=headers, data=payload)
+            status_match = response.status_code == expected_status
+            body = response.text
+        except Exception as e:
+            results.append((test_name, "Error", str(e)))
+            continue
+        
+        if not status_match:
+            result = "Fail"
+        elif body == expected_body:
+            result = "Pass"
+        else:
+            sim = util.cos_sim(model.encode(expected_body), model.encode(body)).item()
+            resp_length = len(body)
             
-            if similarity >= 0.75:
-                result['Outcome'] = 'PASS'
-                result['Notes'] = f'High semantic similarity: {similarity:.1%}'
-            elif similarity >= 0.5:
-                result['Outcome'] = 'UNCERTAIN'
-                result['Notes'] = f'Medium semantic similarity: {similarity:.1%} (manual review needed)'
+            # If classifier trained, use it
+            if len(feedback_data) > 10:
+                pred = classifier.predict([[sim, resp_length]])[0]
+                result = "Pass" if pred == 1 else "Fail"
             else:
-                result['Outcome'] = 'UNCERTAIN'
-                result['Notes'] = f'Low semantic similarity: {similarity:.1%} (manual review needed)'
-        except Exception as sim_error:
-            result['Outcome'] = 'UNCERTAIN'
-            result['Notes'] = f'Cannot compute similarity: {str(sim_error)}'
-        
-        return result
-        
-    except requests.exceptions.Timeout:
-        result['Notes'] = 'Request timeout'
-        return result
-    except requests.exceptions.ConnectionError:
-        result['Notes'] = 'Connection error'
-        return result
-    except Exception as e:
-        result['Notes'] = f'Error: {str(e)}'
-        return result
-
-def process_excel_file(file, bearer_token: str) -> Tuple[pd.DataFrame, str]:
-    """Process uploaded Excel file and run all tests"""
-    try:
-        df = pd.read_excel(file)
-        
-        required_columns = ['Endpoint', 'Method', 'Expected Status Code', 'Expected Response Body']
-        missing_columns = [col for col in required_columns if col not in df.columns]
-        
-        if missing_columns:
-            return None, f"❌ Missing columns: {', '.join(missing_columns)}"
-        
-        results = []
-        for _, row in df.iterrows():
-            result = run_single_test(
-                endpoint=str(row['Endpoint']),
-                method=str(row['Method']),
-                bearer_token=bearer_token,
-                expected_status=int(row['Expected Status Code']),
-                expected_body=str(row['Expected Response Body'])
-            )
-            results.append(result)
-        
-        results_df = pd.DataFrame(results)
-        
-        total = len(results_df)
-        passed = len(results_df[results_df['Outcome'] == 'PASS'])
-        failed = len(results_df[results_df['Outcome'] == 'FAIL'])
-        uncertain = len(results_df[results_df['Outcome'] == 'UNCERTAIN'])
-        
-        summary = f"""
-        📊 **Test Summary**
-        - Total Tests: {total}
-        - ✅ Passed: {passed}
-        - ❌ Failed: {failed}
-        - ⚠️ Uncertain: {uncertain}
-        """
-        
-        return results_df, summary
+                result = "Pass" if sim > 0.75 else "Uncertain"
+            
+            # Store similarity for feedback loop
+            results.append((test_name, result, sim))
+            continue
         
-    except Exception as e:
-        return None, f"❌ Error processing file: {str(e)}"
-
-def download_results(df):
-    """Convert DataFrame to Excel for download"""
-    if df is None or df.empty:
-        return None
-    
-    output = io.BytesIO()
-    with pd.ExcelWriter(output, engine='openpyxl') as writer:
-        df.to_excel(writer, index=False, sheet_name='Test Results')
+        results.append((test_name, result, None))
     
-    output.seek(0)
-    return output
+    return pd.DataFrame(results, columns=["TestName", "Result", "Similarity"])
 
-# Gradio UI
-with gr.Blocks(title="API Test Runner with AI", theme=gr.themes.Soft()) as app:
-    gr.Markdown("""
-    # 🚀 API Test Runner with AI-Powered Validation
-    
-    Upload an Excel file with your API test cases and get automated validation with semantic similarity analysis.
-    """)
-    
-    with gr.Row():
-        with gr.Column(scale=2):
-            file_input = gr.File(label="Upload Excel File (.xlsx)", file_types=['.xlsx'])
-            token_input = gr.Textbox(
-                label="Bearer Token (Optional)",
-                placeholder="Enter your API bearer token if required",
-                type="password"
-            )
-            run_button = gr.Button("🔍 Run Tests", variant="primary", size="lg")
-        
-        with gr.Column(scale=1):
-            summary_output = gr.Markdown(label="Summary")
-    
-    with gr.Row():
-        results_output = gr.Dataframe(
-            label="Test Results",
-            headers=['Endpoint', 'Method', 'Expected Status', 'Actual Status', 
-                    'Outcome', 'Similarity', 'Notes'],
-            interactive=False
-        )
-    
-    with gr.Row():
-        download_button = gr.Button("💾 Download Results")
-        download_output = gr.File(label="Download Excel Report")
-    
-    run_button.click(
-        fn=process_excel_file,
-        inputs=[file_input, token_input],
-        outputs=[results_output, summary_output]
-    )
-    
-    download_button.click(
-        fn=download_results,
-        inputs=[results_output],
-        outputs=[download_output]
-    )
+def feedback_handler(similarity, resp_length, label):
+    global feedback_data
+    feedback_data = feedback_data.append({"similarity": similarity,
+                                          "resp_length": resp_length,
+                                          "label": label}, ignore_index=True)
+    feedback_data.to_csv(feedback_file, index=False)
+    return "Feedback saved!"
+
+demo = gr.Interface(
+    fn=run_tests,
+    inputs=[gr.File(label="Upload Excel"), gr.Textbox(label="Bearer Token")],
+    outputs="dataframe",
+    title="Adaptive API Test Runner",
+    description="Upload test cases in Excel, run API tests, and classify results with semantic similarity + ML feedback loop."
+)
 
-if __name__ == "__main__":
-    app.launch()
+demo.launch()