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 io, os # Load semantic similarity model once model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2') def calculate_semantic_similarity(text1: str, text2: str) -> float: embeddings = model.encode([text1, text2]) return float(cosine_similarity([embeddings[0]], [embeddings[1]])[0][0]) def is_empty_response(response): return response is None or (isinstance(response, str) and response.strip() in ['', 'null', 'undefined']) def validate_json_response(expected, actual): try: expected_json = json.loads(expected) if isinstance(expected, str) else expected actual_json = json.loads(actual) if isinstance(actual, str) else actual return expected_json == actual_json, "Exact match" if expected_json == actual_json else "Mismatch" except Exception as e: return False, f"Invalid JSON: {str(e)}" def check_connectivity(endpoint, headers): try: resp = requests.head(endpoint, headers=headers, timeout=5) return True, resp.status_code except requests.exceptions.RequestException as e: return False, str(e) def run_single_test(endpoint, method, bearer_token, expected_status, expected_body, payload=None, test_name=None): result = { 'TestName': test_name, 'Endpoint': endpoint, 'Method': method, 'Expected Status': expected_status, 'Actual Status': None, 'Expected Body': expected_body[:100] + '...' if isinstance(expected_body, str) and len(expected_body) > 100 else expected_body, 'Actual Body': None, 'Outcome': 'FAIL', 'Similarity': None, 'Notes': '' } try: headers = {} if bearer_token and bearer_token.strip(): headers['Authorization'] = f'Bearer {bearer_token.strip()}' # Connectivity pre-check reachable, info = check_connectivity(endpoint, headers) if not reachable: result['Outcome'] = 'UNREACHABLE' result['Notes'] = f'Cannot reach endpoint: {info}' return result data, json_payload = None, None if payload and isinstance(payload, str) and payload.strip(): if payload.lower().startswith("header:"): try: kv = payload.split(":",1)[1] key, value = kv.split("=",1) headers[key.strip()] = value.strip().strip('"') except Exception: result['Notes'] = f'Invalid header format: {payload}' return result else: try: json_payload = json.loads(payload) except Exception: data = payload if method.upper() == 'GET': response = requests.get(endpoint, headers=headers, timeout=10) elif method.upper() == 'POST': response = requests.post(endpoint, headers=headers, data=data, json=json_payload, timeout=10) elif method.upper() == 'PUT': response = requests.put(endpoint, headers=headers, data=data, json=json_payload, 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 if response.status_code != expected_status: result['Outcome'] = 'FAIL' result['Notes'] = f'Status mismatch: expected {expected_status}, got {response.status_code}' return result if not expected_body or is_empty_response(expected_body): result['Outcome'] = 'PASS' result['Notes'] = 'Status matches, no body validation' return result if is_empty_response(actual_body): result['Outcome'] = 'UNCERTAIN' result['Similarity'] = 0.0 result['Notes'] = 'Response is empty (manual review needed)' return result json_match, _ = validate_json_response(expected_body, actual_body) if json_match: result['Outcome'] = 'PASS' result['Similarity'] = 1.0 result['Notes'] = 'Exact JSON match' return result try: similarity = calculate_semantic_similarity(expected_body, actual_body) result['Similarity'] = round(similarity, 3) 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 similarity: {similarity:.1%} (review needed)' else: result['Outcome'] = 'UNCERTAIN' result['Notes'] = f'Low similarity: {similarity:.1%}' 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 normalize_to_csv(file): filename = file if isinstance(file, str) else getattr(file, "name", None) ext = os.path.splitext(filename)[1].lower() try: if ext in [".xlsx", ".xlsm", ".xls"]: df = pd.read_excel(file, engine="openpyxl") elif ext == ".csv": df = pd.read_csv(file, encoding="utf-8") elif ext in [".tsv", ".txt"]: try: df = pd.read_csv(file, sep="\t", encoding="utf-8") except UnicodeDecodeError: df = pd.read_csv(file, sep="\t", encoding="latin1") else: raise ValueError(f"Unsupported file type: {ext}") except Exception as e: raise RuntimeError(f"Error reading file: {e}") df.columns = [c.strip() for c in df.columns] return df def process_file(file, bearer_token: str): try: df = normalize_to_csv(file) required_columns = ['Endpoint', 'Method', 'ExpectedStatus', 'ExpectedBody'] missing = [c for c in required_columns if c not in df.columns] if missing: return pd.DataFrame([{"Outcome":"ERROR","Notes":f"Missing columns: {', '.join(missing)}"}]), f"❌ Missing columns: {', '.join(missing)}" results = [] for _, row in df.iterrows(): results.append(run_single_test( endpoint=str(row['Endpoint']), method=str(row['Method']), bearer_token=bearer_token, expected_status=int(row['ExpectedStatus']), expected_body=str(row['ExpectedBody']), payload=row.get('Payload', None), test_name=row.get('TestName', None) )) results_df = pd.DataFrame(results) summary = f""" ### 📊 Test Summary - Total Tests: {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"])} - 🚫 Unreachable: {len(results_df[results_df['Outcome']=="UNREACHABLE"])} """ return results_df, summary except Exception as e: return pd.DataFrame([{"Outcome":"ERROR","Notes":str(e)}]), f"❌ Error processing file: {str(e)}" def download_results(df): 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") output.seek(0) return output # Gradio UI with gr.Blocks(css="* { font-family: 'Times New Roman', serif; }") as app: gr.Markdown("## 🚀 API Test Runner with AI-Powered Validation") token_state = gr.State("") with gr.Tab("Run Tests"): with gr.Group(): file_input = gr.File(label="📂 Upload Test Cases", file_types=['.xlsx', '.xls', '.csv', '.tsv', '.txt']) token_input = gr.Textbox(label="🔑 Bearer Token", type="password") run_button = gr.Button("▶️ Run Tests", variant="primary") with gr.Tab("Results"): summary_output = gr.Markdown() results_output = gr.Dataframe( headers=['TestName','Endpoint','Method','Expected Status','Actual Status','Outcome','Similarity','Notes'], interactive=False ) download_button = gr.Button("💾 Download Results") download_output = gr.File() def run_with_feedback(file, token): df, summary = process_file(file, token) return df, summary token_input.change(fn=lambda t: t, inputs=token_input, outputs=token_state) run_button.click(fn=run_with_feedback, inputs=[file_input, token_state], outputs=[results_output, summary_output]) download_button.click(fn=download_results, inputs=[results_output], outputs=[download_output]) if __name__ == "__main__": app.launch()