tests/test_csic_trainer.py

"""Tests for the CSIC 2010 ML training pipeline."""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from intelliscan.modules.csic_trainer import (
    FEATURE_NAMES,
    CSICTrainer,
    CSICTrainingReport,
    _clean_url,
    _entropy,
    _parse_content_len,
    _special_ratio,
    extract_features,
    features_from_request,
)

# ── sample rows for unit tests ───────────────────────────────────────────────


@pytest.fixture
def normal_row():
    return pd.Series(
        {
            "URL": "http://localhost:8080/tienda1/index.jsp HTTP/1.1",
            "Method": "GET",
            "content": float("nan"),
            "lenght": float("nan"),
            "cookie": "JSESSIONID=ABCDEF1234567890",
            "classification": 0,
        }
    )


@pytest.fixture
def attack_row_sqli():
    return pd.Series(
        {
            "URL": "http://localhost:8080/tienda1/publico/autenticar.jsp?"
            "login=admin%27+OR+%271%27%3D%271&passwd=foo HTTP/1.1",
            "Method": "GET",
            "content": "' OR '1'='1",
            "lenght": "Content-Length: 22",
            "cookie": float("nan"),
            "classification": 1,
        }
    )


@pytest.fixture
def attack_row_xss():
    return pd.Series(
        {
            "URL": "http://localhost:8080/tienda1/publico/buscar.jsp?cadena=<script>alert(1)</script> HTTP/1.1",
            "Method": "GET",
            "content": "<script>alert(1)</script>",
            "lenght": "Content-Length: 25",
            "cookie": float("nan"),
            "classification": 1,
        }
    )


@pytest.fixture
def attack_row_lfi():
    return pd.Series(
        {
            "URL": "http://localhost:8080/tienda1/../../etc/passwd HTTP/1.1",
            "Method": "GET",
            "content": float("nan"),
            "lenght": float("nan"),
            "cookie": float("nan"),
            "classification": 1,
        }
    )


@pytest.fixture
def post_row():
    return pd.Series(
        {
            "URL": "http://localhost:8080/tienda1/publico/anadir.jsp HTTP/1.1",
            "Method": "POST",
            "content": "id=3&nombre=Vino+Rioja&precio=100&cantidad=55",
            "lenght": "Content-Length: 44",
            "cookie": "JSESSIONID=XYZ",
            "classification": 0,
        }
    )


# ── helper function tests ────────────────────────────────────────────────────


def test_clean_url_strips_http_version():
    raw = "http://localhost:8080/path?q=1 HTTP/1.1"
    assert _clean_url(raw) == "http://localhost:8080/path?q=1"


def test_clean_url_handles_non_string():
    assert _clean_url(None) == ""
    assert _clean_url(float("nan")) == ""


def test_parse_content_len_from_header():
    assert _parse_content_len("Content-Length: 68") == 68.0


def test_parse_content_len_plain_number():
    assert _parse_content_len("44") == 44.0


def test_parse_content_len_nan():
    assert _parse_content_len(float("nan")) == 0.0


def test_special_ratio_empty():
    assert _special_ratio("") == 0.0


def test_special_ratio_all_special():
    s = "''<>"
    ratio = _special_ratio(s)
    assert ratio == 1.0


def test_special_ratio_mixed():
    s = "abc<de"  # 1 special out of 6
    assert abs(_special_ratio(s) - 1 / 6) < 1e-6


def test_entropy_empty():
    assert _entropy("") == 0.0


def test_entropy_single_char_is_zero():
    assert _entropy("aaaa") == 0.0


def test_entropy_uniform_two_chars_is_one_bit():
    assert abs(_entropy("abab") - 1.0) < 1e-9


# ── extract_features tests ───────────────────────────────────────────────────


def test_extract_features_returns_15_values(normal_row):
    feats = extract_features(normal_row)
    assert len(feats) == len(FEATURE_NAMES) == 15


def test_extract_features_normal_row(normal_row):
    feats = extract_features(normal_row)
    (
        url_length,
        url_depth,
        url_param_count,
        url_special_ratio,
        url_encoded,
        content_len,
        content_special,
        has_sql,
        has_xss,
        has_lfi,
        method_enc,
        has_cookie,
        url_entropy,
        url_max_param_len,
        url_digit_ratio,
    ) = feats

    assert url_length > 0
    assert url_depth > 0
    assert url_param_count == 0.0  # no query params
    assert url_encoded == 0.0  # no %XX
    assert content_len == 0.0  # NaN lenght
    assert has_sql == 0.0
    assert has_xss == 0.0
    assert has_lfi == 0.0
    assert method_enc == 0.0  # GET=0
    assert has_cookie == 1.0  # cookie present
    assert url_entropy > 0.0  # any real URL has entropy
    assert url_max_param_len == 0.0  # no query params
    assert 0.0 <= url_digit_ratio <= 1.0


def test_extract_features_sqli_detected(attack_row_sqli):
    feats = extract_features(attack_row_sqli)
    has_sql = feats[7]
    url_encoded = feats[4]
    assert has_sql == 1.0
    assert url_encoded == 1.0  # %27 etc. in URL


def test_extract_features_xss_detected(attack_row_xss):
    feats = extract_features(attack_row_xss)
    has_xss = feats[8]
    assert has_xss == 1.0


def test_extract_features_lfi_detected(attack_row_lfi):
    feats = extract_features(attack_row_lfi)
    has_lfi = feats[9]
    assert has_lfi == 1.0


def test_extract_features_post_method(post_row):
    feats = extract_features(post_row)
    method_enc = feats[10]
    content_len = feats[5]
    assert method_enc == 1.0  # POST=1
    assert content_len == 44.0


def test_extract_features_all_floats(normal_row):
    feats = extract_features(normal_row)
    assert all(isinstance(f, float) for f in feats)


def test_max_param_len_reflects_longest_value(attack_row_sqli):
    feats = extract_features(attack_row_sqli)
    idx = FEATURE_NAMES.index("url_max_param_len")
    # parse_qs URL-decodes: login=admin%27+OR+%271%27%3D%271 -> "admin' OR '1'='1"
    assert feats[idx] == float(len("admin' OR '1'='1"))


def test_features_from_request_matches_row_wrapper(attack_row_xss):
    """The plain-Python entry point must agree with the Series wrapper."""
    direct = features_from_request(
        url=attack_row_xss["URL"],
        content=attack_row_xss["content"],
        method=attack_row_xss["Method"],
        content_length=attack_row_xss["lenght"],
        has_cookie=False,
    )
    assert direct == extract_features(attack_row_xss)


def test_features_from_request_defaults():
    feats = features_from_request(url="http://localhost:8080/index.jsp")
    assert len(feats) == 15
    assert all(isinstance(f, float) for f in feats)


# ── CSICTrainer unit tests (synthetic DataFrame) ─────────────────────────────


@pytest.fixture
def synthetic_df():
    """Small synthetic dataframe mimicking CSIC structure."""
    rows = []
    # Normal requests
    for i in range(30):
        rows.append(
            {
                "URL": f"http://localhost:8080/tienda1/page{i}.jsp HTTP/1.1",
                "Method": "GET",
                "content": float("nan"),
                "lenght": float("nan"),
                "cookie": "SESSION=ABC",
                "classification": 0,
            }
        )
    # Attack requests (SQLi)
    for i in range(20):
        rows.append(
            {
                "URL": f"http://localhost:8080/tienda1/item.jsp?id={i}'+OR+'1'%3D'1 HTTP/1.1",
                "Method": "GET",
                "content": "' OR '1'='1",
                "lenght": "Content-Length: 11",
                "cookie": float("nan"),
                "classification": 1,
            }
        )
    return pd.DataFrame(rows)


def test_trainer_build_features_shape(synthetic_df):
    trainer = CSICTrainer()
    X, y = trainer.build_features(synthetic_df)
    assert X.shape == (50, 15)
    assert y.shape == (50,)


def test_trainer_build_features_labels(synthetic_df):
    trainer = CSICTrainer()
    _, y = trainer.build_features(synthetic_df)
    assert set(y.tolist()) == {0, 1}


def test_trainer_train_returns_reports(synthetic_df, tmp_path):
    trainer = CSICTrainer(n_estimators=10)

    # Inject load_dataset shortcut
    X, y = trainer.build_features(synthetic_df)

    import numpy as np
    from sklearn.model_selection import train_test_split

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )
    rf_report = trainer._train_rf(X_train, X_test, y_train, y_test, X, y)
    lr_report = trainer._train_lr(X_train, X_test, y_train, y_test, X, y)

    assert isinstance(rf_report, CSICTrainingReport)
    assert isinstance(lr_report, CSICTrainingReport)
    assert 0.0 <= rf_report.accuracy <= 1.0
    assert 0.0 <= lr_report.f1_weighted <= 1.0


def test_trainer_save_load(synthetic_df, tmp_path):
    trainer = CSICTrainer(n_estimators=10)
    X, y = trainer.build_features(synthetic_df)

    from sklearn.model_selection import train_test_split

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )
    trainer._train_rf(X_train, X_test, y_train, y_test, X, y)

    save_path = tmp_path / "test_csic_model.pkl"
    trainer.save(save_path)
    assert save_path.exists()

    loaded = CSICTrainer.load(save_path)
    assert hasattr(loaded, "predict")


def test_feature_names_count():
    assert len(FEATURE_NAMES) == 15


def test_training_report_summary():
    report = CSICTrainingReport(
        model_name="Test Model",
        accuracy=0.95,
        precision=0.93,
        recall=0.97,
        f1_weighted=0.95,
        cv_mean_f1=0.94,
        cv_std_f1=0.01,
        confusion_matrix=[[100, 5], [3, 92]],
        classification_report_str="mock report",
        feature_importances={"url_length": 0.3, "has_sql_keywords": 0.25},
        n_train=160,
        n_test=40,
        roc_auc=0.99,
    )
    summary = report.summary()
    assert "95.00%" in summary
    assert "Test Model" in summary
    assert "url_length" in summary
    assert "0.9900" in summary  # ROC-AUC line