app.py

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()