ci_cd/test_api.py

# Importo le librerie necessarie:

import pandas as pd
import requests

def integration_test():

    """
    Funzione di test di integrazione che riceve i dati dal dataset campione, verifica le previsioni del modello,
    verifica l'endpoint API e fornisce i risultati dei test effettuati 
    """

    # Carico i dati per effettuare i test, gestisco l'errore di caricamento dati:
    print("Caricamento dati test...")
    try:
        df = pd.read_csv("data/tweets.csv")
    except FileNotFoundError:
        print("Errore durante il caricamento: file non trovato")
        return
    
    # Prendo il campione di 10 tweet:
    sample = df.head(10)

    # Definisco l'endpoint dell'API locale:
    url = "http://127.0.0.1:8000/predict"

    # Creo una variabile-contatore per i test di successo:
    test_passati = 0

    print(f"Inizio del test di integrazione su {len(sample)} campioni...\n")

    # Itero sulle righe del dataset per ottenere "features" e "target":
    for index, row in sample.iterrows():
        payload = {"text": row["text"]}
        expected = row["sentiment_label"]

        # Invio richesta di post:
        try:
            response = requests.post(url, json=payload)
            if response.status_code == 200:
                result = response.json()
                prediction = result["Verdetto"]
            else:
                print(f"Errore: {response.status_code}")
                continue
            
            # Confronto i risultati ottenuti con quelli del dataset:
            is_correct = (prediction == expected)
            match_status = "[OK]" if is_correct else "[FALLITO]"
            if is_correct:
                test_passati +=1

            # Stampo i test:
            print(f"Test n° {index + 1}:")
            print(f"Testo: {row['text'][:60]}")
            print(f"Risultato API: {prediction} | Atteso: {expected} | Status: {match_status}\n")

        # Gestisco l'eccezione di API non attiva:
        except Exception as e:
            print(f"Errore: {e}")
            print("Verificare la porta 8000")   
            break
    
    # Stampo le statistiche:
    precisione = (test_passati/len(sample)) *100
    print("-" *40)
    print("Riepilogo test")
    print("-" *40)
    print(f"Precisione: {test_passati}/{len(sample)} {precisione}%")

# Punto di avvio del programma:
if __name__ == "__main__":
    integration_test()