Spaces:
Build error
Build error
| 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() | |